RedisDefaultPublish

A reply with no named policy goes through RedisDefaultPublish, which writes each name the way this service reads it, and XADDs a name it does not read.

The runtime wires it for every registration that replies and names no policy: b.include(handler) with no .out_reply(..). Its live form, RedisDefaultPublisher, looks up how the service reads the destination. A name the service reads as a list gets an LPUSH, a channel a PUBLISH, a stream an XADD, each in the framing that subscription decodes. A name the service does not read gets an XADD. The choice is made per name, so one handler's replies may go to a list and another's to a stream. A mount site that names a policy, such as RedisPublish, overrides the default.

Example

The planner replies to jobs without naming a policy; the service reads jobs as a reliable list, so each job is pushed onto it.

Macros
Manual

examples/redis_publishing_default.rs
/// Plans a job per order; the reply names no policy.
#[subscriber(RedisStream::new("orders").group("planners"), reply)]
async fn plan(order: &Order) -> Job {
    Job { order_id: order.id }
}

/// Pops the jobs off the `jobs` list.
#[subscriber(RedisList::new("jobs").reliable())]
async fn work(job: &Job) -> HandlerOutcome {
    tracing::info!(job.order_id, "job done");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("jobs", "0.1.0")).with_broker(
        RedisBroker::standalone("redis://localhost:6379"),
        |b| {
            // `jobs` is a list this service reads, so the default publisher pushes onto it.
            b.include(plan);
            b.include(work);
        },
    )
}

examples/manual/redis_publishing_default.rs
/// Plans a job per order; the reply names no policy.
struct Plan;

impl Handle<Order, Job> for Plan {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<Job, HandlerOutcome>> {
        ready(Ok(Job { order_id: order.id }))
    }
}

/// Pops the jobs off the `jobs` list.
struct Work;

impl Handle<Job> for Work {
    fn handle(
        &self,
        job: &Job,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(job.order_id, "job done");
        ready(Ok(()))
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("jobs", "0.1.0")).with_broker(
        RedisBroker::standalone("redis://localhost:6379"),
        |b| {
            let plan = subscriber(RedisStream::new("orders").group("planners"), Plan);
            // `jobs` is a list this service reads, so the default publisher pushes onto it.
            b.include(plan.reply().build());
            b.include(subscriber(RedisList::new("jobs").reliable(), Work).build());
        },
    )
}

Example​

Example