Skip to main content

Choosing a transport

This guide compares the three Redis structures a service reads: a stream, a list and a Pub/Sub channel.

The example is a shop service that uses all three. Orders are billed off a stream, thumbnails are rendered off a list, and price changes arrive on a channel.

Streams: a durable log​

src/main.rs
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "orders")]
struct Order {
    id: u64,
    amount_cents: u64,
}

#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "invoices")]
struct Invoice {
    order_id: u64,
    amount_cents: u64,
}

#[subscriber(RedisStream::new("orders").group("billing"), reply)]
async fn bill(order: &Order) -> Invoice {
    tracing::info!(order.id, "order billed");
    Invoice {
        order_id: order.id,
        amount_cents: order.amount_cents,
    }
}

A Redis stream keeps every entry until the service trims it. RedisStream reads it through a consumer group. Each group receives every entry once, and inside a group one consumer handles it. An acknowledged entry is XACKed. An entry a consumer never acknowledged stays in the group's pending entries list, so a crash loses nothing. A stream can be read again from an earlier entry with a seek.

Orders are money, so they go on a stream. bill answers each order with an Invoice, which lands on the stream invoices.

Lists: a work queue​

src/main.rs
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "thumbnails")]
struct Thumbnail {
    image: String,
}

#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "thumbnails.ready")]
struct ThumbnailReady {
    image: String,
}

#[subscriber(RedisList::new("thumbnails").reliable(), reply)]
async fn render(job: &Thumbnail) -> ThumbnailReady {
    tracing::info!(job.image, "thumbnail rendered");
    ThumbnailReady {
        image: job.image.clone(),
    }
}

A Redis list is a queue: a producer pushes an entry, and exactly one consumer pops it. RedisList::new pops with BRPOP, so the entry leaves Redis before the handler runs, and a crash in the handler loses it. reliable moves the entry to a processing list instead and removes it only once the handler acknowledges it. A list has no groups and no replay.

Rendering a thumbnail is work for one worker, so the jobs go on a reliable list. render announces each finished thumbnail on the channel thumbnails.ready.

Pub/Sub: a broadcast​

src/main.rs
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "prices")]
struct PriceChange {
    sku: String,
    price_cents: u64,
}

#[subscriber(RedisPubSub::new("prices"))]
async fn refresh_price(change: &PriceChange) -> HandlerOutcome {
    tracing::info!(change.sku, change.price_cents, "price cache refreshed");
    HandlerOutcome::ack()
}

A Pub/Sub message reaches every subscriber connected at the moment it is published. Redis keeps nothing: a subscriber that is not connected never sees the message. RedisPubSub has no acknowledgement, so the outcome a handler returns settles nothing.

A price change matters only to the instances running now, which refresh their cache, so it goes on a channel.

Side by side​

StreamListPub/Sub
Kept while nobody readsuntil trimmeduntil poppedno
Who receives a messageone consumer in every groupone consumerevery connected subscriber
ackXACKreliable: LREM off the processing list; simple: nothingnothing
retrya copy appended to the streamreliable: back to the list; simple: lostlost
retry_aftera copy after the delaya copy after the delaya copy after the delay
A batchone read of up to n entriesassembled by the clientassembled by the client
Replaya seek of the groupnonenone

The acknowledgement pages give each outcome in full, and Retries, delays and dead letters shows how a delay and a cap work on each structure. A batch on a stream is the COUNT of one read. A list and a channel deliver one entry at a time, so the client collects a batch of the same size.

A name holds one structure. A stream and a list are keys of a type, and a write of the other type fails with WRONGTYPE. A channel is no key at all: an XADD under a channel's name writes a stream its subscribers never see.

One service, three transports​

src/main.rs
#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("shop", "0.1.0")).with_broker(
        RedisBroker::standalone("redis://localhost:6379"),
        |b| {
            b.include(bill).out_reply(stream::Publish);
            b.include(render).out_reply(pubsub::Publish::default());
            b.include(refresh_price);
        },
    )
}

A service on one transport imports that transport's prelude, such as ruststream_fred::stream::prelude. This one mixes transports, so it imports ruststream_fred::prelude. There each transport's publish policy keeps its module's name: stream::Publish writes with XADD, pubsub::Publish with PUBLISH. A reply type names where the reply goes, and the policy at the mount site names how it gets there.

The three handlers have the same shape. Moving a handler to another transport changes the descriptor in its attribute and the policy at its mount site, not its body.

The tests in the example file publish to each transport and read what the handlers sent back, all in process.