loopback

loopback sends a stage's own output back to its input, so a stdio service runs in one process; it is a test aid.

Without it, a line the service writes reaches the next process of the pipeline, and the service's own subscriptions never read it. With it, the line also reaches every subscription of the service on the same stream key. So two stages of a pipeline run in one process, and a test drives both without a shell.

In a TestApp test the pipe is held in memory and follows the same rule: the service reads its own output only under loopback. Without it, a test reads what the service wrote with published::<T>(key). A service ships without loopback: a stage that works only under it stops working in a real pipeline.

Example

The counting stage and the totalling stage run in one process. The count the first one writes under counts comes back to the second.

Macros

examples/stdio_broker_loopback.rs

/// Both stages in one process, as a test runs them.
#[ruststream::app]
fn app() -> impl App {
    let broker = StdioBroker::new().loopback();
    RustStream::new(AppInfo::new("word-count", "0.1.0")).with_broker(broker, |b| {
        b.include(count).out_retry(Publish).to("lines.retry");
        b.include(total).out_retry(Publish).to("counts.retry");
    })
}

Manual

examples/manual/stdio_broker_loopback.rs

/// Both stages in one process, as a test runs them.
fn app() -> impl App {
    let broker = StdioBroker::new().loopback();
    RustStream::new(AppInfo::new("word-count", "0.1.0")).with_broker(broker, |b| {
        b.include(subscriber("lines", Count).reply().build())
            .out_retry(Publish)
            .to("lines.retry");
        b.include(subscriber("counts", Total).build())
            .out_retry(Publish)
            .to("counts.retry");
    })
}