Add encoder and decoder for Raw Message

[JiaYingZhang]

Is your feature request related to a problem? Please describe.

Python streams client publishes messages as raw messages by default. In this library, the default encoder and decoder seems to be AMQP messages, so it cannot parse deliveries from raw messages

Describe the solution you'd like

First, I'm not quite sure if the raw message provided in the Python streams client is a standard supported by RabbitMQ streams. If it is, I hope the Rust client can add support for it.Thanks a lot.

Describe alternatives you've considered

I tried modifying the source code to write the input into the body of the Message for raw message decoding when AMQP message decoding fails. It works, but it's not an ideal approach.

The problem is that the decoder and encoder seem to be part of the core code, making it difficult for me to modify them effectively.

//  protocol/src/message/amqp/message.rs
impl AmqpDecoder for Message {
    fn decode(input: &[u8]) -> Result<(&[u8], Self), AmqpDecodeError> {
        let mut message = Message::default();
        let mut input = input;
        loop {
            if input.is_empty() {
                break;
            }
           // The problem arises here.
            let (inner, sec) = MessageSection::decode(input)?;
            input = inner;

Additional context

No response

[Gsantomaggio]

@JiaYingZhang The stream data is protocol agnostic, so you can store what you want.

We (the rabbitmq team) decided to save the data using AMQP 1.0 (message format).

The Python rstream was initially built by the community, and the maintainer decided to use raw data.

Please read the documentation about that: https://github.com/qweeze/rstream?tab=readme-ov-file#client-codecs

I'd avoid adding the raw codec for this client because we should do that for all our clients, and it is a significant effort.

So, your solution would be to read the raw data in Python and save them in Python using the AMQP 1.0 codec in another stream or the same stream but consuming in RUST by using the correct offset.

Before closing the issue, I'd like to hear @wolf4ood @michaelklishin @acogoluegnes @DanielePalaia

Thank you for raising the issue @JiaYingZhang

[acogoluegnes]

You should use the AMQP 1.0 message format for streams as long as it is possible in the client library, even if it is not the default for the library. This is what provides the best interoperability between protocols: you will be able to publish to and consume from streams with the other protocols supported by RabbitMQ (AMQP 091 and 1.0, MQTT, etc), because AMQP 1.0 message format is what the broker expects as soon as the stream protocol is not used.

If you have a stream with raw messages you should write a one-time program to consume them, convert them, and publish them to a new stream.

[JiaYingZhang]

There is many way to solve the raw data format issue. Initially, I was just confused about why the Python streams client provides a raw data format instead of an AMQP message format. @Gsantomaggio @acogoluegnes Thank you for your responses. I will close this issue.

[michaelklishin]

The default of the Python client can be changed. Feel free to file an issue with some background in that respective repo.