capp_queue/
dispatch.rs

1use thiserror::Error;
2use tokio::sync::mpsc;
3
4use crate::Task;
5
6/// Message sent from user code into the dispatcher so enqueue operations are
7/// centralized in a single task.
8#[derive(Debug)]
9pub enum ProducerMsg<D: Clone> {
10    Enqueue(Task<D>),
11}
12
13/// Cloneable handle that lets user-facing services enqueue tasks without
14/// touching the queue directly. Internally this just forwards to the dispatcher
15/// over a bounded channel.
16#[derive(Clone, Debug)]
17pub struct ProducerHandle<D: Clone> {
18    tx: mpsc::Sender<ProducerMsg<D>>,
19}
20
21impl<D> ProducerHandle<D>
22where
23    D: Clone + Send + 'static,
24{
25    pub fn new(tx: mpsc::Sender<ProducerMsg<D>>) -> Self {
26        Self { tx }
27    }
28
29    /// Convenience to build a bounded producer channel and the receiving side.
30    pub fn channel(
31        buffer: usize,
32    ) -> (ProducerHandle<D>, mpsc::Receiver<ProducerMsg<D>>) {
33        let (tx, rx) = mpsc::channel(buffer);
34        (ProducerHandle { tx }, rx)
35    }
36
37    pub async fn enqueue(&self, task: Task<D>) -> Result<(), ProducerError> {
38        self.tx
39            .send(ProducerMsg::Enqueue(task))
40            .await
41            .map_err(|_| ProducerError::ChannelClosed)
42    }
43
44    pub fn sender(&self) -> mpsc::Sender<ProducerMsg<D>> {
45        self.tx.clone()
46    }
47}
48
49/// Result sent from a worker back to the dispatcher after a task finishes.
50#[derive(Debug)]
51pub enum WorkerResult<D: Clone> {
52    Ack { task: Task<D> },
53    Nack { task: Task<D>, error: String },
54}
55
56#[derive(Debug, Error)]
57pub enum ProducerError {
58    #[error("producer channel closed")]
59    ChannelClosed,
60}
61
62#[cfg(test)]
63mod tests {
64    use super::*;
65    use crate::Task;
66
67    #[tokio::test]
68    async fn channel_enqueue_delivers_task() {
69        let (producer, mut rx) = ProducerHandle::channel(4);
70        let task = Task::new(42u32);
71
72        producer.enqueue(task.clone()).await.unwrap();
73
74        match rx.recv().await {
75            Some(ProducerMsg::Enqueue(received)) => {
76                assert_eq!(received.payload, task.payload);
77            }
78            other => panic!("unexpected message: {other:?}"),
79        }
80    }
81
82    #[tokio::test]
83    async fn sender_clone_enqueues() {
84        let (tx, mut rx) = mpsc::channel(4);
85        let producer = ProducerHandle::new(tx);
86        let sender = producer.sender();
87        let task = Task::new("hi");
88
89        sender.send(ProducerMsg::Enqueue(task)).await.unwrap();
90
91        let msg = rx.recv().await;
92        assert!(matches!(msg, Some(ProducerMsg::Enqueue(_))));
93    }
94
95    #[tokio::test]
96    async fn enqueue_errors_when_channel_closed() {
97        let (producer, rx) = ProducerHandle::channel(1);
98        drop(rx);
99
100        let result = producer.enqueue(Task::new(7u32)).await;
101        assert!(matches!(result, Err(ProducerError::ChannelClosed)));
102    }
103}
capp_queue/dispatch.rs

capp_queue/
dispatch.rs
