How to handle AMQP messages

[aeb-dev]

Hi,

I have a fanout exchange that publishes to both classic queue and stream. When I consume from stream I get extra values inside Message.Data.Contents appended to my original payload. I believe it is related to AMQP. The extra data looks similar to the following:

x-routing-key��somevalue�
x-exchange�somevalue

I could not copy all the extra data because browser does not let me paste it has weird characters

I have seen this issue. I got confused whether should I parse this myself or the stream client should handle it?

[Gsantomaggio]

Which RabbitMQ and Client library are you using?

[aeb-dev]

RabbitMQ 3.11.5
Stream Client 1.2.0

[Gsantomaggio]

I can't reproduce the issue:

Then I publish a message from the management UI:

The amqp values a correctly stored inside the annotation field

I could not copy all the extra data because browser does not let me paste it has weird characters

Not clear to me here from where you are trying to copy the values

[Gsantomaggio]

Can you please provide a repo where I can reproduce the issue?

[aeb-dev]

Not clear to me here from where you are trying to copy the values

I write the contents of Message.Data.Contents to a file. That file contains amqp values. I was not expecting amqp values inside Contents

[Gsantomaggio]

@aeb-dev, if you don't provide more information I am afraid that we can't help.

• How do you save the message into the files?
• is it systematic?
• Can you please provide the code to reproduce the issue, we are not able to reproduce it

[aeb-dev]

I will try to provide a reproducible example

[aeb-dev]

@Gsantomaggio I want to make sure that I understand it right. Using your example, when I deserialize Message.Data.Contents to string I should only get "aaaaaaaa" without any amqp values right?

[Gsantomaggio]

No, in my example you should see:

My message 5

Inside Message.Data.Contents

and all the routing information in the Annotation field, like:

x-routing-key = aaaaaaa
etc... 

[aeb-dev]

Sorry my bad, that is what I meant

[aeb-dev]

This is probably related to us. I will report my findings when I find the problem

[aeb-dev]

Sorry for late reply. You can use this repo to reproduce: https://github.com/aeb-dev/rabbitmq-stream-bug

There are 3 variables you should configure to get the result.
MESSAGE_COUNT: number of messages to produce & consume. You can play with this if you like...
consumer: whether to run consumer or producer. Initially, you should start with consumer set to false to create data. Then you can consume the same data over and over again.
working: When this is set true the file will have valid json(Ignore the last comma). If it is set to false it will run the bugged code. You might need to run bugged code multiple times to get the wrong output

[Gsantomaggio]

The example seems complex; what if you write directly to the file in the consumer handler? here:
https://github.com/aeb-dev/rabbitmq-stream-bug/blob/main/Program.cs#L131

[aeb-dev]

It works normally when you do it like that

[jonnepmyra]

Don't think you're supposed to store a reference to a Message outside the scope of the Messagehandler

Message.Data is a ReadOnlySequence<byte> and in general, it is not safe to store references to ReadOnlySequence instances for later use.

It's underlying memory regions could for example be re-used by the runtime at any time. I guess that's why you end up with corrupted data (json output) in the Bugged() scenario?

[lukebakken]

@aeb-dev ^^^^ this is why Bugged() does not work correctly. This code works correctly because it copies the data.

[aeb-dev]

I was actually suspecting that but although I am not sure about the alternative (Memory maybe ?) but creating another copy of byte[] does not make sense?

[lukebakken]

but creating another copy of byte[] does not make sense?

Why not? When data is read from the socket it has to go somewhere, so we rent memory from the ArrayPool<byte>.Shared object. The ReadOnlySequence<byte> is a "window" into that rented memory, and once the scope of the callback is gone the memory isn't yours anymore. You must copy the contents somehow - either into your own memory or parse it into an object.

[aeb-dev]

Can't it be made in such a way that it would create one less copy. Although I am not sure about cost but depending on the size it might be better:

Currently: socket buffer -> stream client creates array pool -> I create another copy
Suggestion: socket buffer -> stream client creates memory -> I use the same memory

[lukebakken]

Currently: socket buffer -> stream client creates array pool -> I create another copy

This isn't how it currently works. Please read the documentation.