chore(ci): run ci on all PR

.github/workflows/github-ci.yml

@@ -5,8 +5,6 @@ on:
     branches:
       - main
   pull_request:
-    branches:
-      - main
 
 jobs:
   format:

crates/socketioxide/docs/operators/broadcast.md

@@ -7,11 +7,11 @@ If you want to include the current socket use emit operators from the [`io`] glo
 ```rust
 # use socketioxide::{SocketIo, extract::*};
 # use serde_json::Value;
-fn handler(io: SocketIo, socket: SocketRef, Data(data): Data::<Value>) {
+async fn handler(io: SocketIo, socket: SocketRef, Data(data): Data::<Value>) {
     // This message will be broadcast to all sockets in this namespace except this one.
-    socket.broadcast().emit("test", &data);
+    socket.broadcast().emit("test", &data).await;
     // This message will be broadcast to all sockets in this namespace, including this one.
-    io.emit("test", &data);
+    io.emit("test", &data).await;
 }
 
 let (_, io) = SocketIo::new_svc();

crates/socketioxide/docs/operators/disconnect.md

@@ -1,11 +1,25 @@
 # Disconnect all sockets selected with the previous operators.
 
-# Example
+This will return a `Future` that must be awaited because socket.io may communicate with remote instances
+if you use horizontal scaling through remote adapters.
+
+# Example from a socket
+```rust
+# use socketioxide::{SocketIo, extract::*};
+async fn handler(socket: SocketRef) {
+    // Disconnect all sockets in the room1 and room3 rooms, except for room2.
+    socket.within("room1").within("room3").except("room2").disconnect().await.unwrap();
+}
+let (_, io) = SocketIo::new_svc();
+io.ns("/", |s: SocketRef| s.on("test", handler));
+```
+
+# Example from the io struct
 ```rust
 # use socketioxide::{SocketIo, extract::*};
-fn handler(socket: SocketRef) {
+async fn handler(socket: SocketRef, io: SocketIo) {
     // Disconnect all sockets in the room1 and room3 rooms, except for room2.
-    socket.within("room1").within("room3").except("room2").disconnect().unwrap();
+    io.within("room1").within("room3").except("room2").disconnect().await.unwrap();
 }
 let (_, io) = SocketIo::new_svc();
 io.ns("/", |s: SocketRef| s.on("test", handler));

crates/socketioxide/docs/operators/emit.md

@@ -44,9 +44,7 @@ fn handler(socket: SocketRef, Data(data): Data::<Value>) {
 }
 
 let (_, io) = SocketIo::new_svc();
-io.ns("/", |socket: SocketRef| {
-    socket.on("test", handler);
-});
+io.ns("/", |socket: SocketRef| socket.on("test", handler));
 ```
 
 # Single-socket binary example with the `bytes` crate
@@ -66,27 +64,29 @@ fn handler(socket: SocketRef, Data(data): Data::<(String, Bytes, Bytes)>) {
 }
 
 let (_, io) = SocketIo::new_svc();
-io.ns("/", |socket: SocketRef| {
-    socket.on("test", handler);
-});
+io.ns("/", |socket: SocketRef| socket.on("test", handler));
 ```
 
 # Broadcast example
+
+Here the emit method will return a `Future` that must be awaited because socket.io may communicate
+with remote instances if you use horizontal scaling through remote adapters.
+
 ```rust
 # use socketioxide::{SocketIo, extract::*};
 # use serde_json::Value;
 # use std::sync::Arc;
 # use bytes::Bytes;
-fn handler(socket: SocketRef, Data(data): Data::<(String, Bytes, Bytes)>) {
+async fn handler(socket: SocketRef, Data(data): Data::<(String, Bytes, Bytes)>) {
     // Emit a test message in the room1 and room3 rooms, except for room2, with the received binary payload
-    socket.to("room1").to("room3").except("room2").emit("test", &data);
+    socket.to("room1").to("room3").except("room2").emit("test", &data).await;
 
     // Emit a test message with multiple arguments to the client
-    socket.to("room1").emit("test", &("world", "hello", 1)).ok();
+    socket.to("room1").emit("test", &("world", "hello", 1)).await;
 
     // Emit a test message with an array as the first argument
     let arr = [1, 2, 3, 4];
-    socket.to("room2").emit("test", &[arr]).ok();
+    socket.to("room2").emit("test", &[arr]).await;
 }
 
 let (_, io) = SocketIo::new_svc();

crates/socketioxide/docs/operators/emit_with_ack.md

@@ -45,9 +45,7 @@ async fn handler(socket: SocketRef, Data(data): Data::<Value>) {
 }
 
 let (_, io) = SocketIo::new_svc();
-io.ns("/", |socket: SocketRef| {
-    socket.on("test", handler);
-});
+io.ns("/", |socket: SocketRef| socket.on("test", handler));
 ```
 
 # Single-socket example with custom acknowledgment timeout
@@ -65,12 +63,14 @@ async fn handler(socket: SocketRef, Data(data): Data::<Value>) {
 }
 
 let (_, io) = SocketIo::new_svc();
-io.ns("/", |socket: SocketRef| {
-    socket.on("test", handler);
-});
+io.ns("/", |socket: SocketRef| socket.on("test", handler));
 ```
 
 # Broadcast example
+
+Here the emit method will return a `Future` that must be awaited because socket.io may communicate
+with remote instances if you use horizontal scaling through remote adapters.
+
 ```rust
 # use socketioxide::{SocketIo, extract::*};
 # use serde_json::Value;
@@ -82,6 +82,7 @@ async fn handler(socket: SocketRef, Data(data): Data::<Value>) {
         .to("room3")
         .except("room2")
         .emit_with_ack::<_, String>("message-back", &data)
+        .await
         .unwrap();
     ack_stream.for_each(|(id, ack)| async move {
         match ack {

crates/socketioxide/docs/operators/except.md

@@ -4,16 +4,16 @@
 ```rust
 # use socketioxide::{SocketIo, extract::*};
 # use serde_json::Value;
-fn handler(socket: SocketRef, Data(data): Data::<Value>) {
+async fn handler(socket: SocketRef, Data(data): Data::<Value>) {
     // This message will be broadcast to all sockets in the namespace,
     // except for those in room1 and the current socket
-    socket.broadcast().except("room1").emit("test", &data);
+    socket.broadcast().except("room1").emit("test", &data).await;
 }
 
 let (_, io) = SocketIo::new_svc();
 io.ns("/", |socket: SocketRef| {
-    socket.on("register1", |s: SocketRef| s.join("room1").unwrap());
-    socket.on("register2", |s: SocketRef| s.join("room2").unwrap());
+    socket.on("register1", |s: SocketRef| async move { s.join("room1").await.unwrap() });
+    socket.on("register2", |s: SocketRef| async move { s.join("room2").await.unwrap() });
     socket.on("test", handler);
 });
 ```

crates/socketioxide/docs/operators/join.md

@@ -1,12 +1,16 @@
 # Add all sockets selected with the previous operators to the specified room(s).
 
+This will return a `Future` that must be awaited because socket.io may communicate with remote instances
+if you use horizontal scaling through remote adapters.
+
 # Example
 ```rust
 # use socketioxide::{SocketIo, extract::*};
-fn handler(socket: SocketRef) {
+async fn handler(socket: SocketRef) {
     // Add all sockets that are in room1 and room3 to room4 and room5
-    socket.within("room1").within("room3").join(["room4", "room5"]).unwrap();
-    let sockets = socket.within("room4").within("room5").sockets().unwrap();
+    socket.within("room1").within("room3").join(["room4", "room5"]).await.unwrap();
+    // We should retrieve all the sockets that are in room3 and room5
+    let sockets = socket.within("room4").within("room5").sockets().await.unwrap();
 }
 
 let (_, io) = SocketIo::new_svc();

crates/socketioxide/docs/operators/leave.md

@@ -1,12 +1,14 @@
 # Remove all sockets selected with the previous operators from the specified room(s).
 
+This will return a `Future` that must be awaited because socket.io may communicate with remote instances
+if you use horizontal scaling through remote adapters.
+
 # Example
 ```rust
 # use socketioxide::{SocketIo, extract::*};
-fn handler(socket: SocketRef) {
+async fn handler(socket: SocketRef) {
     // Remove all sockets that are in room1 and room3 from room4 and room5
-    socket.within("room1").within("room3").leave(["room4", "room5"]).unwrap();
-    let sockets = socket.within("room4").within("room5").sockets().unwrap();
+    socket.within("room1").within("room3").leave(["room4", "room5"]).await.unwrap();
 }
 
 let (_, io) = SocketIo::new_svc();

crates/socketioxide/docs/operators/local.md

@@ -5,10 +5,10 @@ When using the default in-memory adapter, this operator is a no-op.
 ```rust
 # use socketioxide::{SocketIo, extract::*};
 # use serde_json::Value;
-fn handler(socket: SocketRef, Data(data): Data::<Value>) {
+async fn handler(socket: SocketRef, Data(data): Data::<Value>) {
     // This message will be broadcast to all sockets in this
     // namespace that are connected to this node
-    socket.local().emit("test", &data);
+    socket.local().emit("test", &data).await;
 }
 
 let (_, io) = SocketIo::new_svc();

crates/socketioxide/docs/operators/rooms.md

@@ -1,11 +1,14 @@
 # Get all room names in the current namespace.
 
+This will return a `Future` that must be awaited because socket.io may communicate with remote instances
+if you use horizontal scaling through remote adapters.
+
 # Example
 ```rust
 # use socketioxide::{SocketIo, extract::SocketRef};
-fn handler(socket: SocketRef, io: SocketIo) {
+async fn handler(socket: SocketRef, io: SocketIo) {
     println!("Socket connected to the / namespace with id: {}", socket.id);
-    let rooms = io.rooms().unwrap();
+    let rooms = io.rooms().await.unwrap();
     println!("All rooms in the / namespace: {:?}", rooms);
 }
 

crates/socketioxide/docs/operators/sockets.md

@@ -2,12 +2,15 @@
 
 This can be used to retrieve any extension data (with the `extensions` feature enabled) from the sockets or to make certain sockets join other rooms.
 
+This will return a `Future` that must be awaited because socket.io may communicate with remote instances
+if you use horizontal scaling through remote adapters.
+
 # Example
 ```rust
 # use socketioxide::{SocketIo, extract::*};
-fn handler(socket: SocketRef) {
+async fn handler(socket: SocketRef) {
     // Find extension data in each socket in the room1 and room3 rooms, except for room2
-    let sockets = socket.within("room1").within("room3").except("room2").sockets().unwrap();
+    let sockets = socket.within("room1").within("room3").except("room2").sockets().await.unwrap();
     for socket in sockets {
         println!("Socket custom string: {:?}", socket.extensions.get::<String>());
     }

crates/socketioxide/docs/operators/timeout.md

@@ -17,6 +17,7 @@ async fn handler(socket: SocketRef, Data(data): Data::<Value>) {
           .except("room2")
           .timeout(Duration::from_secs(5))
           .emit_with_ack::<_, Value>("message-back", &data)
+          .await
           .unwrap()
           .for_each(|(id, ack)| async move {
             match ack {

crates/socketioxide/docs/operators/to.md

@@ -15,13 +15,15 @@ async fn handler(socket: SocketRef, io: SocketIo, Data(data): Data::<Value>) {
     socket
         .to("room1")
         .to(["room2", "room3"])
-        .emit("test", &data);
+        .emit("test", &data)
+        .await;
 
     // Emit a message to all sockets in room1, room2, room3, and room4, including the current socket
     io
         .to("room1")
         .to(["room2", "room3"])
-        .emit("test", &data);
+        .emit("test", &data)
+        .await;
 }
 
 let (_, io) = SocketIo::new_svc();

crates/socketioxide/docs/operators/within.md

@@ -18,7 +18,8 @@ async fn handler(socket: SocketRef, Data(data): Data::<Value>) {
         .within("room1")
         .within(["room2", "room3"])
         .within(vec![other_rooms])
-        .emit("test", &data);
+        .emit("test", &data)
+        .await;
 }
 
 let (_, io) = SocketIo::new_svc();

crates/socketioxide/src/ack.rs

@@ -94,6 +94,7 @@ pin_project_lite::pin_project! {
     ///
     ///     // We apply the `for_each` StreamExt fn to the AckStream
     ///     socket.broadcast().emit_with_ack::<_, String>("test", "test")
+    ///         .await
     ///         .unwrap()
     ///         .for_each(|(id, ack)| async move { println!("Ack: {} {:?}", id, ack); }).await;
     /// });

crates/socketioxide/src/io.rs

@@ -490,9 +490,11 @@ impl<A: Adapter> SocketIo<A> {
     ///
     /// // Later in your code you can select the custom_ns namespace
     /// // and show all sockets connected to it
-    /// let sockets = io.of("custom_ns").unwrap().sockets().unwrap();
-    /// for socket in sockets {
-    ///    println!("found socket on /custom_ns namespace with id: {}", socket.id);
+    /// async fn test(io: SocketIo) {
+    ///     let sockets = io.of("custom_ns").unwrap().sockets().await.unwrap();
+    ///     for socket in sockets {
+    ///        println!("found socket on /custom_ns namespace with id: {}", socket.id);
+    ///     }
     /// }
     /// ```
     #[inline]

crates/socketioxide/src/operators.rs

@@ -30,7 +30,7 @@ use crate::{
 /// A trait for types that can be used as a room parameter.
 ///
 /// [`String`], [`Vec<String>`], [`Vec<&str>`], [`&'static str`](str) and const arrays are implemented by default.
-pub trait RoomParam: 'static {
+pub trait RoomParam: Send + 'static {
     /// The type of the iterator returned by `into_room_iter`.
     type IntoIter: Iterator<Item = Room>;
 
@@ -351,7 +351,6 @@ impl<A: Adapter> BroadcastOperators<A> {
     }
 
     #[doc = include_str!("../docs/operators/sockets.md")]
-<<<<<<< HEAD
     pub async fn sockets(self) -> Result<Vec<SocketRef<A>>, A::Error> {
         self.ns.adapter.fetch_sockets(self.opts).await
     }
@@ -362,42 +361,20 @@ impl<A: Adapter> BroadcastOperators<A> {
     }
 
     #[doc = include_str!("../docs/operators/join.md")]
-    pub async fn join(self, rooms: impl RoomParam) -> Result<(), A::Error> {
-        self.ns.adapter.add_sockets(self.opts, rooms).await
+    #[allow(clippy::manual_async_fn)] // related to issue: https://github.com/rust-lang/rust-clippy/issues/12664
+    pub fn join(self, rooms: impl RoomParam) -> impl Future<Output = Result<(), A::Error>> + Send {
+        async move { self.ns.adapter.add_sockets(self.opts, rooms).await }
     }
 
     #[doc = include_str!("../docs/operators/leave.md")]
-    pub async fn leave(self, rooms: impl RoomParam) -> Result<(), A::Error> {
-        self.ns.adapter.del_sockets(self.opts, rooms).await
+    #[allow(clippy::manual_async_fn)] // related to issue: https://github.com/rust-lang/rust-clippy/issues/12664
+    pub fn leave(self, rooms: impl RoomParam) -> impl Future<Output = Result<(), A::Error>> + Send {
+        async move { self.ns.adapter.del_sockets(self.opts, rooms).await }
     }
 
     #[doc = include_str!("../docs/operators/rooms.md")]
     pub async fn rooms(self) -> Result<Vec<Room>, A::Error> {
         self.ns.adapter.rooms().await
-=======
-    pub fn sockets(self) -> Result<Vec<SocketRef<A>>, A::Error> {
-        self.ns.adapter.fetch_sockets(self.opts)
-    }
-
-    #[doc = include_str!("../docs/operators/disconnect.md")]
-    pub fn disconnect(self) -> Result<(), Vec<DisconnectError>> {
-        self.ns.adapter.disconnect_socket(self.opts)
-    }
-
-    #[doc = include_str!("../docs/operators/join.md")]
-    pub fn join(self, rooms: impl RoomParam) -> Result<(), A::Error> {
-        self.ns.adapter.add_sockets(self.opts, rooms)
-    }
-
-    #[doc = include_str!("../docs/operators/leave.md")]
-    pub fn leave(self, rooms: impl RoomParam) -> Result<(), A::Error> {
-        self.ns.adapter.del_sockets(self.opts, rooms)
-    }
-
-    #[doc = include_str!("../docs/operators/rooms.md")]
-    pub fn rooms(self) -> Result<Vec<Room>, A::Error> {
-        self.ns.adapter.rooms()
->>>>>>> feat-adapter-rework
     }
 
     #[doc = include_str!("../docs/operators/get_socket.md")]