libp2p_webrtc/tokio/
req_res_chan.rs

1// Copyright 2022 Parity Technologies (UK) Ltd.
2//
3// Permission is hereby granted, free of charge, to any person obtaining a
4// copy of this software and associated documentation files (the "Software"),
5// to deal in the Software without restriction, including without limitation
6// the rights to use, copy, modify, merge, publish, distribute, sublicense,
7// and/or sell copies of the Software, and to permit persons to whom the
8// Software is furnished to do so, subject to the following conditions:
9//
10// The above copyright notice and this permission notice shall be included in
11// all copies or substantial portions of the Software.
12//
13// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
14// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
15// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
16// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
17// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
18// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
19// DEALINGS IN THE SOFTWARE.
20
21use std::{
22    io,
23    task::{Context, Poll},
24};
25
26use futures::{
27    channel::{mpsc, oneshot},
28    SinkExt, StreamExt,
29};
30
31pub(crate) fn new<Req, Res>(capacity: usize) -> (Sender<Req, Res>, Receiver<Req, Res>) {
32    let (sender, receiver) = mpsc::channel(capacity);
33
34    (
35        Sender {
36            inner: futures::lock::Mutex::new(sender),
37        },
38        Receiver { inner: receiver },
39    )
40}
41
42pub(crate) struct Sender<Req, Res> {
43    inner: futures::lock::Mutex<mpsc::Sender<(Req, oneshot::Sender<Res>)>>,
44}
45
46impl<Req, Res> Sender<Req, Res> {
47    pub(crate) async fn send(&self, req: Req) -> io::Result<Res> {
48        let (sender, receiver) = oneshot::channel();
49
50        self.inner
51            .lock()
52            .await
53            .send((req, sender))
54            .await
55            .map_err(|e| io::Error::new(io::ErrorKind::Other, e))?;
56        let res = receiver
57            .await
58            .map_err(|e| io::Error::new(io::ErrorKind::Other, e))?;
59
60        Ok(res)
61    }
62}
63
64pub(crate) struct Receiver<Req, Res> {
65    inner: mpsc::Receiver<(Req, oneshot::Sender<Res>)>,
66}
67
68impl<Req, Res> Receiver<Req, Res> {
69    pub(crate) fn poll_next_unpin(
70        &mut self,
71        cx: &mut Context<'_>,
72    ) -> Poll<Option<(Req, oneshot::Sender<Res>)>> {
73        self.inner.poll_next_unpin(cx)
74    }
75}
libp2p_webrtc/tokio/req_res_chan.rs

libp2p_webrtc/tokio/
req_res_chan.rs
