libp2p_quic/connection/
stream.rs

1// Copyright 2022 Protocol Labs.
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::{self},
23    pin::Pin,
24    task::{Context, Poll},
25};
26
27use futures::{AsyncRead, AsyncWrite};
28
29/// A single stream on a connection
30pub struct Stream {
31    /// A send part of the stream
32    send: quinn::SendStream,
33    /// A receive part of the stream
34    recv: quinn::RecvStream,
35    /// Whether the stream is closed or not
36    close_result: Option<Result<(), io::ErrorKind>>,
37}
38
39impl Stream {
40    pub(super) fn new(send: quinn::SendStream, recv: quinn::RecvStream) -> Self {
41        Self {
42            send,
43            recv,
44            close_result: None,
45        }
46    }
47}
48
49impl AsyncRead for Stream {
50    fn poll_read(
51        mut self: Pin<&mut Self>,
52        cx: &mut Context,
53        buf: &mut [u8],
54    ) -> Poll<io::Result<usize>> {
55        if let Some(close_result) = self.close_result {
56            if close_result.is_err() {
57                return Poll::Ready(Ok(0));
58            }
59        }
60        Pin::new(&mut self.recv).poll_read(cx, buf)
61    }
62}
63
64impl AsyncWrite for Stream {
65    fn poll_write(
66        mut self: Pin<&mut Self>,
67        cx: &mut Context,
68        buf: &[u8],
69    ) -> Poll<io::Result<usize>> {
70        Pin::new(&mut self.send)
71            .poll_write(cx, buf)
72            .map_err(Into::into)
73    }
74
75    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<io::Result<()>> {
76        Pin::new(&mut self.send).poll_flush(cx)
77    }
78
79    fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<io::Result<()>> {
80        if let Some(close_result) = self.close_result {
81            // For some reason poll_close needs to be 'fuse'able
82            return Poll::Ready(close_result.map_err(Into::into));
83        }
84        let close_result = futures::ready!(Pin::new(&mut self.send).poll_close(cx));
85        self.close_result = Some(close_result.as_ref().map_err(|e| e.kind()).copied());
86        Poll::Ready(close_result)
87    }
88}
libp2p_quic/connection/stream.rs

libp2p_quic/connection/
stream.rs
