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
use std::{fmt, io};
use ntex_codec::{Decoder, Encoder};
use ntex_util::future::Either;
use crate::IoBoxed;
/// A codec-driven interface to an underlying I/O stream.
///
/// Uses the `Encoder` and `Decoder` traits to encode and decode frames.
pub struct Framed<U> {
io: IoBoxed,
codec: U,
}
impl<U> Framed<U> {
/// Wraps an I/O object with a codec.
pub fn new<Io>(io: Io, codec: U) -> Framed<U>
where
IoBoxed: From<Io>,
{
Framed {
codec,
io: IoBoxed::from(io),
}
}
#[inline]
/// Returns a reference to the underlying I/O stream wrapped by `Framed`.
pub fn get_io(&self) -> &IoBoxed {
&self.io
}
#[inline]
/// Returns a reference to the underlying codec.
pub fn get_codec(&self) -> &U {
&self.codec
}
#[inline]
/// Returns the underlying I/O object and codec.
pub fn into_inner(self) -> (IoBoxed, U) {
(self.io, self.codec)
}
}
impl<U> Framed<U>
where
U: Decoder + Encoder,
{
/// Flushes encoded data to the transport.
///
/// If `full` is `true`, waits until all buffered data has been written.
pub async fn flush(&self, full: bool) -> io::Result<()> {
self.io.flush(full).await
}
/// Gracefully shuts down the I/O stream.
pub async fn shutdown(&self) -> io::Result<()> {
self.io.shutdown().await
}
}
impl<U> Framed<U>
where
U: Decoder,
{
#[inline]
/// Reads and decodes the next item.
///
/// Returns `Ok(None)` when the connection can no longer produce another
/// complete item and no error was recorded. Codec errors are returned in
/// `Either::Left`; dispatcher timeouts and connection errors from the
/// transport, a filter, or shutdown are returned in `Either::Right`.
pub async fn recv(&self) -> Result<Option<U::Item>, Either<U::Error, io::Error>> {
self.io.recv(&self.codec).await
}
}
impl<U> Framed<U>
where
U: Encoder,
{
#[inline]
/// Encodes an item and fully flushes it to the transport.
///
/// Codec errors are returned in `Either::Left`; connection errors from the
/// transport, a filter, or shutdown, including an expired write timeout,
/// are returned in `Either::Right`.
pub async fn send(
&self,
item: <U as Encoder>::Item,
) -> Result<(), Either<U::Error, io::Error>> {
self.io.send(item, &self.codec).await
}
}
impl<U> fmt::Debug for Framed<U>
where
U: fmt::Debug,
{
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("Framed")
.field("codec", &self.codec)
.finish()
}
}
#[cfg(test)]
mod tests {
use ntex_bytes::Bytes;
use ntex_codec::BytesCodec;
use super::*;
use crate::{Io, testing::IoTest};
#[ntex::test]
async fn framed() {
let (client, server) = IoTest::create();
client.remote_buffer_cap(1024);
client.write(b"chunk-0");
let server = Framed::new(Io::from(server), BytesCodec);
server.get_codec();
server.get_io();
assert!(format!("{server:?}").contains("Framed"));
let item = server.recv().await.unwrap().unwrap();
assert_eq!(item, b"chunk-0".as_ref());
let data = Bytes::from_static(b"chunk-1");
server.send(data).await.unwrap();
server.flush(true).await.unwrap();
assert_eq!(client.read_any(), b"chunk-1".as_ref());
server.shutdown().await.unwrap();
assert!(client.is_closed());
let (io, _codec) = server.into_inner();
assert!(io.is_closed());
}
}