Node.js streams: new offset tracking tutorials

[l4mby]

Tutorial 2 for RabbitMQ streaming nodejs client

[DanielePalaia]

Hi @l4mby,

thanks for your contributions. I made a few tests, here a few comments:

• When doing this scenario (with store_offset enabled): Send, Receive, Send, Receive in the second Receive I have
  Done consuming, first offset was 100, last offset was 208
  But during the first Receive we already stored some offsets in the stream so the first_offset should be greater than 100 already

For example this is what the java stream example is printing during the second Receive:

Started consuming...
First message received.
Done consuming, first offset 106, last offset 205.

• The other clients examples when receiving a marker are closing gracefully, while here the client even if closing the consumer stays on (I need to control+C to terminate). This can create some confusion during the scenario executions, if we can do the same here too.

• Probably we can avoid printing the "Storing offset" line everytime (The other examples are not doing that). But if this will take too long because you need to modify the web-site too you can leave it.

• If you can update also the README.md file with the new scripts

[l4mby]

Hi @DanielePalaia,

• We updated the example, so the behaviour should be correct now
• Now when the receiver ends the process closes, we removed all the "Storing offset" console logs and updated the website tutorial accordingly.
• We also updated the readme to include the new commands of the example

Thank you

[DanielePalaia]

HI @l4mby

I'm having some issues with this scenario now: Send, Send and then Receive.

The receive client is returning an exception:

dpalaia@C02F541YMD6R javascript-nodejs-stream % npm run offset-tracking-receive

> rabbitmq-stream-node-tutorial@1.0.0 offset-tracking-receive
> node offset_tracking_receive.js

Connecting...
Start consuming...
First message received
Marker found
Done consuming, first offset was 0, last offset was 99
node:internal/process/promises:391
    triggerUncaughtException(err, true /* fromPromise */);
    ^

Error: write EPIPE
    at afterWriteDispatched (node:internal/stream_base_commons:161:15)

[l4mby]

Hi @DanielePalaia, i couldn't replicate the error you were having, on my machine. I pushed a fix, can you check if you get the same error?

[DanielePalaia]

Hi @l4mby,

no unforunately I still have this issue even after your last commit.

If it can help I'm running RabbitMQ 3.13.3 inside a container and this is the full exception I'm having:

Connecting...
Start consuming...
First message received
Marker found
node:internal/process/promises:391
    triggerUncaughtException(err, true /* fromPromise */);
    ^

Error: write EPIPE
    at afterWriteDispatched (node:internal/stream_base_commons:161:15)
    at writeGeneric (node:internal/stream_base_commons:152:3)
    at Socket._writeGeneric (node:net:952:11)
    at Socket._write (node:net:964:8)
    at writeOrBuffer (node:internal/streams/writable:570:12)
    at _write (node:internal/streams/writable:499:10)
    at Writable.write (node:internal/streams/writable:508:10)
    at /Users/dpalaia/projects/rabbitmq-tutorials/javascript-nodejs-stream/node_modules/rabbitmq-stream-js-client/dist/connection.js:299:25
    at new Promise (<anonymous>)
    at Connection.send (/Users/dpalaia/projects/rabbitmq-tutorials/javascript-nodejs-stream/node_modules/rabbitmq-stream-js-client/dist/connection.js:295:16) {
  errno: -32,
  code: 'EPIPE',
  syscall: 'write'
}

Node.js v22.5.1

[l4mby]

Hello @DanielePalaia i cannot replicate the problem, i am using the docker file as so:

version: "2"

services:
  rabbitmq-workshop:
    image: rabbitmq:3.13.3
    container_name: rabbitmq-workshop
    restart: unless-stopped
    hostname: "rabbitmq"
    ports:
      - "15672:15672"
      - "5672:5672"
      - "5552:5552"
    environment:
      RABBITMQ_DEFAULT_USER: "rabbit"
      RABBITMQ_DEFAULT_PASS: "rabbit"
    volumes:
      - ./conf/enabled_plugins:/etc/rabbitmq/enabled_plugins

and i am enabling the rabbitmq_stream plugin, i am also using nodejs 22.5.1 and my output is

Connecting...
Start consuming...
First message received
Marker found
Marker found
Done consuming, first offset was 0, last offset was 199

[Gsantomaggio]

@DanielePalaia Looks good to me. I can run the example without problems

[Gsantomaggio]

thank you

[Gsantomaggio]

@DanielePalaia is right.

When you publish more than one time:

npm run offset-tracking-publish;

The consumer gets:

Connecting...                                                                        
Start consuming...                                                               
First message received                                                               
Marker found                                                                         
Marker found                                                                      
Marker found

The consumer is not closed correctly.

Also if you run the consumer before the publisher:

Error while receiving message! Error: Stream was not found on any node

Sorry for the quick approve. I tested only the happy path :)!