remove bulk bench

Signed-off-by: iGxnon <[email protected]>

.github/workflows/ci.yml

@@ -54,7 +54,6 @@ jobs:
       - name: Run bench with codecov
         run: |
           cargo llvm-cov --no-report --bench micro --features="micro-bench"
-          cargo llvm-cov --no-report --bench bulk
 
       - name: Generate codecov report
         run: |

Cargo.toml

@@ -43,11 +43,3 @@ micro-bench = []
 name = "micro"
 harness = false
 required-features = ["micro-bench"]
-
-[[bench]]
-name = "bulk"
-harness = false
-
-[profile.bench]
-opt-level = 3
-lto = true

examples/proxy.rs

@@ -4,9 +4,8 @@ use std::net::SocketAddr;
 use bytes::Bytes;
 use futures::{SinkExt, StreamExt};
 use raknet_rs::client::{self, ConnectTo};
-use raknet_rs::io::IO;
 use raknet_rs::server::{self, MakeIncoming};
-use raknet_rs::Reliability;
+use raknet_rs::{Message, Reliability};
 use tokio::net::UdpSocket;
 
 #[tokio::main]
@@ -27,13 +26,12 @@ async fn main() -> Result<(), Box<dyn Error>> {
 
     tokio::spawn(async move {
         loop {
-            let io = incoming.next().await.unwrap();
+            let (src, dst) = incoming.next().await.unwrap();
             tokio::spawn(async move {
-                tokio::pin!(io);
-                println!("[server] set default reliability to Reliable");
-                io.as_mut().set_default_reliability(Reliability::Reliable);
+                tokio::pin!(src);
+                tokio::pin!(dst);
                 loop {
-                    if let Some(data) = io.next().await {
+                    if let Some(data) = src.next().await {
                         println!(
                             "[server] got proxy data: '{}'",
                             String::from_utf8_lossy(&data)
@@ -45,7 +43,13 @@ async fn main() -> Result<(), Box<dyn Error>> {
                             .send()
                             .await
                             .unwrap();
-                        io.send(res.bytes().await.unwrap()).await.unwrap();
+                        dst.send(Message::new(
+                            Reliability::Reliable,
+                            0,
+                            res.bytes().await.unwrap(),
+                        ))
+                        .await
+                        .unwrap();
                         continue;
                     }
                     break;
@@ -62,7 +66,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
 async fn client(addr: SocketAddr, name: &str) -> Result<(), Box<dyn Error>> {
     let socket = UdpSocket::bind("0.0.0.0:0").await?;
     println!("[{name}] I am listening on {}", socket.local_addr()?);
-    let conn = socket
+    let (src, dst) = socket
         .connect_to(
             addr,
             client::Config::new()
@@ -72,10 +76,15 @@ async fn client(addr: SocketAddr, name: &str) -> Result<(), Box<dyn Error>> {
                 .protocol_version(11),
         )
         .await?;
-    tokio::pin!(conn);
-    conn.send(Bytes::from_static(b"Hello, Anyone there?"))
-        .await?;
-    let res = conn.next().await.unwrap();
+    tokio::pin!(src);
+    tokio::pin!(dst);
+    dst.send(Message::new(
+        Reliability::Reliable,
+        0,
+        Bytes::from_static(b"Hello, Anyone there?"),
+    ))
+    .await?;
+    let res = src.next().await.unwrap();
     println!(
         "[{name}] got server response: {}",
         String::from_utf8_lossy(&res)

examples/tracing.rs

@@ -9,7 +9,7 @@ use fastrace::collector::{SpanContext, SpanId, SpanRecord, TraceId};
 use fastrace::Span;
 use futures::{SinkExt, StreamExt};
 use raknet_rs::client::{self, ConnectTo};
-use raknet_rs::io::{TraceInfo, IO};
+use raknet_rs::opts::TraceInfo;
 use raknet_rs::server::{self, MakeIncoming};
 use raknet_rs::{Message, Reliability};
 use tokio::net::UdpSocket;
@@ -37,14 +37,13 @@ async fn main() -> Result<(), Box<dyn Error>> {
 
     tokio::spawn(async move {
         loop {
-            let io = incoming.next().await.unwrap();
+            let (reader, writer) = incoming.next().await.unwrap();
             tokio::spawn(async move {
-                let (read, write) = IO::split(io);
-                tokio::pin!(read);
-                tokio::pin!(write);
+                tokio::pin!(reader);
+                tokio::pin!(writer);
                 loop {
-                    if let Some(data) = read.next().await {
-                        let trace_id = read.last_trace_id().unwrap_or_else(|| {
+                    if let Some(data) = reader.next().await {
+                        let trace_id = reader.last_trace_id().unwrap_or_else(|| {
                             eprintln!("Please run with `--features fastrace/enable` and try again");
                             exit(0)
                         });
@@ -55,7 +54,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
                         // do something with data
                         tokio::time::sleep(Duration::from_millis(10)).await;
                         let _span = Span::enter_with_parent("user child span", &root_span);
-                        write
+                        writer
                             .send(Message::new(Reliability::ReliableOrdered, 0, data))
                             .await
                             .unwrap();
@@ -76,7 +75,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
 
 async fn client(addr: SocketAddr) -> Result<(), Box<dyn Error>> {
     let socket = UdpSocket::bind("0.0.0.0:0").await?;
-    let conn = socket
+    let (src, dst) = socket
         .connect_to(
             addr,
             client::Config::new()
@@ -86,11 +85,22 @@ async fn client(addr: SocketAddr) -> Result<(), Box<dyn Error>> {
                 .protocol_version(11),
         )
         .await?;
-    tokio::pin!(conn);
-    conn.send(Bytes::from_static(b"User pack1")).await?;
-    conn.send(Bytes::from_static(b"User pack2")).await?;
-    let pack1 = conn.next().await.unwrap();
-    let pack2 = conn.next().await.unwrap();
+    tokio::pin!(src);
+    tokio::pin!(dst);
+    dst.send(Message::new(
+        Reliability::ReliableOrdered,
+        0,
+        Bytes::from_static(b"User pack1"),
+    ))
+    .await?;
+    dst.send(Message::new(
+        Reliability::ReliableOrdered,
+        0,
+        Bytes::from_static(b"User pack2"),
+    ))
+    .await?;
+    let pack1 = src.next().await.unwrap();
+    let pack2 = src.next().await.unwrap();
     assert_eq!(pack1, Bytes::from_static(b"User pack1"));
     assert_eq!(pack2, Bytes::from_static(b"User pack2"));
     Ok(())

src/client/conn/mod.rs

@@ -1,9 +1,12 @@
 use std::io;
 use std::net::ToSocketAddrs;
 
+use bytes::Bytes;
+use futures::{Sink, Stream};
+
 use super::handler::offline;
-use crate::io::{Ping, IO};
-use crate::{codec, Role};
+use crate::opts::Ping;
+use crate::{codec, Message, Role};
 
 /// Connection implementation by using tokio's UDP framework
 #[cfg(feature = "tokio-udp")]
@@ -130,5 +133,8 @@ pub trait ConnectTo: Sized {
         self,
         addr: impl ToSocketAddrs,
         config: Config,
-    ) -> io::Result<impl IO + Ping>;
+    ) -> io::Result<(
+        impl Stream<Item = Bytes>,
+        impl Sink<Message, Error = io::Error> + Ping,
+    )>;
 }