ruststream-fred 0.7.0

//! Redis Pub/Sub subscribers: classic broadcast, a glob pattern, and sharded (cluster-scalable)
//! delivery, plus re-publishing from a handler with the macro `reply` clause.
//!
//! Pub/Sub is fire-and-forget: no durability, no consumer groups, no ack. A descriptor selects the
//! mode. Classic (`SUBSCRIBE`) broadcasts cluster-wide and supports patterns; sharded
//! (`SSUBSCRIBE`, Redis 7+) is slot-local so it scales across a cluster but has no patterns.
//!
//! ```text
//! cargo run --example fred_pubsub --features macros,json -- run
//! ```
//!
//! Publish from another terminal:
//!
//! ```text
//! redis-cli PUBLISH events '{"kind":"login"}'
//! ```

use ruststream_fred::pubsub::prelude::*;
use serde::{Deserialize, Serialize};

#[derive(Debug, Deserialize)]
struct Event {
    kind: String,
}

// --8<-- [start:classic]
// The audit copy is a message this service sends, so its type derives `Outgoing`, and the channel
// it goes to is a property of that type: `#[outgoing(name = "audit")]` is the destination.
#[derive(Debug, Serialize, Outgoing)]
#[outgoing(name = "audit")]
struct AuditEntry {
    kind: String,
}

// Classic broadcast subscription that re-publishes each event to the `audit` channel. The bare
// `reply` clause sends the handler's return value through the publisher wired at mount, to the
// destination the reply type declares.
#[subscriber(RedisPubSub::new("events"), reply)]
async fn on_event(event: &Event) -> AuditEntry {
    println!("event: {}", event.kind);
    AuditEntry {
        kind: event.kind.clone(),
    }
}
// --8<-- [end:classic]

// --8<-- [start:pattern]
// A glob subscription (`PSUBSCRIBE`) is its own descriptor: it reads every channel the pattern
// matches, so it has no single channel of its own. Patterns are classic-only.
#[subscriber(RedisPubSubPattern::new("events.*"))]
async fn on_any_event(event: &Event) -> HandlerOutcome {
    println!("matched event: {}", event.kind);
    HandlerOutcome::ack()
}
// --8<-- [end:pattern]

// --8<-- [start:sharded]
// Sharded subscription (`SSUBSCRIBE`): on a cluster this stays slot-local and scales. It belongs on
// a cluster broker (below), and pairs with a sharded publish policy
// (`Publish::new().mode(Sharded)`).
#[subscriber(RedisPubSub::new("events").mode(PubSubMode::Sharded))]
async fn on_event_sharded(event: &Event) -> HandlerOutcome {
    println!("sharded event: {}", event.kind);
    HandlerOutcome::ack()
}
// --8<-- [end:sharded]

// One service, two brokers: classic broadcast on a standalone server and sharded delivery on a
// cluster. RustStream wires each handler onto its own broker.
// --8<-- [start:app]
#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("events", "0.1.0"))
        .with_broker(RedisBroker::standalone("redis://localhost:6379"), |b| {
            // The reply names where it goes; this names how it gets there: the Pub/Sub policy
            // (PUBLISH), not the default stream publisher (XADD). The position `out_reply` binds
            // is the value the handler returns, and the policy is pure declaration: the runtime
            // pairs it with the connected broker at startup.
            b.include(on_event).out_reply(Publish::default());
            // A pattern names no channel a `PUBLISH` can reach, so this registration says where a
            // retry copy of a delivery goes. Without it the service refuses to start.
            b.include(on_any_event)
                .out_retry(Publish::default())
                .to("events.retry");
        })
        .with_broker(RedisBroker::cluster(["redis://localhost:7000"]), |b| {
            b.include(on_event_sharded);
        })
}
// --8<-- [end:app]