refactor!: Reimplement stream using futures and without split

Cargo.toml

@@ -12,7 +12,7 @@ exclude = [".github"]
 [features]
 default = ["stream"]
 expose_stream = []
-stream = ["dep:bytes", "dep:tokio", "dep:tokio-rustls"]
+stream = ["dep:futures-io"]
 
 # <Forward to imap-codec>
 arbitrary = ["imap-codec/arbitrary"]
@@ -30,11 +30,9 @@ ext_metadata = ["imap-codec/ext_metadata"]
 # </Forward to imap-codec>
 
 [dependencies]
-bytes = { version = "1.7.1", optional = true }
+futures-io = { version = "0.3.31", optional = true }
 imap-codec = { version = "2.0.0-alpha.5", features = ["quirk_crlf_relaxed"] }
 thiserror = "1.0.63"
-tokio = { version = "1.40.0", optional = true, features = ["io-util", "macros", "net"] }
-tokio-rustls = { version = "0.26.0", optional = true, default-features = false }
 tracing = "0.1.40"
 
 [dev-dependencies]
@@ -43,6 +41,7 @@ imap-next = { path = ".", features = ["tag_generator"] }
 
 rand = "0.8.5"
 tokio = { version = "1.40.0", features = ["full"] }
+tokio-util = { version = "0.7.12", features = ["compat"] }
 
 [workspace]
 resolver = "2"

README.md

@@ -55,10 +55,11 @@ use imap_next::{
     stream::Stream,
 };
 use tokio::net::TcpStream;
+use tokio_util::compat::TokioAsyncReadCompatExt;
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn Error>> {
-    let mut stream = Stream::insecure(TcpStream::connect("127.0.0.1:1143").await?);
+    let mut stream = Stream::new(TcpStream::connect("127.0.0.1:1143").await?.compat());
     let mut client = Client::new(Options::default());
 
     loop {

examples/client.rs

@@ -7,11 +7,12 @@ use imap_next::{
     stream::Stream,
 };
 use tokio::net::TcpStream;
+use tokio_util::compat::TokioAsyncReadCompatExt;
 
 #[tokio::main(flavor = "current_thread")]
 async fn main() {
     let stream = TcpStream::connect("127.0.0.1:12345").await.unwrap();
-    let mut stream = Stream::insecure(stream);
+    let mut stream = Stream::new(stream.compat());
     let mut client = Client::new(Options::default());
 
     let greeting = loop {

examples/client_authenticate.rs

@@ -10,11 +10,12 @@ use imap_next::{
     stream::Stream,
 };
 use tokio::net::TcpStream;
+use tokio_util::compat::TokioAsyncReadCompatExt;
 
 #[tokio::main(flavor = "current_thread")]
 async fn main() {
     let stream = TcpStream::connect("127.0.0.1:12345").await.unwrap();
-    let mut stream = Stream::insecure(stream);
+    let mut stream = Stream::new(stream.compat());
     let mut client = Client::new(Options::default());
 
     loop {

examples/client_idle.rs

@@ -10,11 +10,12 @@ use imap_next::{
     stream::Stream,
 };
 use tokio::{net::TcpStream, sync::mpsc::Receiver};
+use tokio_util::compat::TokioAsyncReadCompatExt;
 
 #[tokio::main(flavor = "current_thread")]
 async fn main() {
     let stream = TcpStream::connect("127.0.0.1:12345").await.unwrap();
-    let mut stream = Stream::insecure(stream);
+    let mut stream = Stream::new(stream.compat());
     let mut client = Client::new(Options::default());
 
     loop {

examples/server.rs

@@ -6,12 +6,13 @@ use imap_next::{
     stream::Stream,
 };
 use tokio::net::TcpListener;
+use tokio_util::compat::TokioAsyncReadCompatExt;
 
 #[tokio::main(flavor = "current_thread")]
 async fn main() {
     let listener = TcpListener::bind("127.0.0.1:12345").await.unwrap();
     let (stream, _) = listener.accept().await.unwrap();
-    let mut stream = Stream::insecure(stream);
+    let mut stream = Stream::new(stream.compat());
     let mut server = Server::new(
         Options::default(),
         Greeting::ok(None, "server (example)").unwrap(),

examples/server_authenticate.rs

@@ -5,12 +5,13 @@ use imap_next::{
     types::CommandAuthenticate,
 };
 use tokio::net::TcpListener;
+use tokio_util::compat::TokioAsyncReadCompatExt;
 
 #[tokio::main(flavor = "current_thread")]
 async fn main() {
     let listener = TcpListener::bind("127.0.0.1:12345").await.unwrap();
     let (stream, _) = listener.accept().await.unwrap();
-    let mut stream = Stream::insecure(stream);
+    let mut stream = Stream::new(stream.compat());
     let mut server = Server::new(
         Options::default(),
         Greeting::ok(None, "server_idle (example)").unwrap(),

examples/server_idle.rs

@@ -11,12 +11,13 @@ use imap_next::{
     stream::Stream,
 };
 use tokio::{net::TcpListener, sync::mpsc::Receiver};
+use tokio_util::compat::TokioAsyncReadCompatExt;
 
 #[tokio::main(flavor = "current_thread")]
 async fn main() {
     let listener = TcpListener::bind("127.0.0.1:12345").await.unwrap();
     let (stream, _) = listener.accept().await.unwrap();
-    let mut stream = Stream::insecure(stream);
+    let mut stream = Stream::new(stream.compat());
     let mut server = Server::new(
         Options::default(),
         Greeting::ok(None, "server_idle (example)").unwrap(),

integration-test/Cargo.toml

@@ -8,9 +8,10 @@ publish = false
 [dependencies]
 bstr = { version = "1.10.0", default-features = false }
 bytes = "1.7.1"
-imap-codec = { version = "2.0.0-alpha.4" }
+imap-codec = { version = "2.0.0-alpha.5" }
 imap-next = { path = ".." }
-tokio = { version = "1.40.0", features = ["macros", "net", "rt", "time"] }
+tokio = { version = "1.40.0", features = ["io-util", "macros", "net", "rt", "time"] }
+tokio-util = { version = "0.7.12", features = ["compat"] }
 tracing = "0.1.40"
 tracing-subscriber = { version = "0.3.18", features = ["env-filter"] }
 

integration-test/src/client_tester.rs

@@ -7,6 +7,7 @@ use imap_next::{
     stream::{self, Stream},
 };
 use tokio::net::TcpStream;
+use tokio_util::compat::{Compat, TokioAsyncReadCompatExt};
 use tracing::trace;
 
 use crate::codecs::Codecs;
@@ -25,7 +26,7 @@ impl ClientTester {
     ) -> Self {
         let stream = TcpStream::connect(server_address).await.unwrap();
         trace!(?server_address, "Client is connected");
-        let stream = Stream::insecure(stream);
+        let stream = Stream::new(stream.compat());
         let client = Client::new(client_options);
         Self {
             codecs,
@@ -344,13 +345,16 @@ impl ClientTester {
 #[allow(clippy::large_enum_variant)]
 enum ConnectionState {
     /// Connection to server established.
-    Connected { stream: Stream, client: Client },
+    Connected {
+        stream: Stream<Compat<TcpStream>>,
+        client: Client,
+    },
     /// Connection dropped.
     Disconnected,
 }
 
 impl ConnectionState {
-    fn connected(&mut self) -> (&mut Stream, &mut Client) {
+    fn connected(&mut self) -> (&mut Stream<Compat<TcpStream>>, &mut Client) {
         match self {
             ConnectionState::Connected { stream, client } => (stream, client),
             ConnectionState::Disconnected => panic!("Client is already disconnected"),

integration-test/src/server_tester.rs

@@ -4,7 +4,8 @@ use imap_next::{
     server::{self, ResponseHandle, Server},
     stream::{self, Stream},
 };
-use tokio::net::TcpListener;
+use tokio::net::{TcpListener, TcpStream};
+use tokio_util::compat::{Compat, TokioAsyncReadCompatExt};
 use tracing::trace;
 
 use crate::codecs::Codecs;
@@ -24,7 +25,7 @@ impl ServerTester {
     ) -> Self {
         let (stream, client_address) = server_listener.accept().await.unwrap();
         trace!(?client_address, "Server accepts connection");
-        let stream = Stream::insecure(stream);
+        let stream = Stream::new(stream.compat());
         Self {
             codecs,
             server_options,
@@ -308,15 +309,20 @@ impl ServerTester {
 #[allow(clippy::large_enum_variant)]
 enum ConnectionState {
     // Connection to client established.
-    Connected { stream: Stream },
+    Connected {
+        stream: Stream<Compat<TcpStream>>,
+    },
     // Server greeted client.
-    Greeted { stream: Stream, server: Server },
+    Greeted {
+        stream: Stream<Compat<TcpStream>>,
+        server: Server,
+    },
     // Connection dropped.
     Disconnected,
 }
 
 impl ConnectionState {
-    fn greeted(&mut self) -> (&mut Stream, &mut Server) {
+    fn greeted(&mut self) -> (&mut Stream<Compat<TcpStream>>, &mut Server) {
         match self {
             ConnectionState::Connected { .. } => panic!("Server has not greeted yet"),
             ConnectionState::Greeted { stream, server } => (stream, server),