pyra-streams 0.4.4

Redis Stream consumer infrastructure for Pyra services
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

use async_trait::async_trait;

/// Trait for handling messages from a Redis Stream.
///
/// Services implement this trait with their domain-specific processing logic.
/// The consumer framework handles XREADGROUP, XACK, XCLAIM, and dead-letter
/// automatically based on the return value:
///
/// - `Ok(())` — message is ACKed and removed from pending.
/// - `Err(_)` — message stays pending for retry via XCLAIM.
#[async_trait]
pub trait StreamHandler: Send + Sync + 'static {
    /// The error type returned by handler processing.
    type Error: std::fmt::Display + Send + Sync + 'static;

    /// Process a single stream message.
    ///
    /// `msg_id` is the Redis stream message ID (e.g. "1234567890-0").
    /// `data` is the value of the "data" field from the stream entry.
    ///
    /// Return `Ok(())` to acknowledge the message, or `Err` to leave it
    /// pending for retry.
    async fn handle_message(&self, msg_id: &str, data: &str) -> Result<(), Self::Error>;

    /// Called when a message exceeds max retries and is moved to the dead-letter stream.
    /// Default implementation is a no-op.
    async fn on_dead_letter(&self, _msg_id: &str, _data: &str) {}
}