MessagePack

MsgpackCodec reads and writes MessagePack through rmp-serde; a struct travels as an array of its field values. It is on with the msgpack feature.

A message type derives the same serde traits as for JSON. A struct is written as an array, in the order of its fields, so both ends must declare the fields in the same order. Decoding accepts a struct written as a map of field names too. A peer that expects maps, such as a service in a dynamic language, needs a codec of your own; its example is that codec. The media type is application/msgpack. The example names the codec on the scope with with_broker_codec; with msgpack as the only codec feature, DefaultCodec is MsgpackCodec. A payload in another format is a decode failure, settled by the subscriber's failure policy.

Example

The example builds with MessagePack as the only codec:

cargo test --example codec_msgpack --no-default-features --features testing,macros,memory,msgpack

Macros

examples/codec_msgpack.rs

#[subscriber("orders")]
async fn bill(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "billed");
    HandlerOutcome::ack()
}

fn app() -> impl App {
    // Every handler of this scope decodes MessagePack.
    RustStream::new(AppInfo::new("billing", "0.1.0")).with_broker_codec(
        MemoryBroker::new(),
        MsgpackCodec,
        |b| {
            b.include(bill);
        },
    )
}

Manual

examples/manual/codec_msgpack.rs

struct Bill;

impl Handle<Order> for Bill {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(order.id, "billed");
        ready(Ok(()))
    }
}

fn app() -> impl App {
    // Every registration of this scope decodes MessagePack.
    RustStream::new(AppInfo::new("billing", "0.1.0")).with_broker_codec(
        MemoryBroker::new(),
        MsgpackCodec,
        |b| {
            b.include(subscriber("orders", Bill).build());
        },
    )
}