Protobuf

A message generated by prost carries its own bytes with #[wire(prost)]: no serde, no codec feature, encoded once into the buffer that leaves.

The message derives Serialized and Deserialized next to prost::Message, and #[wire(prost)] tells both derives to call prost's encoder and decoder. The type then rides the byte lanes: a handler reads &Order, decoded by prost before the call, and a publish or a reply writes the value with prost into the publish path's own buffer. No codec is resolved for the type anywhere, so the service builds with no codec feature. In a real service prost-build generates the type: its message_attribute setting adds the derives and the #[wire(prost)] line to every message it emits. The example writes the generated struct out by hand. Another format uses the general form, #[wire(encode = <path>, decode = <path>)], which names the format's own writer and reader: Cap'n Proto, FlatBuffers and a hand-rolled frame fit it. A payload that prost cannot decode is a decode failure, settled by the subscriber's failure policy. In the manual form the impls the derives write are spelled out, and wire_bytes and from_payload call prost directly.

Example

The example builds with no codec feature:

cargo test --example protobuf --no-default-features --features testing,macros,memory

Macros

examples/protobuf.rs

#[derive(Clone, PartialEq, prost::Message, Outgoing, Serialized, Deserialized)]
#[outgoing(name = "orders")]
#[wire(prost)]
struct Order {
    #[prost(uint64, tag = "1")]
    id: u64,
    #[prost(string, tag = "2")]
    sku: String,
}

Manual

examples/manual/protobuf.rs

/// What `prost-build` emits from the schema. On the macro path a `message_attribute` line adds
/// `#[derive(Outgoing, Serialized, Deserialized)]` and `#[wire(prost)]` to it; here the impls
/// below take their place, and the struct is what the generator wrote and nothing more.
#[derive(Clone, PartialEq, prost::Message)]
struct Order {
    #[prost(uint64, tag = "1")]
    id: u64,
    #[prost(string, tag = "2")]
    sku: String,
}

// `#[derive(Serialized)]` under `#[wire(encode = ..)]`: the format's own encoder writes into the
// buffer the publish path already carries, and the value lends what it wrote - so the message is
// encoded once and nothing intermediate is allocated. The error is the format's, named here
// because a hand-written impl can see it (the derive cannot, and erases it instead).
impl Serialized for Order {
    type Error = prost::EncodeError;

    fn wire_bytes(&self, buf: &mut BytesMut) -> Result<WireBytes<'_>, Self::Error> {
        prost::Message::encode(self, &mut *buf)?;
        Ok(WireBytes::InBuffer)
    }
}

// The two spellings that route the type onto the serialized wire. They carry no bytes of their
// own: what differs between a byte bag and a generated message is `wire_bytes`, not where the
// value may be used, so the derive writes these two the same way for both.
impl MessageWire for Order {
    type Wire = SerializedWire;
}

impl ReplyShape for Order {
    type Body = Self;
    type Headers = ();
    type Wire = SerializedReply;
}

// `#[derive(Deserialized)]` under `#[wire(decode = ..)]`, plus the `Input` spelling that routes
// `&Order` onto the self-deserializing lane. `Output<'a>` is `Self` because a generated message
// owns its fields; a zero-copy reader names its own borrowing view there instead.
impl Deserialized for Order {
    type Output<'a> = Self;
    type Error = prost::DecodeError;

    fn from_payload(payload: &[u8]) -> Result<Self, Self::Error> {
        prost::Message::decode(payload)
    }
}

impl Input for Order {
    type Axis = SoloDeserialized<Self>;
}

// `#[derive(Outgoing)]` with `#[outgoing(name = "orders")]`: where the type goes, what header
// contract it declares, and what it contributes to the generated document. `with_serialized`
// records the lane, which is what the derive's probe reports for a type carrying its own bytes.
impl OutgoingDestination for Order {
    type Form = FixedName;
    const DESTINATION: &'static str = "orders";
}

impl MessageHeaders for Order {
    type Contract = NoHeaders;
}

impl<M: OutSlot> OutMessages<M> for Order {
    fn outgoing() -> Vec<OutgoingMessageMetadata> {
        vec![
            OutgoingMessageMetadata::new(Self::DESTINATION, std::any::type_name::<Self>())
                .with_serialized(true),
        ]
    }
}

The handler reads the message type, and the publish names no codec.

Macros

examples/protobuf.rs

fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        b.include(handle);
        // The outgoing half of the same declaration: the value encodes itself on the way out
        // too, so this publish names no codec either.
        b.after_startup(Publish, async move |publisher| {
            publisher
                .message(&Order {
                    id: 7,
                    sku: "widget".to_owned(),
                })
                .publish()
                .await
        });
    })
}

Manual

examples/manual/protobuf.rs

fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        b.include(subscriber("orders", Receive).build());
        // The outgoing half of the same declaration: the value encodes itself on the way out
        // too, so this publish names no codec either.
        b.after_startup(Publish, async move |publisher| {
            publisher
                .message(&Order {
                    id: 7,
                    sku: "widget".to_owned(),
                })
                .publish()
                .await
        });
    })
}