Error: Not supported format code 83 when messsage is published using amqplib(amqp 0.9.1)

[kashif-ghafoor]

Background
I want to publish message using amqp 0.9.1 (used amqplib) to stream queue. And then consume messages using rabbitmq-stream-js-client. The reason for doing this is to use stream filtering feature for consuming only subset of messages on consumer side.

Problem
When I try to consume messages that are published using amqplib, I get following error
Error

Error: Not supported format code 83

Error Stack

/rabbitmq-stream-js-client/dist/response_decoder.js:215
                throw new Error(`Not supported format code ${formatCode}`);
                ^

Error: Not supported format code 83
    at decodeMessage (rabbitmq-stream-js-client/dist/response_decoder.js:215:23)
    at decodeDeliverResponse (/rabbitmq-stream-js-client/dist/response_decoder.js:173:27)
    at decodeResponse (/rabbitmq-stream-js-client/dist/response_decoder.js:67:64)
    at decode (/rabbitmq-stream-js-client/dist/response_decoder.js:61:41)
    at ResponseDecoder.add (/rabbitmq-stream-js-client/dist/response_decoder.js:530:45)
    at Connection.received (/rabbitmq-stream-js-client/dist/connection.js:283:22)
    at Socket.<anonymous> (/rabbitmq-stream-js-client/dist/connection.js:125:22)
    at Socket.emit (node:events:524:28)
    at addChunk (node:internal/streams/readable:561:12)
    at readableAddChunkPushByteMode (node:internal/streams/readable:512:3)

Code Snippets

Consumer

  console.log("connecting...");
  const client = await connect({
    hostname: "localhost",
    port: 5552,
    username: "guest",
    password: "guest",
    vhost: "/",
  });

  const consumerOptions = {
    stream: "debug_work_queue.stream.dlq",
    offset: Offset.first(),
  };

  const consumer = await client.declareConsumer(
    consumerOptions,
    (message: Message) => {
      console.log("consumed message");
      //   console.log(message);
      //   console.log(message.content.toString()); // it's a Buffer
      //   console.log(message.applicationProperties);
      //   console.log(JSON.stringify(message, null, 2));
    }
  );

[dc1992]

Hi @kashif-ghafoor , thanks for reporting the issue!

I reproduced it but the error message I receive is slightly different. I get Error: ReadCompositeHeader Invalid type 0 instead of 83 as you reported; Can you share the snippet that reproduces your error?

I've used this:

  const conn = await amqplib.connect(
    `amqp://${username}:${password}@${host}:5672${vhost}`
  );

  const channel = await conn.createChannel();

  await channel.sendToQueue(streamName, Buffer.from('{ "hello": "world" }'));

And also, which version of rabbitmq-stream-js-client are you using?
Thanks

[kashif-ghafoor]

Thanks @dc1992 for quick response.

Here Publisher code to reproduce the error

  const Queue = "test_queue";
  const connection = await amqp.connect("amqp://localhost:5672");
  const channel = await connection.createChannel();
  await channel.assertQueue(Queue, {
    durable: true,
    arguments: {
      "x-queue-type": "stream",
    },
  });
  channel.sendToQueue(Queue, Buffer.from("Hello, world!"), {
    deliveryMode: 2,
    priority: 5,
  });
  await channel.close();
  await connection.close();

I figured out that if we try to consume message published with priority then we get to see the error code 83.

[ThomasShih]

Just a +1 that I'm getting an invalid format code as well.

order_app_2  | > order_app_2@1.0.0 main
order_app_2  | > node main.js
order_app_2  | 
order_app_2  | Connecting...
order_app_2  | Making sure the stream exists...
order_app_2  | Declaring the consumer with offset...
order_app_2  | /node_modules/rabbitmq-stream-js-client/dist/response_decoder.js:364
order_app_2  |             throw new Error(`ReadCompositeHeader Invalid type ${formatCode}`);
order_app_2  |             ^
order_app_2  | 
order_app_2  | Error: ReadCompositeHeader Invalid type 0

This is recreated simply from using the rabbitmq example code snippit and just using the management client to publish an empty message.

...
I assume the management ui is using amqp 0.9.1? If so, this is a recreation of #232 (comment)

This issue exists using both the example code's lib version (0.4.2) and the latest from npm.

[thaoula]

Hi Team,

I raised something similar to this in the early days of the library and was revisiting the library again.
#181

In version 0.4.2 of the library, I get error code 83 and in Version 0.5.1 I get error code 0.

In my case I am using Rascal to publish to a rabbitmq stream using amqp 0.9.1.

[dc1992]

We fixed this, that's going to be released in the next version. Thanks for reporting!

[kashif-ghafoor]

Awesome Thanks @dc1992 for fix and responses.

[Gsantomaggio]

Great job guys !

[ThomasShih]

Thank you!