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
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

//! Shared in-memory broker mock and helpers used by unit tests across modules.

use std::collections::VecDeque;
use std::sync::Arc;
use std::sync::Mutex;
use std::sync::atomic::AtomicUsize;
use std::sync::atomic::Ordering;
use std::time::Duration;

use bytes::Bytes;
use smallvec::SmallVec;

use crate::core::Broker;
use crate::core::Envelope;
use crate::core::Subscription;

/// Build an [`Envelope`] for tests.
pub(crate) fn make_envelope(stream: &str, id: &str) -> Envelope {
    Envelope {
        stream: Arc::from(stream),
        id: Arc::from(id),
        payload: Bytes::from_static(b"{}"),
        headers: SmallVec::new(),
        ts_unix_ms: 0,
    }
}

/// Build a mock ack token (just the entry id).
pub(crate) fn token(id: &str) -> String { id.to_owned() }

/// Minimal subscription that records the streams the runtime injects.
#[derive(Debug, Clone, Default)]
pub(crate) struct MockSubscription {
    pub streams: Vec<Arc<str>>,
}

impl Subscription for MockSubscription {
    fn set_streams(&mut self, streams: Vec<Arc<str>>) { self.streams = streams; }
}

/// Trivial error type for the mock broker.
#[derive(Debug, Clone, thiserror::Error)]
#[error("mock broker error")]
pub(crate) struct MockError;

#[derive(Default)]
struct MockState {
    ack_attempts: AtomicUsize,
    ack_success: AtomicUsize,
    ack_fail_remaining: AtomicUsize,
    dead_letters: Mutex<Vec<(String, String)>>,
    poll_batches: Mutex<VecDeque<Vec<(Envelope, String)>>>,
    reclaim_batches: Mutex<VecDeque<Vec<(Envelope, String)>>>,
}

/// In-memory [`Broker`] for tests, with scripted poll/reclaim batches and
/// configurable ack failures.
#[derive(Clone, Default)]
pub(crate) struct MockBroker {
    inner: Arc<MockState>,
}

impl MockBroker {
    /// Number of acks that have succeeded.
    pub(crate) fn ack_count(&self) -> usize { self.inner.ack_success.load(Ordering::SeqCst) }

    /// Total ack attempts (successful or not).
    pub(crate) fn ack_attempts(&self) -> usize { self.inner.ack_attempts.load(Ordering::SeqCst) }

    /// Make the next `n` acks fail.
    pub(crate) fn fail_next_acks(&self, n: usize) { self.inner.ack_fail_remaining.store(n, Ordering::SeqCst); }

    /// Recorded dead-letter writes as `(message_id, reason)`.
    pub(crate) fn dead_letters(&self) -> Vec<(String, String)> { self.inner.dead_letters.lock().unwrap().clone() }

    /// Queue a batch to be returned by the next `poll`.
    pub(crate) fn enqueue_poll(&self, batch: Vec<(Envelope, String)>) {
        self.inner.poll_batches.lock().unwrap().push_back(batch);
    }

    /// Queue a batch to be returned by the next `reclaim`.
    pub(crate) fn enqueue_reclaim(&self, batch: Vec<(Envelope, String)>) {
        self.inner.reclaim_batches.lock().unwrap().push_back(batch);
    }
}

impl Broker for MockBroker {
    type AckToken = String;
    type Subscription = MockSubscription;
    type Error = MockError;

    async fn poll(&self, _sub: &Self::Subscription) -> Result<Vec<(Envelope, String)>, MockError> {
        let batch = self.inner.poll_batches.lock().unwrap().pop_front();
        match batch {
            Some(batch) => Ok(batch),
            None => {
                tokio::time::sleep(Duration::from_millis(5)).await;
                Ok(Vec::new())
            }
        }
    }

    async fn ack(&self, _token: String) -> Result<(), MockError> {
        self.inner.ack_attempts.fetch_add(1, Ordering::SeqCst);
        // Fetch-and-decrement the remaining-failures counter without going below 0.
        let mut remaining = self.inner.ack_fail_remaining.load(Ordering::SeqCst);
        while remaining > 0 {
            match self.inner.ack_fail_remaining.compare_exchange(
                remaining,
                remaining - 1,
                Ordering::SeqCst,
                Ordering::SeqCst,
            ) {
                Ok(_) => return Err(MockError),
                Err(actual) => remaining = actual,
            }
        }
        self.inner.ack_success.fetch_add(1, Ordering::SeqCst);
        Ok(())
    }

    async fn reclaim(&self, _sub: &Self::Subscription) -> Result<Vec<(Envelope, String)>, MockError> {
        let batch = self.inner.reclaim_batches.lock().unwrap().pop_front();
        Ok(batch.unwrap_or_default())
    }

    async fn dead_letter(&self, envelope: &Envelope, reason: &str) -> Result<(), MockError> {
        self.inner.dead_letters.lock().unwrap().push((envelope.id.to_string(), reason.to_owned()));
        Ok(())
    }
}