tcp_handler/streams/
compress.rs

1//! `TcpHandler` warps the [`crate::protocols::compress`] protocol.
2
3use bytes::{Buf, BytesMut};
4use tokio::io::{AsyncRead, AsyncWrite};
5use crate::protocols::common::{PacketError, StarterError};
6use crate::protocols::compress::{self, send, recv};
7use crate::streams::impl_tcp_handler;
8
9/// The server side `TcpHandler` of the `compress` protocol.
10#[derive(Debug)]
11pub struct TcpServerHandlerCompress<R: AsyncRead + Unpin, W: AsyncWrite + Unpin> {
12    reader: R,
13    writer: W,
14    version: String,
15}
16
17impl<R: AsyncRead + Unpin, W: AsyncWrite + Unpin> TcpServerHandlerCompress<R, W> {
18    /// Create and init a new `TcpServerHandlerCompress`.
19    pub async fn new<P: FnOnce(&str) -> bool>(mut reader: R, mut writer: W, identifier: &str, version_prediction: P, version: &str) -> Result<Self, StarterError> {
20        let init = compress::server_init(&mut reader, identifier, version_prediction).await;
21        let (_protocol_version, application_version) = compress::server_start(&mut writer, identifier, version, init).await?;
22        Ok(Self { reader, writer, version: application_version })
23    }
24
25    /// Deconstruct the `TcpServerHandlerCompress` into the inner parts.
26    #[inline]
27    pub fn into_inner(self) -> (R, W) {
28        (self.reader, self.writer)
29    }
30
31    /// **Unsafe!!!**
32    /// Construct the `TcpServerHandlerCompress` from the inner parts.
33    #[inline]
34    pub fn from_inner(reader: R, writer: W, version: String) -> Self {
35        Self { reader, writer, version }
36    }
37
38    /// Send a message to the client.
39    #[inline]
40    pub async fn send<B: Buf>(&mut self, message: &mut B) -> Result<(), PacketError> {
41        send(&mut self.writer, message).await
42    }
43
44    /// Receive a message from the client.
45    #[inline]
46    pub async fn recv(&mut self) -> Result<BytesMut, PacketError> {
47        recv(&mut self.reader).await
48    }
49}
50
51impl_tcp_handler!(server TcpServerHandlerCompress);
52
53
54/// The client side `TcpHandler` of the `compress` protocol.
55#[derive(Debug)]
56pub struct TcpClientHandlerCompress<R: AsyncRead + Unpin, W: AsyncWrite + Unpin> {
57    reader: R,
58    writer: W,
59}
60
61impl<R: AsyncRead + Unpin, W: AsyncWrite + Unpin> TcpClientHandlerCompress<R, W> {
62    /// Create and init a new `TcpClientHandlerCompress`.
63    pub async fn new(mut reader: R, mut writer: W, identifier: &str, version: &str) -> Result<Self, StarterError> {
64        let init = compress::client_init(&mut writer, identifier, version).await;
65        compress::client_start(&mut reader, init).await?;
66        Ok(Self { reader, writer })
67    }
68
69    /// Deconstruct the `TcpClientHandlerCompress` into the inner parts.
70    #[inline]
71    pub fn into_inner(self) -> (R, W) {
72        (self.reader, self.writer)
73    }
74
75    /// **Unsafe!!!**
76    /// Construct the `TcpClientHandlerCompress` from the inner parts.
77    #[inline]
78    pub fn from_inner(reader: R, writer: W) -> Self {
79        Self { reader, writer }
80    }
81
82    /// Send a message to the server.
83    #[inline]
84    pub async fn send<B: Buf>(&mut self, message: &mut B) -> Result<(), PacketError> {
85        send(&mut self.writer, message).await
86    }
87
88    /// Receive a message from the server.
89    #[inline]
90    pub async fn recv(&mut self) -> Result<BytesMut, PacketError> {
91        recv(&mut self.reader).await
92    }
93}
94
95impl_tcp_handler!(client TcpClientHandlerCompress);
96
97
98#[cfg(test)]
99mod tests {
100    use anyhow::Result;
101    use bytes::{Buf, BufMut};
102    use tokio::{spawn, try_join};
103    use variable_len_reader::{VariableReader, VariableWriter};
104    use crate::streams::compress::{TcpClientHandlerCompress, TcpServerHandlerCompress};
105    use crate::streams::tests::{check_send_recv, create};
106
107    #[tokio::test(flavor = "multi_thread")]
108    async fn connect() -> Result<()> {
109        let (cr, cw, sr, sw) = create().await?;
110        let server = spawn(TcpServerHandlerCompress::new(sr, sw, "test", |v| v == "0", "0"));
111        let client = spawn(TcpClientHandlerCompress::new(cr, cw, "test", "0"));
112        let (server, client) = try_join!(server, client)?;
113        let (mut server, mut client) = (server?, client?);
114
115        assert_eq!("0", server.get_client_version());
116        check_send_recv!(server, client, "Hello tcp-handler compress.");
117        check_send_recv!(client, server, "Hello tcp-handler compress.");
118        Ok(())
119    }
120}
tcp_handler/streams/compress.rs

tcp_handler/streams/
compress.rs
