Payloads

A pipe carries binary, multi-line and padded payloads intact: the stdio publisher envelopes any payload a text line would change.

A line on a pipe is UTF-8 text: the reader splits its input at newlines and trims the space around each payload. So the stdio publisher envelopes a payload with no headers in four cases: it is not UTF-8, it holds a newline, it begins or ends with whitespace, or it begins with rs1: or rs2:. Any other payload leaves as it is, so a plain JSON message stays a readable line. The next stage opens the envelope and hands its handler the bytes that were published. A stream file holds bytes, so the file publisher envelopes only a message with headers and a payload that begins with rs1: or rs2:. A handler that keeps bytes as they are takes a Deserialized input and answers with a Serialized type, so no codec touches them.

Example

The stage keeps the first kilobyte of each binary image as its thumbnail, for the next stage.

Macros
Manual

examples/stdio_payloads.rs
/// An image as it arrives, bytes the codec does not touch.
#[derive(Deserialized)]
struct Image<'a>(&'a [u8]);

/// The bytes this stage hands the next one.
#[derive(Outgoing, Serialized)]
#[outgoing(name = "thumbnails")]
struct Thumbnail(Vec<u8>);

/// Keeps the first kilobyte of each image as its thumbnail.
#[subscriber("images", reply)]
async fn thumbnail(image: &Image<'_>) -> Thumbnail {
    let kept = image.0.len().min(1024);
    tracing::info!(bytes = image.0.len(), kept, "thumbnail");
    Thumbnail(image.0[..kept].to_vec())
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("thumbnails", "0.1.0")).with_broker(StdioBroker::new(), |b| {
        b.include(thumbnail).out_retry(Publish).to("images.retry");
    })
}

examples/manual/stdio_payloads.rs
/// An image as it arrives, bytes the codec does not touch.
#[derive(Deserialized)]
struct Image<'a>(&'a [u8]);

/// The bytes this stage hands the next one.
#[derive(Outgoing, Serialized)]
#[outgoing(name = "thumbnails")]
struct Thumbnail(Vec<u8>);

/// Keeps the first kilobyte of each image as its thumbnail.
struct Shrink;

impl<'p> Handle<Image<'p>, Thumbnail> for Shrink {
    fn handle(
        &self,
        image: &Image<'p>,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<Thumbnail, HandlerOutcome>> {
        let kept = image.0.len().min(1024);
        tracing::info!(bytes = image.0.len(), kept, "thumbnail");
        ready(Ok(Thumbnail(image.0[..kept].to_vec())))
    }
}

fn app() -> impl App<State = ()> {
    RustStream::new(AppInfo::new("thumbnails", "0.1.0")).with_broker(StdioBroker::new(), |b| {
        b.include(subscriber("images", Shrink).reply().build())
            .out_retry(Publish)
            .to("images.retry");
    })
}

Example​

Example