pulses 0.2.0

A robust, high-performance background job processing library for Rust.
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

//! The pluggable abstractions a backend and a user implement: [`Broker`],
//! [`Subscription`], and [`Handler`].

use std::future::Future;
use std::sync::Arc;

use crate::core::types::Context;
use crate::core::types::Envelope;
use crate::core::types::HandlerError;
use crate::core::types::Outcome;

/// Broker-specific subscription/configuration object.
///
/// The runtime computes the exact set of streams it needs from the registered
/// handlers and injects them via [`Subscription::set_streams`], so callers
/// never have to keep a stream list in sync with their handlers by hand.
pub trait Subscription: Clone + Send + Sync + 'static {
    /// Replace the set of streams this subscription should consume.
    fn set_streams(&mut self, streams: Vec<Arc<str>>);
}

/// A message source/sink backend (e.g. Redis Streams).
///
/// All methods return `Send` futures so the runtime can drive brokers on a
/// multi-threaded Tokio runtime.
pub trait Broker: Clone + Send + Sync + 'static {
    /// Opaque handle identifying a delivered message for acknowledgement.
    type AckToken: Clone + Send + Sync + 'static;
    /// Backend-specific subscription type.
    type Subscription: Subscription;
    /// Backend error type.
    type Error: std::error::Error + Send + Sync + 'static;

    /// Fetch the next batch of messages. May block up to a backend-defined
    /// duration; returns an empty `Vec` when nothing is available.
    fn poll(
        &self, subscription: &Self::Subscription,
    ) -> impl Future<Output = Result<Vec<(Envelope, Self::AckToken)>, Self::Error>> + Send;

    /// Acknowledge a successfully processed message.
    fn ack(&self, token: Self::AckToken) -> impl Future<Output = Result<(), Self::Error>> + Send;

    /// Reclaim messages that were delivered to a consumer but never acked
    /// (e.g. after a crash), so they can be retried.
    fn reclaim(
        &self, subscription: &Self::Subscription,
    ) -> impl Future<Output = Result<Vec<(Envelope, Self::AckToken)>, Self::Error>> + Send;

    /// Durably record a message in the backend's dead-letter location.
    ///
    /// The caller acknowledges the source message only if this succeeds.
    fn dead_letter(&self, envelope: &Envelope, reason: &str) -> impl Future<Output = Result<(), Self::Error>> + Send;
}

/// User-defined processing logic for one or more streams.
pub trait Handler<B: Broker>: Send + Sync + 'static {
    /// Streams this handler consumes. The runtime unions these across all
    /// registered handlers to build both the routing table and the broker
    /// subscription.
    ///
    /// An empty slice is legal but means the handler is never delivered any
    /// message. When two handlers list the same stream, each receives every
    /// message on it (fan-out) and acknowledges independently.
    const STREAMS: &'static [&'static str];

    /// Maximum number of messages this handler processes concurrently. The
    /// effective ceiling is `min(MAX_IN_FLIGHT, AppConfig::max_in_flight_per_handler)`.
    const MAX_IN_FLIGHT: usize = 16;

    /// Process a single message.
    fn handle(
        &self, ctx: &Context<B>, message: &Envelope,
    ) -> impl Future<Output = Result<Outcome, HandlerError>> + Send;
}