Lines

A stdio subscription reads the lines of its stream key; a line with no key in its header reaches every subscription.

#[subscriber("jobs")] on StdioBroker reads the lines of standard input whose header names the key jobs. A line is a header in square brackets, then the payload: [timestamp | stream_key | sequence | shard_id] payload. Every field of the header is optional, so [jobs] {"id":7} names the key alone. A line under another key is passed over, so a stage takes the keys it subscribed to out of a shared stream. A line with no header names no key, and the client hands it to every subscription of the stage, whatever key the subscription reads. The client parses each header before the service sees the line. A stage stops on a line it cannot read: input that is not UTF-8, or a leading [...] that is not a header, a JSON array among them. So a producer writes a header in front of a payload that opens with [. A stdio subscription reads lines as they arrive. A delivery reports the sequence its header names, or the count of its key's lines from zero when the header names none. A reply is a line on standard output under the key its type declares, and the next stage reads it. Every registration on a pipe names where a deferred copy goes, with out_retry(Publish).to(key).

Example

The stage answers each jobs line with a results line for the next stage.

Macros
Manual

examples/stdio_lines.rs
/// Answers each `jobs` line with a `results` line for the next stage.
#[subscriber("jobs", reply)]
async fn work(job: &Job) -> Done {
    Done { id: job.id }
}

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

examples/manual/stdio_lines.rs
/// Answers each `jobs` line with a `results` line for the next stage.
struct Work;

impl Handle<Job, Done> for Work {
    fn handle(
        &self,
        job: &Job,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<Done, HandlerOutcome>> {
        ready(Ok(Done { id: job.id }))
    }
}

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

Example​

Example