Merge pull request #203 from fundon/viz-echo

chore: add viz-echo example

Cargo.toml

@@ -24,6 +24,7 @@ tracing-subscriber = { version = "0.3.17", features = ["env-filter"] }
 criterion = { version = "0.5.1", features = ["html_reports"] }
 axum = "0.7.2"
 salvo = { version = "0.63.0", features = ["tower-compat"] }
+viz = "0.7.0"
 
 [workspace.package]
 version = "0.9.0"

README.md

@@ -1,6 +1,6 @@
 # [`Socketioxide`](https://github.com/totodore/socketioxide) 🚀🦀
 
-A [***`socket.io`***](https://socket.io) server implementation in Rust that integrates with the [***`Tower`***](https://tokio.rs/#tk-lib-tower) ecosystem and the [***`Tokio stack`***](https://tokio.rs). It integrates with any server framework based on tower like [***`Axum`***](https://docs.rs/axum/latest/axum/), [***`Warp`***](https://docs.rs/warp/latest/warp/), [***`Salvo`***](https://salvo.rs) or [***`Hyper`***](https://docs.rs/hyper/latest/hyper/). Add any other tower based middleware on top of socketioxide such as CORS, authorization, compression, etc with [***`tower-http`***](https://docs.rs/tower-http/latest/tower_http/).
+A [***`socket.io`***](https://socket.io) server implementation in Rust that integrates with the [***`Tower`***](https://tokio.rs/#tk-lib-tower) ecosystem and the [***`Tokio stack`***](https://tokio.rs). It integrates with any server framework based on tower like [***`Axum`***](https://docs.rs/axum/latest/axum/), [***`Warp`***](https://docs.rs/warp/latest/warp/), [***`Salvo`***](https://salvo.rs), [***`Viz`***](https://viz.rs) or [***`Hyper`***](https://docs.rs/hyper/latest/hyper/). Add any other tower based middleware on top of socketioxide such as CORS, authorization, compression, etc with [***`tower-http`***](https://docs.rs/tower-http/latest/tower_http/).
 
 > ⚠️ This crate is under active development and the API is not yet stable.
 
@@ -24,13 +24,15 @@ With the recent migration of all frameworks to hyper v1. It can be complicated t
 | [🦀Warp 0.3](https://docs.rs/warp/0.3/warp/)             | 0.14   | < 0.9  |
 | [🦀Salvo 0.63](https://docs.rs/salvo/latest/salvo)       | 1.0    | >= 0.9 |
 | [🦀Salvo 0.62](https://docs.rs/salvo/0.62/salvo)         | 1-rc*  | < 0.9  |
+| [🦀Viz 0.7](https://docs.rs/viz/latest/viz)              | 1.0    | >= 0.9 |
 
 ## Features :
 * Integrates with :
   * [Axum](https://docs.rs/axum/latest/axum/): [🏓echo example](./examples/axum-echo/axum_echo.rs)
   * [Warp](https://docs.rs/warp/latest/warp/): [🏓echo example](https://github.com/Totodore/socketioxide/blob/v0.8.0/examples/warp-echo/warp_echo.rs) (Not supported with `socketioxide >= 0.9.0` as long as warp doesn't migrate to hyper v1)
   * [Hyper](https://docs.rs/hyper/latest/hyper/): [🏓echo example](./examples/hyper-echo/hyper_echo.rs)
   * [Salvo](https://salvo.rs): [🏓echo example](./examples/salvo-echo/salvo_echo.rs)
+  * [Viz](https://viz.rs): [🏓echo example](./examples/viz-echo/viz_echo.rs)
 * Out of the box support for any other middleware based on tower :
   * [🔓CORS](https://docs.rs/tower-http/latest/tower_http/cors)
   * [📁Compression](https://docs.rs/tower-http/latest/tower_http/compression)

engineioxide/src/service/mod.rs

@@ -36,7 +36,6 @@ use bytes::Bytes;
 use http::{Request, Response};
 use http_body::Body;
 use http_body_util::Empty;
-use hyper::body::Incoming;
 use hyper::service::Service as HyperSvc;
 use tower::Service as TowerSvc;
 
@@ -210,12 +209,12 @@ impl<ReqBody> TowerSvc<Request<ReqBody>> for NotFoundService {
 }
 
 /// Implement a custom hyper [`Service`](HyperSvc) for the [`NotFoundService`]
-impl HyperSvc<Request<Incoming>> for NotFoundService {
+impl<ReqBody> HyperSvc<Request<ReqBody>> for NotFoundService {
     type Response = Response<ResponseBody<Empty<Bytes>>>;
     type Error = Infallible;
     type Future = Ready<Result<Response<ResponseBody<Empty<Bytes>>>, Infallible>>;
 
-    fn call(&self, _: Request<Incoming>) -> Self::Future {
+    fn call(&self, _: Request<ReqBody>) -> Self::Future {
         future::ready(Ok(Response::builder()
             .status(404)
             .body(ResponseBody::empty_response())

examples/viz-echo/Cargo.toml

@@ -0,0 +1,18 @@
+[package]
+name = "viz-echo"
+version.workspace = true
+edition.workspace = true
+
+[dependencies]
+socketioxide = { path = "../../socketioxide" }
+viz.workspace = true
+hyper.workspace = true
+http-body-util.workspace = true
+tokio = { workspace = true, features = ["rt-multi-thread", "macros"] }
+tracing-subscriber.workspace = true
+tracing.workspace = true
+serde_json.workspace = true
+
+[[bin]]
+name = "viz-echo"
+path = "viz_echo.rs"

examples/viz-echo/viz_echo.rs

@@ -0,0 +1,55 @@
+use serde_json::Value;
+use socketioxide::{
+    extract::{AckSender, Bin, Data, SocketRef},
+    SocketIo,
+};
+use std::sync::Arc;
+use tracing::info;
+use tracing_subscriber::FmtSubscriber;
+use viz::{handler::ServiceHandler, serve_with_upgrades, Result, Router, Tree};
+
+fn on_connect(socket: SocketRef, Data(data): Data<Value>) {
+    info!("Socket.IO connected: {:?} {:?}", socket.ns(), socket.id);
+    socket.emit("auth", data).ok();
+
+    socket.on(
+        "message",
+        |socket: SocketRef, Data::<Value>(data), Bin(bin)| {
+            info!("Received event: {:?} {:?}", data, bin);
+            socket.bin(bin).emit("message-back", data).ok();
+        },
+    );
+
+    socket.on(
+        "message-with-ack",
+        |Data::<Value>(data), ack: AckSender, Bin(bin)| {
+            info!("Received event: {:?} {:?}", data, bin);
+            ack.bin(bin).send(data).ok();
+        },
+    );
+}
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    tracing::subscriber::set_global_default(FmtSubscriber::default())?;
+
+    let (svc, io) = SocketIo::new_svc();
+
+    io.ns("/", on_connect);
+    io.ns("/custom", on_connect);
+
+    let app = Router::new()
+        .get("/", |_| async { Ok("Hello, World!") })
+        .any("/*", ServiceHandler::new(svc));
+    let tree = Arc::new(Tree::from(app));
+
+    info!("Starting server");
+
+    let listener = tokio::net::TcpListener::bind("0.0.0.0:3000").await.unwrap();
+
+    loop {
+        let (stream, addr) = listener.accept().await?;
+        let tree = tree.clone();
+        tokio::task::spawn(serve_with_upgrades(stream, tree, Some(addr)));
+    }
+}

socketioxide/Cargo.toml

@@ -51,6 +51,7 @@ engineioxide = { path = "../engineioxide", features = [
 tokio-tungstenite.workspace = true
 axum.workspace = true
 salvo.workspace = true
+viz.workspace = true
 tokio = { workspace = true, features = [
     "macros",
     "parking_lot",

socketioxide/benches/packet_decode.rs

@@ -1,6 +1,9 @@
 use criterion::{black_box, criterion_group, criterion_main, Criterion};
 use engineioxide::sid::Sid;
-use socketioxide::{Packet, PacketData, ProtocolVersion};
+use socketioxide::{
+    packet::{Packet, PacketData},
+    ProtocolVersion,
+};
 fn criterion_benchmark(c: &mut Criterion) {
     c.bench_function("Decode packet connect on /", |b| {
         let packet: String =

socketioxide/benches/packet_encode.rs

@@ -1,6 +1,9 @@
 use criterion::{black_box, criterion_group, criterion_main, Criterion};
 use engineioxide::sid::Sid;
-use socketioxide::{Packet, PacketData, ProtocolVersion};
+use socketioxide::{
+    packet::{Packet, PacketData},
+    ProtocolVersion,
+};
 fn criterion_benchmark(c: &mut Criterion) {
     c.bench_function("Encode packet connect on /", |b| {
         let packet = Packet::connect(black_box("/"), black_box(Sid::ZERO), ProtocolVersion::V5);