StdioPublish

A stdio publish writes one line to standard output under its stream key; an empty message is refused because a pipe drops empty lines.

StdioPublish, Publish in stdio::prelude, pairs into a StdioPublisher. A publish writes one line in the line format, with the message's stream key in its header. The next stage of the pipeline reads that line. This process's own subscriptions do not, because standard output never reaches the same process's standard input; loopback changes that for a test. A payload a text line would change, such as binary data or a payload with a newline, leaves in an envelope and arrives whole (Payloads). A message with no payload and no headers would be an empty line, which the line format drops. So the publisher refuses it with SeaFileError::Invalid, and nothing is written.

Example

The stage answers each word with the same word in capitals, as a line under shouts.

Macros
Manual

examples/stdio_publish.rs
/// The line this stage writes for the next one.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "shouts")]
struct Shout {
    text: String,
}

/// Answers each word with the same word in capitals.
#[subscriber("words", reply)]
async fn shout(word: &Word) -> Shout {
    Shout {
        text: word.text.to_uppercase(),
    }
}

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

examples/manual/stdio_publish.rs
/// The line this stage writes for the next one.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
#[outgoing(name = "shouts")]
struct Shout {
    text: String,
}

/// Answers each word with the same word in capitals.
struct ShoutBack;

impl Handle<Word, Shout> for ShoutBack {
    fn handle(
        &self,
        word: &Word,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<Shout, HandlerOutcome>> {
        ready(Ok(Shout {
            text: word.text.to_uppercase(),
        }))
    }
}

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

Example​

Example