Framing

A list entry or a channel message carries its headers in a binary frame, or in a readable envelope when both ends set one codec.

A stream entry has fields, so it keeps the payload in the field _payload and each header in a field named h:<header>. A list element and a channel message are one value, so the headers travel in a frame around the payload. The default frame is a compact binary one, which tools show as bytes. A codec set on both ends frames the value as a {headers, payload} envelope serialized with that codec. With JSON, redis-cli and RedisInsight show the envelope as text. The reading end sets it with RedisList::codec, RedisPubSub::codec or RedisPubSubPattern::codec. The writing end sets it with RedisListPublish::codec or RedisPubSubPublish::codec. The broker's default publisher writes a list or a channel this service reads in that subscription's framing. Both framings carry any bytes. The envelope writes a field whose bytes are valid UTF-8 as text and any other bytes as themselves. A subscription with a codec reads either framing. A subscription without one reads an envelope as its payload, and the decode of that payload fails. A value an outside client wrote without a frame arrives as the payload with no headers. The codecs cbor and msgpack of the core frame the envelope the same way.

Example

Orders become jobs on a list, with the tenant in a header; both ends name JSON.

Macros
Manual

examples/redis_subscriptions_framing.rs
/// The list jobs leave through.
#[derive(OutSlot)]
#[publishes(Job)]
struct Jobs;

/// Queues a job for each order, with the tenant in a header.
#[subscriber(RedisStream::new("orders").group("dispatch"))]
async fn dispatch(order: &Order, Out(jobs): Out<impl Publisher, Jobs>) -> HandlerOutcome {
    let mut headers = HeaderMap::new();
    headers.insert("tenant", order.tenant.clone());
    let job = Job { order: order.id };
    if jobs
        .message(&job)
        .with_headers(headers)
        .publish()
        .await
        .is_err()
    {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

/// Runs each job for the tenant its header names.
#[subscriber(RedisList::new("jobs").codec(JsonCodec))]
async fn run(job: &Job, ctx: &mut Context<'_>) -> HandlerOutcome {
    let Some(tenant) = ctx.headers().get_str("tenant") else {
        tracing::warn!(job.order, "a job without a tenant is dropped");
        return HandlerOutcome::drop();
    };
    tracing::info!(job.order, tenant, "job run");
    HandlerOutcome::ack()
}

fn app() -> impl App {
    let broker = RedisBroker::standalone("redis://localhost:6379");
    RustStream::new(AppInfo::new("jobs", "0.1.0")).with_broker(broker, |b| {
        // Both ends name JSON, so `redis-cli` and RedisInsight show each entry as text.
        b.include(dispatch)
            .out(Jobs, RedisListPublish::new().codec(JsonCodec))
            .build();
        b.include(run);
    })
}

examples/manual/redis_subscriptions_framing.rs
/// The list jobs leave through.
struct Jobs;

impl OutSlot for Jobs {
    const NAME: &'static str = "Jobs";
    // `Job` fixes its destination, so no transform on this slot may name one.
    type Destination = Reads;
}

impl PublishedThrough<Jobs> for Job {}

/// Queues a job for each order, with the tenant in a header.
struct Dispatch;

impl<Entry> Handle<Order, (), Outs<(Entry,)>> for Dispatch
where
    Entry: OutEntry<Jobs, Wire: Publisher>,
{
    async fn handle(
        &self,
        order: &Order,
        outs: &Outs<(Entry,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        let mut headers = HeaderMap::new();
        headers.insert("tenant", order.tenant.clone());
        let job = Job { order: order.id };
        outs.get(Jobs)
            .message(&job)
            .with_headers(headers)
            .publish()
            .await
            .map_err(|_| HandlerOutcome::retry())
    }
}

/// Runs each job for the tenant its header names.
struct Run;

impl Handle<Job> for Run {
    fn handle(
        &self,
        job: &Job,
        _outs: &(),
        ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        let Some(tenant) = ctx.headers().get_str("tenant") else {
            tracing::warn!(job.order, "a job without a tenant is dropped");
            return ready(Err(HandlerOutcome::drop()));
        };
        tracing::info!(job.order, tenant, "job run");
        ready(Ok(()))
    }
}

fn app() -> impl App {
    let broker = RedisBroker::standalone("redis://localhost:6379");
    RustStream::new(AppInfo::new("jobs", "0.1.0")).with_broker(broker, |b| {
        // Both ends name JSON, so `redis-cli` and RedisInsight show each entry as text.
        let orders = RedisStream::new("orders").group("dispatch");
        b.include(subscriber(orders, Dispatch).build())
            .out(Jobs, RedisListPublish::new().codec(JsonCodec))
            .build();
        let jobs = RedisList::new("jobs").codec(JsonCodec);
        b.include(subscriber(jobs, Run).build());
    })
}

Example​

Example