uf-photon 0.1.0

Pub/sub event pipeline facade — composable storage adapters
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57

//! Keyed topic with partition filter — **Creating topics**.
//!
//! Run: `cargo run -p uf-photon --example keyed_topic --features runtime,mem`
//!
//! Requires `PHOTON_TRANSPORT_KEY` (see `docs/configuration.md`).
#![allow(missing_docs)]
#![allow(clippy::unused_async)]

use std::time::Duration;

use futures::StreamExt;
use photon::{topic, Photon, SubscribeOpts};

// keyed_by routes events to partitions; subscribe can filter to one key.
#[topic(name = "examples.orders", keyed_by = "customer_id")]
pub struct OrderPlaced {
    pub customer_id: String,
    pub amount_cents: u64,
}

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    tracing_subscriber::fmt()
        .with_env_filter("info")
        .init();

    let photon = Photon::builder().auto_registry().build()?;

    // Ephemeral typed stream — only "alice" partitions are delivered.
    let opts = SubscribeOpts::default_ephemeral().topic_key_filter("alice");
    let mut stream = OrderPlaced::subscribe_on(&photon, opts).await?;

    // Publish bob first (filtered out), then alice (should arrive).
    OrderPlaced {
        customer_id: "bob".into(),
        amount_cents: 200,
    }
    .publish_on(&photon)
    .await?;
    OrderPlaced {
        customer_id: "alice".into(),
        amount_cents: 100,
    }
    .publish_on(&photon)
    .await?;

    let result = tokio::time::timeout(Duration::from_secs(2), stream.next()).await?;
    if let Some(Ok(envelope)) = result {
        tracing::info!(
            customer = %envelope.payload.customer_id,
            cents = envelope.payload.amount_cents,
            "filtered event"
        );
    }

    Ok(())
}