Interoperating with other clients

This guide shows what a RustStream service reads and writes in Redis, so that producers and consumers in other languages share its streams, lists and channels.

The example is an order service. A shop written in another language appends orders to a stream, and a mailer written in another language pops emails off a list.

Reading a stream entry

Macros
Manual

src/main.rs
/// An order as the shop's producer appends it: `XADD orders * _payload '{"id":1,...}'`.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "orders")]
struct Order {
    id: u64,
    email: String,
}

/// An email for the mailer, a worker written in another language.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "emails")]
struct Email {
    to: String,
    subject: String,
}

#[subscriber(RedisStream::new("orders").group("mail"), reply)]
async fn take_order(order: &Order) -> Email {
    tracing::info!(order.id, "order taken");
    Email {
        to: order.email.clone(),
        subject: format!("Order {} received", order.id),
    }
}

src/main.rs
/// An order as the shop's producer appends it: `XADD orders * _payload '{"id":1,...}'`.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
#[outgoing(name = "orders")]
struct Order {
    id: u64,
    email: String,
}

/// An email for the mailer, a worker written in another language.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
#[outgoing(name = "emails")]
struct Email {
    to: String,
    subject: String,
}

struct TakeOrder;

impl Handle<Order, Email> for TakeOrder {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<Email, HandlerOutcome>> {
        tracing::info!(order.id, "order taken");
        ready(Ok(Email {
            to: order.email.clone(),
            subject: format!("Order {} received", order.id),
        }))
    }
}

A Redis stream entry is a set of fields. A RustStream service keeps the message body in the field _payload and each header in a field named h: followed by the header's name. A producer in any language writes an order the service reads like this:

redis-cli XADD orders '*' _payload '{"id":1,"email":"ann@example.com"}' h:trace-id 4bf92f35

The body is what the codec decodes: JSON with the default codec, so the producer writes the order as JSON. A header field is optional; an entry with _payload alone arrives with no headers. Fields with other names are ignored, so a producer can add its own. An entry without _payload has an empty body, which the codec fails to decode; the decode failure policy (FailurePolicy) settles it, and its default drops the entry.

A consumer in another language reads entries the service writes the same way: _payload holds the body, and the h: fields hold the headers.

Writing a list value

Macros
Manual

src/main.rs
#[ruststream::app]
fn app() -> impl App<State = ()> {
    service(RedisBroker::standalone("redis://localhost:6379"))
}

fn service(broker: RedisBroker) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("shop", "0.1.0")).with_broker(broker, |b| {
        // The envelope in JSON: a value the mailer reads with any JSON parser.
        b.include(take_order)
            .out_reply(list::Publish::default().codec(JsonCodec));
    })
}

src/main.rs
fn app() -> impl App<State = ()> {
    service(RedisBroker::standalone("redis://localhost:6379"))
}

fn service(broker: RedisBroker) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("shop", "0.1.0")).with_broker(broker, |b| {
        let orders = RedisStream::new("orders").group("mail");
        // The envelope in JSON: a value the mailer reads with any JSON parser.
        b.include(subscriber(orders, TakeOrder).reply().build())
            .out_reply(list::Publish::default().codec(JsonCodec));
    })
}

fn main() -> ExitCode {
    run_main(app)
}

A list entry and a Pub/Sub message are one value, so the headers travel in a frame around the body. By default the frame is binary:

Bytes	Meaning
4	the number of headers, a big-endian `u32`
4, then the name	for each header: the name's length and the name
4, then the value	the value's length and the value
the rest	the body

A codec on the publish policy writes an envelope in that codec instead. With JsonCodec the mailer pops a JSON object:

redis-cli RPOP emails
# {"headers":{},"payload":"{\"to\":\"ann@example.com\",\"subject\":\"Order 1 received\"}"}

payload is the body, here the email as JSON text, so the mailer parses the envelope and then its payload. A body or a header that is valid UTF-8 is written as a string, and any other bytes as an array of numbers. The same envelope shows readable values in a tool like RedisInsight.

Reading a list value or a channel message

A subscription reads the frame its publisher writes: RedisList::codec and RedisPubSub::codec set the envelope's codec, and without one the subscription reads the binary frame. A subscription with a codec also reads binary frames, which a service leaves behind when it switches.

A value that is neither, such as one another client pushes or publishes as plain JSON, arrives as the body with no headers:

redis-cli LPUSH jobs '{"id":7}'
redis-cli PUBLISH prices '{"sku":"tea","cents":450}'

So a producer in another language writes plain values unless it needs headers, and then it writes the JSON envelope.

Testing

The test in the example files runs in process: the order arrives as the shop writes it, and the value on emails is the JSON envelope, byte for byte. A live test reads both from a server with a second fred client, as another program would: the newest entry of orders holds the order under _payload, and the newest value of emails is the envelope.

Reading a stream entry​

Writing a list value​

Reading a list value or a channel message​

Testing​

Reading a stream entry

Writing a list value

Reading a list value or a channel message

Testing