shove 0.11.5

Async tasks via pubsub on steroids. Comes with built-in support for complex queue configurations, audit logs, autoscaling consumer groups and more.
Documentation 
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
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139

//! FIFO (sequenced) Redis Streams publish/consume example.
//!
//! Demonstrates per-account ordering using `SequencedTopic` + `register_fifo`.
//! Messages for the same account key are always delivered to the same shard and
//! processed in strict publish order, regardless of how many consumer tasks run.
//!
//! Run with:
//!   docker run --rm -p 6379:6379 redis:7-alpine
//!   cargo run --example redis_sequenced --features redis-streams

use std::sync::OnceLock;
use std::time::Duration;

use serde::{Deserialize, Serialize};

use shove::consumer_group::ConsumerGroupConfig;
use shove::redis::{RedisConfig, RedisConsumerGroupConfig, RedisMode};
use shove::{
    Broker, JsonCodec, MessageHandler, MessageMetadata, Outcome, Redis, SequenceFailure,
    SequencedTopic, Topic, TopologyBuilder,
};

// ---------------------------------------------------------------------------
// Message type
// ---------------------------------------------------------------------------

#[derive(Debug, Clone, Serialize, Deserialize)]
struct LedgerEntry {
    account_id: String,
    amount_cents: i64,
    seq: u64,
}

// ---------------------------------------------------------------------------
// Topic definition
// ---------------------------------------------------------------------------

struct Ledger;

impl Topic for Ledger {
    type Message = LedgerEntry;
    type Codec = JsonCodec;

    fn topology() -> &'static shove::QueueTopology {
        static T: OnceLock<shove::QueueTopology> = OnceLock::new();
        T.get_or_init(|| {
            TopologyBuilder::new("ledger")
                .sequenced(SequenceFailure::Skip)
                .routing_shards(4)
                .build()
        })
    }

    const SEQUENCE_KEY_FN: Option<fn(&Self::Message) -> String> = Some(Ledger::sequence_key);
}

impl SequencedTopic for Ledger {
    fn sequence_key(msg: &LedgerEntry) -> String {
        msg.account_id.clone()
    }
}

// ---------------------------------------------------------------------------
// Handler
// ---------------------------------------------------------------------------

struct Handler;

impl MessageHandler<Ledger> for Handler {
    type Context = ();

    async fn handle(&self, msg: LedgerEntry, _: MessageMetadata, _: &()) -> Outcome {
        println!(
            "account={} seq={:03} amount={:+}",
            msg.account_id, msg.seq, msg.amount_cents
        );
        Outcome::Ack
    }
}

// ---------------------------------------------------------------------------
// Main
// ---------------------------------------------------------------------------

#[tokio::main]
async fn main() -> Result<(), shove::ShoveError> {
    tracing_subscriber::fmt::init();

    let url = std::env::var("REDIS_URL").unwrap_or_else(|_| "redis://127.0.0.1:6379/".into());

    let broker = Broker::<Redis>::new(RedisConfig::new(RedisMode::Standalone { url })).await?;

    broker.topology().declare::<Ledger>().await?;

    // Publish interleaved entries for three accounts.
    let publisher = broker.publisher().await?;
    let entries = [
        ("acct-001", 1000, 0),
        ("acct-002", 2000, 0),
        ("acct-001", -500, 1),
        ("acct-003", 750, 0),
        ("acct-002", -200, 1),
        ("acct-001", 300, 2),
        ("acct-003", -100, 1),
    ];

    for (account_id, amount_cents, seq) in entries {
        publisher
            .publish::<Ledger>(&LedgerEntry {
                account_id: account_id.into(),
                amount_cents,
                seq,
            })
            .await?;
        println!("published account={account_id} seq={seq:03}");
    }

    // Consume with FIFO (one task per shard — entries for the same account
    // are always processed in the order they were published).
    let mut group = broker.consumer_group();
    group
        .register_fifo::<Ledger, _>(
            ConsumerGroupConfig::new(RedisConsumerGroupConfig::default()),
            || Handler,
        )
        .await?;

    println!("consuming — press Ctrl-C to stop");
    let outcome = group
        .run_until_timeout(
            async {
                tokio::signal::ctrl_c().await.ok();
            },
            Duration::from_secs(5),
        )
        .await;

    std::process::exit(outcome.exit_code());
}