Prelude

Each form has a prelude with its policy under the name Publish; ruststream_fred::prelude serves a file that mixes forms.

A service on one form globs that form's prelude: stream::prelude, list::prelude or pubsub::prelude. Each carries the core's own prelude, so #[subscriber], RustStream, HandlerOutcome and Out come with it. It adds RedisBroker, the form's descriptors and their options, the form's context types, and the capabilities the form offers. The form's policy comes under the name every broker crate uses, Publish, so a mount site reads the same on every broker. The stream prelude also brings TransactionalPublish, the seek types and the keys module of context keys.

ruststream_fred::prelude is for a file that mixes forms. It holds the same items for all three forms, with the policies under their prefixed names, RedisPublish, RedisListPublish and RedisPubSubPublish. It also brings the stream, list and pubsub modules, so a mixed file writes stream::Publish beside pubsub::Publish. It has no bare Publish, because one name would stand for three types.

A handler body names only the capability it needs, such as Out<impl Publisher>, so moving a handler between forms changes the descriptor and not the handler. The one exception is a body that sets a per-message setting: it names RedisPublishOptions in its bound, from this crate's prelude.

Example

A stream subscription replies onto a Pub/Sub channel, and a channel subscription hears the reply. The one glob brings both descriptors and the pubsub module the reply policy comes from.

Macros
Manual

examples/redis_broker_prelude.rs
use ruststream_fred::prelude::*;

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

#[subscriber(RedisStream::new("orders").group("workers"), reply)]
async fn ship(order: &Order) -> Shipped {
    Shipped { order_id: order.id }
}

#[subscriber(RedisPubSub::new("shipments"))]
async fn notify(shipped: &Shipped) -> HandlerOutcome {
    tracing::info!(shipped.order_id, "customer notified");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App<State = ()> {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        RedisBroker::standalone("redis://localhost:6379"),
        |b| {
            // Each form's policy under its module: `stream::Publish`, `list::Publish`, `pubsub::Publish`.
            b.include(ship).out_reply(pubsub::Publish::new());
            b.include(notify);
        },
    )
}

examples/manual/redis_broker_prelude.rs
use ruststream_fred::prelude::*;

struct Ship;

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

struct Notify;

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

fn app() -> impl App<State = ()> {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        RedisBroker::standalone("redis://localhost:6379"),
        |b| {
            // Each form's policy under its module: `stream::Publish`, `list::Publish`, `pubsub::Publish`.
            b.include(
                subscriber(RedisStream::new("orders").group("workers"), Ship)
                    .reply()
                    .build(),
            )
            .out_reply(pubsub::Publish::new());
            b.include(subscriber(RedisPubSub::new("shipments"), Notify).build());
        },
    )
}

Example​

Example