eventuary_core/io/
acker.rs

1mod batched;
2mod either;
3mod noop;
4mod once;
5
6use std::future::Future;
7use std::sync::Arc;
8
9use futures::future::BoxFuture;
10
11use crate::error::Result;
12
13pub use ::either::Either;
14pub use batched::{AckBuffer, AckBufferConfig, AckCmd, BatchFlusher, BatchedAcker};
15pub use noop::NoopAcker;
16pub use once::OnceAcker;
17
18/// Acknowledges or rejects a delivered event message.
19///
20/// Backends define exact durability semantics. In general, [`Acker::ack`]
21/// marks the message as successfully processed. [`Acker::nack`] requests
22/// redelivery when the backend supports it, or leaves the checkpoint
23/// unchanged otherwise. See each backend crate for backend-specific
24/// behavior.
25pub trait Acker: Send + Sync {
26    fn ack(&self) -> impl Future<Output = Result<()>> + Send;
27    fn nack(&self) -> impl Future<Output = Result<()>> + Send;
28}
29
30impl<T: Acker + ?Sized> Acker for Arc<T> {
31    fn ack(&self) -> impl Future<Output = Result<()>> + Send {
32        (**self).ack()
33    }
34    fn nack(&self) -> impl Future<Output = Result<()>> + Send {
35        (**self).nack()
36    }
37}
38
39impl<T: Acker + ?Sized> Acker for Box<T> {
40    fn ack(&self) -> impl Future<Output = Result<()>> + Send {
41        (**self).ack()
42    }
43    fn nack(&self) -> impl Future<Output = Result<()>> + Send {
44        (**self).nack()
45    }
46}
47
48pub trait DynAcker: Send + Sync {
49    fn ack_dyn<'a>(&'a self) -> BoxFuture<'a, Result<()>>;
50    fn nack_dyn<'a>(&'a self) -> BoxFuture<'a, Result<()>>;
51}
52
53impl<T: Acker + ?Sized> DynAcker for T {
54    fn ack_dyn<'a>(&'a self) -> BoxFuture<'a, Result<()>> {
55        Box::pin(<Self as Acker>::ack(self))
56    }
57    fn nack_dyn<'a>(&'a self) -> BoxFuture<'a, Result<()>> {
58        Box::pin(<Self as Acker>::nack(self))
59    }
60}
61
62pub type BoxAcker = Box<dyn DynAcker>;
63pub type ArcAcker = Arc<dyn DynAcker>;
64
65impl Acker for dyn DynAcker + '_ {
66    fn ack(&self) -> impl Future<Output = Result<()>> + Send {
67        DynAcker::ack_dyn(self)
68    }
69    fn nack(&self) -> impl Future<Output = Result<()>> + Send {
70        DynAcker::nack_dyn(self)
71    }
72}
73
74pub trait AckerExt: Acker + Sized + 'static {
75    fn into_boxed(self) -> BoxAcker {
76        Box::new(self)
77    }
78
79    fn into_arced(self) -> ArcAcker {
80        Arc::new(self)
81    }
82}
83
84impl<T: Acker + 'static> AckerExt for T {}
85
86#[cfg(test)]
87mod tests {
88    use super::*;
89
90    use std::sync::atomic::{AtomicUsize, Ordering};
91
92    struct CountingAcker {
93        acks: Arc<AtomicUsize>,
94        nacks: Arc<AtomicUsize>,
95    }
96
97    impl Acker for CountingAcker {
98        async fn ack(&self) -> Result<()> {
99            self.acks.fetch_add(1, Ordering::SeqCst);
100            Ok(())
101        }
102        async fn nack(&self) -> Result<()> {
103            self.nacks.fetch_add(1, Ordering::SeqCst);
104            Ok(())
105        }
106    }
107
108    fn counters() -> (Arc<AtomicUsize>, Arc<AtomicUsize>) {
109        (Arc::new(AtomicUsize::new(0)), Arc::new(AtomicUsize::new(0)))
110    }
111
112    #[tokio::test]
113    async fn into_boxed_yields_dyn_acker() {
114        let (acks, nacks) = counters();
115        let acker: BoxAcker = CountingAcker {
116            acks: Arc::clone(&acks),
117            nacks: Arc::clone(&nacks),
118        }
119        .into_boxed();
120        acker.ack().await.unwrap();
121        acker.nack().await.unwrap();
122        assert_eq!(acks.load(Ordering::SeqCst), 1);
123        assert_eq!(nacks.load(Ordering::SeqCst), 1);
124    }
125
126    #[tokio::test]
127    async fn into_arced_yields_shared_acker() {
128        let (acks, _) = counters();
129        let acker: ArcAcker = CountingAcker {
130            acks: Arc::clone(&acks),
131            nacks: Arc::new(AtomicUsize::new(0)),
132        }
133        .into_arced();
134        let clone = Arc::clone(&acker);
135        acker.ack().await.unwrap();
136        clone.ack().await.unwrap();
137        assert_eq!(acks.load(Ordering::SeqCst), 2);
138    }
139
140    #[tokio::test]
141    async fn box_blanket_passes_as_generic_acker() {
142        async fn take<A: Acker>(a: A) {
143            a.ack().await.unwrap();
144        }
145        let (acks, nacks) = counters();
146        let boxed: BoxAcker = CountingAcker {
147            acks: Arc::clone(&acks),
148            nacks,
149        }
150        .into_boxed();
151        take(boxed).await;
152        assert_eq!(acks.load(Ordering::SeqCst), 1);
153    }
154
155    #[tokio::test]
156    async fn arc_blanket_passes_as_generic_acker() {
157        async fn take<A: Acker>(a: A) {
158            a.ack().await.unwrap();
159        }
160        let (acks, nacks) = counters();
161        let arced: ArcAcker = CountingAcker {
162            acks: Arc::clone(&acks),
163            nacks,
164        }
165        .into_arced();
166        take(arced).await;
167        assert_eq!(acks.load(Ordering::SeqCst), 1);
168    }
169}
eventuary_core/io/acker.rs

eventuary_core/io/
acker.rs
