paladin/channel/
mod.rs

1//! Generic channel behavior for distributed (inter-process) channels.
2//!
3//! It includes a bit more complexity than the traditional
4//! [mpsc](std::sync::mpsc::channel) channel for the following reasons:
5//! - It supports a notion of message acknowledgement.
6//! - It supports a notion of resource release.
7//! - Rather than returning a tuple of `(sender, receiver)`, it breaks each into
8//!   separate methods. This is because generally senders and receivers are
9//!   usually instantiated in separate process, as the channel is meant to
10//!   facilitate inter process communication. This avoids instantiating
11//!   unnecessary resources when only one is needed.
12
13use std::{
14    pin::Pin,
15    task::{Context, Poll},
16};
17
18use anyhow::Result;
19use async_trait::async_trait;
20use futures::Stream;
21use pin_project::{pin_project, pinned_drop};
22
23use crate::{acker::Acker, queue::Publisher, serializer::Serializable};
24
25#[derive(Clone, Copy, Debug, PartialEq, Eq)]
26pub enum ChannelType {
27    ExactlyOnce,
28    Broadcast,
29}
30
31/// Generic channel behavior for distributed (inter-process) channels.
32///
33/// It includes a bit more complexity than the traditional
34/// [mpsc](std::sync::mpsc::channel) channel for the following reasons:
35/// - It supports a notion of message acknowledgement.
36/// - It supports a notion of resource release.
37/// - Rather than returning a tuple of `(sender, receiver)`, it breaks each into
38///   separate methods.
39/// This is because generally senders and receivers are usually instantiated in
40/// separate process, as the channel is meant to facilitate inter process
41/// communication. This avoids instantiating unnecessary resources when only one
42/// is needed.
43#[async_trait]
44pub trait Channel {
45    type Sender<'a, T: Serializable + 'a>: Publisher<T>;
46    type Acker: Acker;
47    type Receiver<'a, T: Serializable + 'a>: Stream<Item = (T, Self::Acker)>;
48
49    async fn close(&self) -> Result<()>;
50
51    /// Acquire the sender side of the channel.
52    async fn sender<'a, T: Serializable + 'a>(&self) -> Result<Self::Sender<'a, T>>;
53
54    /// Acquire the receiver side of the channel.
55    async fn receiver<'a, T: Serializable + 'a>(&self) -> Result<Self::Receiver<'a, T>>;
56
57    /// Mark the channel for release.
58    fn release(&self);
59}
60
61/// Behavior for issuing new channels and retrieving existing channels.
62///
63/// Implementations should take care to ensure that the same channel is returned
64/// for a given identifier, allocating a new channel only when necessary.
65#[async_trait]
66pub trait ChannelFactory {
67    type Channel: Channel;
68
69    /// Retrieve an existing channel. An identifier is provided when a channel
70    /// is issued.
71    async fn get(&self, identifier: String, channel_type: ChannelType) -> Result<Self::Channel>;
72
73    /// Issue a new channel. An identifier is returned which can be used to
74    /// retrieve the channel later in some other process.
75    async fn issue(&self, channel_type: ChannelType) -> Result<(String, Self::Channel)>;
76}
77
78/// Guard a channel and embed a particular pipe in the lease guard.
79/// A single pipe is embedded, as the guard is meant to be held by a single end
80/// of the channel. The lease guard will release the channel when it is dropped.
81///
82/// [`LeaseGuard`] implements [`Stream`] where the pipe is a [`Stream`], and can
83/// be used as a [`Stream`] directly.
84#[pin_project(PinnedDrop)]
85pub struct LeaseGuard<C: Channel, Pipe> {
86    #[pin]
87    pipe: Pipe,
88    channel: Option<C>,
89}
90
91/// Implement [`Stream`] for [`LeaseGuard`] where the pipe is a [`Stream`].
92impl<C: Channel, Pipe: Stream> Stream for LeaseGuard<C, Pipe> {
93    type Item = Pipe::Item;
94
95    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
96        self.project().pipe.poll_next(cx)
97    }
98}
99
100impl<C: Channel, Pipe> std::ops::Deref for LeaseGuard<C, Pipe> {
101    type Target = Pipe;
102
103    fn deref(&self) -> &Self::Target {
104        &self.pipe
105    }
106}
107
108impl<C: Channel, Pipe> std::ops::DerefMut for LeaseGuard<C, Pipe> {
109    fn deref_mut(&mut self) -> &mut Self::Target {
110        &mut self.pipe
111    }
112}
113
114impl<C: Channel, Pipe> LeaseGuard<C, Pipe> {
115    pub fn new(channel: C, pipe: Pipe) -> Self {
116        Self {
117            pipe,
118            channel: Some(channel),
119        }
120    }
121}
122
123#[pinned_drop]
124impl<C: Channel, Pipe> PinnedDrop for LeaseGuard<C, Pipe> {
125    fn drop(self: Pin<&mut Self>) {
126        let this = self.project();
127        if let Some(channel) = this.channel.take() {
128            channel.release();
129        }
130    }
131}
132
133pub mod coordinated_channel;
134pub mod queue;
paladin/channel/mod.rs

paladin/channel/
mod.rs
