eventuary_core/io/
acker.rs

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

eventuary_core/io/
acker.rs
