jsonrpc_pubsub/
oneshot.rs

1//! A futures oneshot channel that can be used for rendezvous.
2
3use crate::core::futures::{self, channel::oneshot, future, Future, FutureExt, TryFutureExt};
4use std::ops::{Deref, DerefMut};
5
6/// Create a new future-base rendezvous channel.
7///
8/// The returned `Sender` and `Receiver` objects are wrapping
9/// the regular `futures::channel::oneshot` counterparts and have the same functionality.
10/// Additionaly `Sender::send_and_wait` allows you to send a message to the channel
11/// and get a future that resolves when the message is consumed.
12pub fn channel<T>() -> (Sender<T>, Receiver<T>) {
13	let (sender, receiver) = oneshot::channel();
14	let (receipt_tx, receipt_rx) = oneshot::channel();
15
16	(
17		Sender {
18			sender,
19			receipt: receipt_rx,
20		},
21		Receiver {
22			receiver,
23			receipt: Some(receipt_tx),
24		},
25	)
26}
27
28/// A sender part of the channel.
29#[derive(Debug)]
30pub struct Sender<T> {
31	sender: oneshot::Sender<T>,
32	receipt: oneshot::Receiver<()>,
33}
34
35impl<T> Sender<T> {
36	/// Consume the sender and queue up an item to send.
37	///
38	/// This method returns right away and never blocks,
39	/// there is no guarantee though that the message is received
40	/// by the other end.
41	pub fn send(self, t: T) -> Result<(), T> {
42		self.sender.send(t)
43	}
44
45	/// Consume the sender and send an item.
46	///
47	/// The returned future will resolve when the message is received
48	/// on the other end. Note that polling the future is actually not required
49	/// to send the message as that happens synchronously.
50	/// The future resolves to error in case the receiving end was dropped before
51	/// being able to process the message.
52	pub fn send_and_wait(self, t: T) -> impl Future<Output = Result<(), ()>> {
53		let Self { sender, receipt } = self;
54
55		if sender.send(t).is_err() {
56			return future::Either::Left(future::ready(Err(())));
57		}
58
59		future::Either::Right(receipt.map_err(|_| ()))
60	}
61}
62
63impl<T> Deref for Sender<T> {
64	type Target = oneshot::Sender<T>;
65
66	fn deref(&self) -> &Self::Target {
67		&self.sender
68	}
69}
70
71impl<T> DerefMut for Sender<T> {
72	fn deref_mut(&mut self) -> &mut Self::Target {
73		&mut self.sender
74	}
75}
76
77/// Receiving end of the channel.
78///
79/// When this object is `polled` and the result is `Ready`
80/// the other end (`Sender`) is also notified about the fact
81/// that the item has been consumed and the future returned
82/// by `send_and_wait` resolves.
83#[must_use = "futures do nothing unless polled"]
84#[derive(Debug)]
85pub struct Receiver<T> {
86	receiver: oneshot::Receiver<T>,
87	receipt: Option<oneshot::Sender<()>>,
88}
89
90impl<T> Future for Receiver<T> {
91	type Output = <oneshot::Receiver<T> as Future>::Output;
92
93	fn poll(mut self: std::pin::Pin<&mut Self>, cx: &mut futures::task::Context) -> futures::task::Poll<Self::Output> {
94		let r = futures::ready!(self.receiver.poll_unpin(cx))?;
95		if let Some(receipt) = self.receipt.take() {
96			let _ = receipt.send(());
97		}
98		Ok(r).into()
99	}
100}
jsonrpc_pubsub/oneshot.rs

jsonrpc_pubsub/
oneshot.rs
