Bytes

A handler that reads the delivery's bytes as they arrived, and a reply that leaves as the bytes it holds, skip the codec.

A newtype over &'a [u8] deriving Deserialized is a view of the payload, borrowed from the broker's buffer. A handler over &Frame<'_> receives it with no copy and no decode step, and &[Frame<'_>] is the batch form. A bare &[u8] parameter does not compile: the newtype names what the bytes are. A newtype over Vec<u8> deriving Serialized publishes its bytes as they are, as a reply, an Out slot or a publish from the builder. The two lanes mix freely with codecs: a handler can decode its input with the scope codec and reply in bytes, or the other way round. A service whose messages are all on the lanes builds with no codec feature. A type that holds fields and writes its own format uses the same derives with #[wire(..)], as Protobuf does. In the manual form the impls the derives write are spelled out.

Example

The example builds with no codec feature:

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

Macros
Manual

examples/byte_lanes.rs
/// The delivery's bytes, borrowed from the broker's buffer.
#[derive(Deserialized)]
struct Frame<'a>(&'a [u8]);

/// Bytes that leave as they are.
#[derive(Outgoing, Serialized)]
struct Export(Vec<u8>);

#[subscriber("frames", reply("exports"))]
async fn reverse(frame: &Frame<'_>) -> Export {
    tracing::info!(len = frame.0.len(), "reversed");
    let mut bytes = frame.0.to_vec();
    bytes.reverse();
    Export(bytes)
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("frames", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        b.include(reverse);
    })
}

examples/manual/byte_lanes.rs
/// The delivery's bytes, borrowed from the broker's buffer.
struct Frame<'a>(&'a [u8]);

// `#[derive(Deserialized)]` by hand: the view borrows the payload, and `&Frame<'_>` takes the
// lane that reads no codec.
impl Deserialized for Frame<'_> {
    type Output<'a> = Frame<'a>;
    type Error = Infallible;

    fn from_payload(payload: &[u8]) -> Result<Frame<'_>, Self::Error> {
        Ok(Frame(payload))
    }
}

impl Input for Frame<'_> {
    type Axis = SoloDeserialized<Frame<'static>>;
}

/// Bytes that leave as they are.
struct Export(Vec<u8>);

// `#[derive(Serialized)]` by hand: the bytes, the wire of a typed publish, the shape of a reply.
impl Serialized for Export {
    type Error = Infallible;

    fn wire_bytes(&self, _buf: &mut BytesMut) -> Result<WireBytes<'_>, Infallible> {
        Ok(WireBytes::Own(&self.0))
    }
}

impl MessageWire for Export {
    type Wire = SerializedWire;
}

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

// `#[derive(Outgoing)]` with no `name`, by hand: the call site names the destination.
impl OutgoingDestination for Export {
    type Form = CallerName;
}

impl MessageHeaders for Export {
    type Contract = NoHeaders;
}

struct Reverse;

impl<'p> Handle<Frame<'p>, Export> for Reverse {
    fn handle(
        &self,
        frame: &Frame<'p>,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<Export, HandlerOutcome>> {
        tracing::info!(len = frame.0.len(), "reversed");
        let mut bytes = frame.0.to_vec();
        bytes.reverse();
        ready(Ok(Export(bytes)))
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("frames", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        b.include(subscriber("frames", Reverse).reply().to("exports").build());
    })
}

Example​

Example