Handle the body is undefined from AMQP 1.0 to AMQP 091

[Gsantomaggio]

Fixes part of #6837

If the body is undefined, the payload returns an empty binary.
Signed-off-by: Gabriele Santomaggio [email protected]

This PR fixes a bug in case the native stream client sends a message without a body and an AMQP client reads the message.
example:

 Producer producer = environment.producerBuilder()
                .stream(stream)
                .name("reference")//
                .build();
        for (int i = 0; i < 20; i++) {
            producer.send(
                    producer.messageBuilder().publishingId(i).properties().messageId(i).messageBuilder().build(),

then try to read in AMQP:

channel.basic_qos(prefetch_count=100)  # mandatory
channel.basic_consume(
    queue=q_name,
    on_message_callback=callback,
    arguments={
        'x-stream-offset': 'first'
        #   'x-stream-offset': 15
        # with first start consuming always from the beginning
    },
    auto_ack=False)
channel.start_consuming()

The server raises:

2023-01-03 16:58:44.428652+00:00 [warning] <0.2303.0> Non-AMQP exit reason '{function_clause,
2023-01-03 16:58:44.428652+00:00 [warning] <0.2303.0>                        [{rabbit_msg_record,to_amqp091,
2023-01-03 16:58:44.428652+00:00 [warning] <0.2303.0>                          [{rabbit_msg_record,
2023-01-03 16:58:44.428652+00:00 [warning] <0.2303.0>                            {cfg},
2023-01-03 16:58:44.428652+00:00 [warning] <0.2303.0>                            {msg,undefined,

This because:

rabbitmq-server/deps/rabbit/src/rabbit_msg_record.erl

Lines 221 to 224 in aa02e32

	to_amqp091(#?MODULE{msg = #msg{properties = P,
	application_properties = APR,
	message_annotations = MAR,
	data = #'v1_0.data'{content = Payload}}}) ->

Does not consider the empty data

Proposed Changes

With this fix, the function returns an empty binary <<>> When the body message is not defined.

Types of Changes

What types of changes does your code introduce to this project?
Put an x in the boxes that apply

• Bug fix (non-breaking change which fixes issue #NNNN)
• New feature (non-breaking change which adds functionality)
• Breaking change (fix or feature that would cause an observable behavior change in existing systems)
• Documentation improvements (corrections, new content, etc)
• Cosmetic change (whitespace, formatting, etc)
• Build system and/or CI

Checklist

Put an x in the boxes that apply.
You can also fill these out after creating the PR.
If you're unsure about any of them, don't hesitate to ask on the mailing list.
We're here to help!
This is simply a reminder of what we are going to look for before merging your code.

• I have read the CONTRIBUTING.md document
• I have signed the CA (see https://cla.pivotal.io/sign/rabbitmq)
• I have added tests that prove my fix is effective or that my feature works
• All tests pass locally with my changes
• If relevant, I have added necessary documentation to https://github.com/rabbitmq/rabbitmq-website
• If relevant, I have added this change to the first version(s) in release-notes that I expect to introduce it

[Gsantomaggio]

thanks to @gpad for reporting the bug

[ansd]

Note that each AMQP message must contain a body:

The body consists of one of the following three choices: one or more data sections, one or more amqp-sequence sections, or a single amqp-value section.

[Gsantomaggio]

@ansd Yes.
We add this fix to make the stream (amqp 10)--> amqp (amqp 091) compatible.

We could add this constraint on the client side or handle it on the server side but with better error handling.