1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
//! Shared `split` / `into_split` halves macro for stream types.
macro_rules! define_stream_split {
($stream:ty) => {
/// Borrowed read half of a stream, created by `split`.
pub struct ReadHalf<'a> {
stream: &'a $stream,
}
/// Borrowed write half of a stream, created by `split`.
pub struct WriteHalf<'a> {
stream: &'a $stream,
}
/// Owned read half of a stream, created by `into_split`.
pub struct OwnedReadHalf {
stream: std::rc::Rc<$stream>,
}
/// Owned write half of a stream, created by `into_split`.
pub struct OwnedWriteHalf {
stream: std::rc::Rc<$stream>,
}
/// Error returned when reuniting mismatched [`OwnedReadHalf`] and [`OwnedWriteHalf`].
pub struct ReuniteError(pub OwnedReadHalf, pub OwnedWriteHalf);
impl std::fmt::Debug for ReuniteError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("ReuniteError").finish_non_exhaustive()
}
}
impl std::fmt::Display for ReuniteError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str("tried to reunite halves that do not originate from the same stream")
}
}
impl std::error::Error for ReuniteError {}
impl<'a> ReadHalf<'a> {
/// Read into `buf`.
pub async fn read<B>(&self, buf: B) -> crate::buf::BufResult<usize, B>
where
B: crate::buf::IoBufMut + Send + 'static,
{
self.stream.read(buf).await
}
/// Read until `buf` is full.
pub async fn read_exact<B>(&self, buf: B) -> crate::buf::BufResult<(), B>
where
B: crate::buf::IoBufMut + Send + 'static,
{
self.stream.read_exact(buf).await
}
}
impl<'a> WriteHalf<'a> {
/// Write from `buf`.
pub async fn write<B>(&self, buf: B) -> crate::buf::BufResult<usize, B>
where
B: crate::buf::IoBuf + Send + 'static,
{
self.stream.write(buf).await
}
/// Write all initialized bytes from `buf`.
pub async fn write_all<B>(&self, buf: B) -> crate::buf::BufResult<(), B>
where
B: crate::buf::IoBuf + Send + 'static,
{
self.stream.write_all(buf).await
}
/// Shut down the write half.
pub fn shutdown(&self, how: std::net::Shutdown) -> std::io::Result<()> {
self.stream.shutdown(how)
}
}
impl OwnedReadHalf {
/// Read into `buf`.
pub async fn read<B>(&self, buf: B) -> crate::buf::BufResult<usize, B>
where
B: crate::buf::IoBufMut + Send + 'static,
{
self.stream.read(buf).await
}
/// Read until `buf` is full.
pub async fn read_exact<B>(&self, buf: B) -> crate::buf::BufResult<(), B>
where
B: crate::buf::IoBufMut + Send + 'static,
{
self.stream.read_exact(buf).await
}
/// Attempt to reunite this read half with its matching write half.
pub fn reunite(self, write: OwnedWriteHalf) -> Result<$stream, ReuniteError> {
if std::rc::Rc::ptr_eq(&self.stream, &write.stream) {
drop(write);
Ok(std::rc::Rc::try_unwrap(self.stream)
.unwrap_or_else(|_| unreachable!("only two Rc handles existed")))
} else {
Err(ReuniteError(self, write))
}
}
}
impl OwnedWriteHalf {
/// Write from `buf`.
pub async fn write<B>(&self, buf: B) -> crate::buf::BufResult<usize, B>
where
B: crate::buf::IoBuf + Send + 'static,
{
self.stream.write(buf).await
}
/// Write all initialized bytes from `buf`.
pub async fn write_all<B>(&self, buf: B) -> crate::buf::BufResult<(), B>
where
B: crate::buf::IoBuf + Send + 'static,
{
self.stream.write_all(buf).await
}
/// Shut down the write half.
pub fn shutdown(&self, how: std::net::Shutdown) -> std::io::Result<()> {
self.stream.shutdown(how)
}
/// Attempt to reunite this write half with its matching read half.
pub fn reunite(self, read: OwnedReadHalf) -> Result<$stream, ReuniteError> {
read.reunite(self)
}
}
impl $stream {
/// Split this stream into borrowed read and write halves.
pub fn split(&self) -> (ReadHalf<'_>, WriteHalf<'_>) {
(ReadHalf { stream: self }, WriteHalf { stream: self })
}
/// Split this stream into owned read and write halves (`Rc`-backed, `!Send`).
pub fn into_split(self) -> (OwnedReadHalf, OwnedWriteHalf) {
let rc = std::rc::Rc::new(self);
(
OwnedReadHalf {
stream: std::rc::Rc::clone(&rc),
},
OwnedWriteHalf { stream: rc },
)
}
}
};
}