Struct SocketIo 

pub struct SocketIo<A: Adapter = LocalAdapter>(/* private fields */);

The SocketIo instance can be cheaply cloned and moved around everywhere in your program. It can be used as the main handle to access the whole socket.io context.

You can also use it as an extractor for all your handlers.

It is generic over the Adapter type. If you plan to use it with another adapter than the default, make sure to have a handler that is generic over the adapter type.

Implementations

impl SocketIo<LocalAdapter>

pub fn builder() -> SocketIoBuilder

Create a new SocketIoBuilder with a default config

pub fn new_svc() -> (SocketIoService<NotFoundService>, SocketIo)

Create a new SocketIoService and a SocketIo instance with a default config. This service will be a standalone service that return a 404 error for every non-socket.io request. It can be used as a Service (see hyper example)

pub fn new_inner_svc<S: Clone>(svc: S) -> (SocketIoService<S>, SocketIo)

Create a new SocketIoService and a SocketIo instance with a default config. It can be used as a Service with an inner service

pub fn new_layer() -> (SocketIoLayer, SocketIo)

Build a SocketIoLayer and a SocketIo instance with a default config. It can be used as a tower Layer (see axum example)

impl<A: Adapter> SocketIo<A>

pub fn config(&self) -> &SocketIoConfig

Return a reference to the SocketIoConfig used by this SocketIo instance

pub fn dyn_ns<C, T>( &self, path: impl Into<String>, callback: C, ) -> Result<(), NsInsertError>

where C: ConnectHandler<A, T>, T: Send + Sync + 'static,

Register a ConnectHandler for the given dynamic namespace.

You can specify dynamic parts in the path by using the {name} syntax. Note that any static namespace will take precedence over a dynamic one.

For more info about namespace routing, see the matchit router documentation.

The dynamic namespace will create a child namespace for any path that matches the given pattern with the given handler.

• See the connect module doc for more details on connect handler.
• See the extract module doc for more details on available extractors.

Errors

If the pattern is invalid, a NsInsertError will be returned.

Example

let (_, io) = SocketIo::new_svc();
io.dyn_ns("/client/{client_id}", async |socket: SocketRef| {
    println!("Socket connected on dynamic namespace with namespace path: {}", socket.ns());
}).unwrap();

let (_, io) = SocketIo::new_svc();
io.dyn_ns("/client/{*remaining_path}", async |socket: SocketRef| {
    println!("Socket connected on dynamic namespace with namespace path: {}", socket.ns());
}).unwrap();

pub fn delete_ns(&self, path: impl AsRef<str>)

Delete the namespace with the given path.

This will disconnect all sockets connected to this namespace in a deferred way.

Panics

If the v4 protocol (legacy) is enabled and the namespace to delete is the default namespace “/”. For v4, the default namespace cannot be deleted. See official doc for more informations.

pub async fn close(&self)

Gracefully close all the connections and drop every sockets

Any on_disconnect handler will called with DisconnectReason::ClosingServer

pub fn nsps(&self) -> Vec<BroadcastOperators<A>>

Get all the namespaces to perform operations on.

This will return a vector of BroadcastOperators for each namespace.

Example

let (_, io) = SocketIo::new_svc();
io.ns("/custom_ns", async |socket: SocketRef| {
    println!("Socket connected on /custom_ns namespace with id: {}", socket.id);
});

// Later in your code you can get all the namespaces
for ns in io.nsps() {
    assert_eq!(ns.ns_path(), "/custom_ns");
}

pub fn of(&self, path: impl AsRef<str>) -> Option<BroadcastOperators<A>>

Select a specific namespace to perform operations on.

Currently you cannot select a dynamic namespace with this method.

Example

let (_, io) = SocketIo::new_svc();
io.ns("custom_ns", async |socket: SocketRef| {
    println!("Socket connected on /custom_ns namespace with id: {}", socket.id);
});

// Later in your code you can select the custom_ns namespace
// and show all sockets connected to it
async fn test(io: SocketIo) {
    let sockets = io.of("custom_ns").unwrap().sockets();
    for socket in sockets {
       println!("found socket on /custom_ns namespace with id: {}", socket.id);
    }
}

pub fn to(&self, rooms: impl RoomParam) -> BroadcastOperators<A>

Alias for io.of("/").unwrap().to(). If the default namespace “/” is not found this fn will panic!

Select all the sockets in the given rooms except for the current socket.

When called from a socket, if you want to also include it, use the within() operator.

However, when called from the io (global context) level, there will be no difference.

Example

async fn handler(socket: SocketRef, io: SocketIo, Data(data): Data::<Value>) {
    // Emit a message to all sockets in room1, room2, room3, and room4, except the current socket
    socket
        .to("room1")
        .to(["room2", "room3"])
        .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)
        .await;
}

let (_, io) = SocketIo::new_svc();
io.ns("/", async |s: SocketRef| s.on("test", handler));

pub fn within(&self, rooms: impl RoomParam) -> BroadcastOperators<A>

Alias for io.of("/").unwrap().within(). If the default namespace “/” is not found this fn will panic!

Select all the sockets in the given rooms (current one included).

This includes the current socket, in contrast to the to() operator.

However, when called from the io (global context) level, there will be no difference.

Example

async fn handler(socket: SocketRef, Data(data): Data::<Value>) {
    let other_rooms = "room4".to_string();
    // Emit a message to all sockets in room1, room2, room3, and room4, including the current socket
    socket
        .within("room1")
        .within(["room2", "room3"])
        .within(vec![other_rooms])
        .emit("test", &data)
        .await;
}

let (_, io) = SocketIo::new_svc();
io.ns("/", async |s: SocketRef| s.on("test", handler));

pub fn except(&self, rooms: impl RoomParam) -> BroadcastOperators<A>

Alias for io.of("/").unwrap().except(). If the default namespace “/” is not found this fn will panic!

Filter out all sockets selected with the previous operators that are in the specified rooms.

Example

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).await;
}

let (_, io) = SocketIo::new_svc();
io.ns("/", async |socket: SocketRef| {
    socket.on("register1", async |s: SocketRef| s.join("room1"));
    socket.on("register2", async |s: SocketRef| s.join("room2"));
    socket.on("test", handler);
});

pub fn local(&self) -> BroadcastOperators<A>

Alias for io.of("/").unwrap().local(). If the default namespace “/” is not found this fn will panic!

Broadcast to all sockets only connected to this node.

When using the default in-memory adapter, this operator is a no-op.

Example

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).await;
}

let (_, io) = SocketIo::new_svc();
io.ns("/", async |s: SocketRef| s.on("test", handler));

pub fn timeout(&self, timeout: Duration) -> BroadcastOperators<A>

Alias for io.of("/").unwrap().timeout(). If the default namespace “/” is not found this fn will panic!

Set a custom timeout when sending a message with an acknowledgement.

• See SocketIoBuilder::ack_timeout for the default timeout.
• See emit_with_ack() for more details on acknowledgements.

Example

async fn handler(socket: SocketRef, Data(data): Data::<Value>) {
    // Emit a test message in the room1 and room3 rooms, except for the room2
    // room with the binary payload received, and wait for 5 seconds for an acknowledgement
    socket.to("room1")
          .to("room3")
          .except("room2")
          .timeout(Duration::from_secs(5))
          .emit_with_ack::<_, Value>("message-back", &data)
          .await
          .unwrap()
          .for_each(async |(id, ack)| {
            match ack {
                Ok(ack) => println!("Ack received from socket {}: {:?}", id, ack),
                Err(err) => println!("Ack error from socket {}: {:?}", id, err),
            }
          }).await;
}

let (_, io) = SocketIo::new_svc();
io.ns("/", async |s: SocketRef| s.on("test", handler));

pub async fn emit<T: ?Sized + Serialize>( &self, event: impl AsRef<str>, data: &T, ) -> Result<(), BroadcastError>

Alias for io.of("/").unwrap().emit(). If the default namespace “/” is not found this fn will panic!

Emit a message to one or many clients

If you provide tuple-like data (tuples, arrays), it will be considered as multiple arguments. Therefore, if you want to send an array as the first argument of the payload, you need to wrap it in an array or a tuple. A Vec will always be considered a single argument.

Emitting binary data

To emit binary data, you must use a data type that implements Serialize as binary data. Currently, if you use Vec<u8>, it will be considered a sequence of numbers rather than binary data. To handle this, you can either use a special type like Bytes or the serde_bytes crate. If you want to emit generic data that may contain binary, use rmpv::Value instead of serde_json::Value, as binary data will otherwise be serialized as a sequence of numbers.

Errors

• When encoding the data, a SendError::Serialize may be returned.
• If the underlying engine.io connection is closed, a SendError::Socket(SocketError::Closed) will be returned, and the data you attempted to send will be included in the error.
• If the packet buffer is full, a SendError::Socket(SocketError::InternalChannelFull) will be returned, and the data you attempted to send will be included in the error. See the SocketIoBuilder::max_buffer_size option for more information on internal buffer configuration.

Single-socket example

async fn handler(socket: SocketRef, Data(data): Data::<Value>) {
    // Emit a test message to the client
    socket.emit("test", &data).ok();
    // Emit a test message with multiple arguments to the client
    socket.emit("test", &("world", "hello", 1)).ok();
    // Emit a test message with an array as the first argument
    let arr = [1, 2, 3, 4];
    socket.emit("test", &[arr]).ok();
}

let (_, io) = SocketIo::new_svc();
io.ns("/", async |socket: SocketRef| socket.on("test", handler));

Single-socket binary example with the bytes crate

async fn handler(socket: SocketRef, Data(data): Data::<(String, Bytes, Bytes)>) {
    // Emit a test message to the client
    socket.emit("test", &data).ok();
    // Emit a test message with multiple arguments to the client
    socket.emit("test", &("world", "hello", Bytes::from_static(&[1, 2, 3, 4]))).ok();
    // Emit a test message with an array as the first argument
    let arr = [1, 2, 3, 4];
    socket.emit("test", &[arr]).ok();
}

let (_, io) = SocketIo::new_svc();
io.ns("/", async |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.

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).await;

    // Emit a test message with multiple arguments to the client
    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]).await;
}

let (_, io) = SocketIo::new_svc();
io.ns("/", async |s: SocketRef| s.on("test", handler));

pub async fn emit_with_ack<T: ?Sized + Serialize, V>( &self, event: impl AsRef<str>, data: &T, ) -> Result<AckStream<V, A>, EmitWithAckError>

Alias for io.of("/").unwrap().emit_with_ack(). If the default namespace “/” is not found this fn will panic!

Emit a message to one or many clients and wait for one or more acknowledgments.

See emit() for more details on emitting messages.

The acknowledgment has a timeout specified in the config (5 seconds by default) (see SocketIoBuilder::ack_timeout) or can be set with the timeout() operator.

To receive acknowledgments, an AckStream is returned. It can be used in two ways:

• As a Stream: This will yield all the acknowledgment responses, along with the corresponding socket ID, received from the client. This is useful when broadcasting to multiple sockets and expecting more than one acknowledgment. To get the socket from this ID, use io::get_socket().
• As a Future: This will yield the first acknowledgment response received from the client, useful when expecting only one acknowledgment.

Errors

If packet encoding fails, an ParserError is immediately returned.

If the socket is full or if it is closed before receiving the acknowledgment, a SendError::Socket will be immediately returned, and the value to send will be given back.

If the client does not respond before the timeout, the AckStream will yield an AckError::Timeout. If the data sent by the client is not deserializable as V, an AckError::Decode will be yielded.

Single-socket example

async fn handler(socket: SocketRef, Data(data): Data::<Value>) {
    // Emit a test message and wait for an acknowledgment with the timeout specified in the global config
    match socket.emit_with_ack::<_, Value>("test", &data).unwrap().await {
        Ok(ack) => println!("Ack received {:?}", ack),
        Err(err) => println!("Ack error {:?}", err),
    }
}

let (_, io) = SocketIo::new_svc();
io.ns("/", async |socket: SocketRef| socket.on("test", handler));

Single-socket example with custom acknowledgment timeout

async fn handler(socket: SocketRef, Data(data): Data::<Value>) {
    // Emit a test message and wait for an acknowledgment with the timeout specified here
    match socket.timeout(Duration::from_millis(2)).emit_with_ack::<_, Value>("test", &data).unwrap().await {
        Ok(ack) => println!("Ack received {:?}", ack),
        Err(err) => println!("Ack error {:?}", err),
    }
}

let (_, io) = SocketIo::new_svc();
io.ns("/", async |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.

async fn handler(socket: SocketRef, Data(data): Data::<Value>) {
    // Emit a test message in the room1 and room3 rooms,
    // except for room2, with the binary payload received
    let ack_stream = socket.to("room1")
        .to("room3")
        .except("room2")
        .emit_with_ack::<_, String>("message-back", &data)
        .await
        .unwrap();
    ack_stream.for_each(async |(id, ack)| {
        match ack {
            Ok(ack) => println!("Ack received, socket {} {:?}", id, ack),
            Err(err) => println!("Ack error, socket {} {:?}", id, err),
        }
    }).await;
}

let (_, io) = SocketIo::new_svc();
io.ns("/", async |s: SocketRef| s.on("test", handler));

pub fn sockets(&self) -> Vec<SocketRef<A>>

Alias for io.of("/").unwrap().sockets(). If the default namespace “/” is not found this fn will panic!

Get all the local sockets selected with the previous operators.

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 only work for local sockets. Use fetch_sockets to get remote sockets.

Example

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();
    for socket in sockets {
        println!("Socket extension: {:?}", socket.extensions.get::<String>());
    }
}

let (_, io) = SocketIo::new_svc();
io.ns("/", async |s: SocketRef| s.on("test", handler));

pub async fn fetch_sockets(&self) -> Result<Vec<RemoteSocket<A>>, A::Error>

Alias for io.of("/").unwrap().fetch_sockets(). If the default namespace “/” is not found this fn will panic!

Get all the local and remote sockets selected with the previous operators.

Use sockets() if you only have a single node.

Avoid using this method if you want to immediately perform actions on the sockets. Instead, directly apply the actions using operators:

Correct Approach

io.within("room1").emit("foo", "bar").await.unwrap();
io.within("room1").disconnect().await.unwrap();

Incorrect Approach

let sockets = io.within("room1").fetch_sockets().await.unwrap();
for socket in sockets {
    socket.emit("test", &"Hello").await.unwrap();
    socket.leave("room1").await.unwrap();
}

Example

let (_, io) = SocketIo::new_svc();
let sockets = io.within("room1").fetch_sockets().await.unwrap();
for socket in sockets {
    println!("Socket ID: {:?}", socket.data().id);
}

pub async fn disconnect(&self) -> Result<(), BroadcastError>

Alias for io.of("/").unwrap().disconnect(). If the default namespace “/” is not found this fn will panic!

Disconnect all sockets selected with the previous operators.

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

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("/", async |s: SocketRef| s.on("test", handler));

Example from the io struct

async fn handler(socket: SocketRef, io: SocketIo) {
    // Disconnect all sockets in the room1 and room3 rooms, except for room2.
    io.within("room1").within("room3").except("room2").disconnect().await.unwrap();
}
let (_, io) = SocketIo::new_svc();
io.ns("/", async |s: SocketRef| s.on("test", handler));

pub async fn join(self, rooms: impl RoomParam) -> Result<(), A::Error>

Alias for io.of("/").unwrap().join(). If the default namespace “/” is not found this fn will panic!

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

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"]).await.unwrap();
    // We should retrieve all the local sockets that are in room3 and room5
    let sockets = socket.within("room4").within("room5").sockets();
}

let (_, io) = SocketIo::new_svc();
io.ns("/", async |s: SocketRef| s.on("test", handler));

pub async fn rooms(&self) -> Result<Vec<Room>, A::Error>

Alias for io.of("/").unwrap().rooms(). If the default namespace “/” is not found this fn will panic!

Get all the rooms selected with the previous operators.

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

async fn handler(socket: SocketRef, io: SocketIo) {
    println!("Socket connected to the / namespace with id: {}", socket.id);
    let rooms = io.rooms().await.unwrap();
    println!("All rooms in the / namespace: {:?}", rooms);
}

let (_, io) = SocketIo::new_svc();
io.ns("/", handler);

pub async fn leave(self, rooms: impl RoomParam) -> Result<(), A::Error>

Alias for io.of("/").unwrap().rooms(). If the default namespace “/” is not found this fn will panic!

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

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"]).await.unwrap();
}

let (_, io) = SocketIo::new_svc();
io.ns("/", async |s: SocketRef| s.on("test", handler));

pub fn get_socket(&self, sid: Sid) -> Option<SocketRef<A>>

Alias for io.of("/").unwrap().get_socket(). If the default namespace “/” is not found this fn will panic!

Get a local SocketRef by the specified Sid.

This will only work for local sockets. Use fetch_socket to get remote sockets.

pub fn broadcast(&self) -> BroadcastOperators<A>

Alias for io.of("/").unwrap().broadcast(). If the default namespace “/” is not found this fn will panic!

Broadcast to all sockets without any filtering (except the current socket).

If you want to include the current socket use the broadcast operators from the io global context.

Example

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).await;
    // This message will be broadcast to all sockets in this namespace, including this one.
    io.emit("test", &data).await;
}

let (_, io) = SocketIo::new_svc();
io.ns("/", async |s: SocketRef| s.on("test", handler));

impl<A: DefinedAdapter + Adapter> SocketIo<A>

pub fn ns<C, T>( &self, path: impl Into<Cow<'static, str>>, callback: C, ) -> A::InitRes

where C: ConnectHandler<A, T>, T: Send + Sync + 'static,

Register a ConnectHandler for the given namespace

• See the connect module doc for more details on connect handler.
• See the extract module doc for more details on available extractors.

Simple example with an async closure:

#[derive(Debug, Serialize, Deserialize)]
struct MyData {
    name: String,
    age: u8,
}

let (_, io) = SocketIo::new_svc();
io.ns("/", async |socket: SocketRef| {
    // Register a handler for the "test" event and extract the data as a `MyData` struct
    // With the Data extractor, the handler is called only if the data can be deserialized as a `MyData` struct
    // If you want to manage errors yourself you can use the TryData extractor
    socket.on("test", async |socket: SocketRef, Data::<MyData>(data)| {
        println!("Received a test message {:?}", data);
        socket.emit("test-test", &MyData { name: "Test".to_string(), age: 8 }).ok(); // Emit a message to the client
    });
});

Example with a closure and an acknowledgement + binary data:

#[derive(Debug, Serialize, Deserialize)]
struct MyData {
    name: String,
    age: u8,
}

let (_, io) = SocketIo::new_svc();
io.ns("/", async |socket: SocketRef| {
    // Register an async handler for the "test" event and extract the data as a `MyData` struct
    // Extract the binary payload as a `Vec<Bytes>` with the Bin extractor.
    // It should be the last extractor because it consumes the request
    socket.on("test", async |socket: SocketRef, Data::<MyData>(data), ack: AckSender| {
        println!("Received a test message {:?}", data);
        tokio::time::sleep(std::time::Duration::from_secs(1)).await;
        ack.send(&data).ok(); // The data received is sent back to the client through the ack
        socket.emit("test-test", &MyData { name: "Test".to_string(), age: 8 }).ok(); // Emit a message to the client
    });
});

Example with a closure and an authentication process:

#[derive(Debug, Deserialize)]
struct MyAuthData {
    token: String,
}
#[derive(Debug, Serialize, Deserialize)]
struct MyData {
    name: String,
    age: u8,
}

let (_, io) = SocketIo::new_svc();
io.ns("/", async |socket: SocketRef, Data(auth): Data<MyAuthData>| {
    if auth.token.is_empty() {
        println!("Invalid token, disconnecting");
        socket.disconnect().ok();
        return;
    }
    socket.on("test", async |socket: SocketRef, Data::<MyData>(data)| {
        println!("Received a test message {:?}", data);
        socket.emit("test-test", &MyData { name: "Test".to_string(), age: 8 }).ok(); // Emit a message to the client
    });
});

With remote adapters, this method is only available on a defined adapter:

// The SocketIo instance is generic over the adapter type.
async fn test<A: Adapter>(io: SocketIo<A>) {
    io.ns("/", async || ());
}

// The SocketIo instance is not generic over the adapter type.
async fn test(io: SocketIo<LocalAdapter>) {
    io.ns("/", async || ());
}
async fn test_default_adapter(io: SocketIo) {
    io.ns("/", async || ());
}

Trait Implementations

impl<A: Adapter> Clone for SocketIo<A>

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl<A: Adapter> Debug for SocketIo<A>

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl<A: Adapter> From<Arc<Client<A>>> for SocketIo<A>

fn from(client: Arc<Client<A>>) -> Self

Converts to this type from the input type.

impl<A: Adapter> FromConnectParts<A> for SocketIo<A>

type Error = Infallible

The error type returned by the extractor

fn from_connect_parts( s: &Arc<Socket<A>>, _: &Option<Value>, ) -> Result<Self, Self::Error>

Extract the arguments from the connect event. If it fails, the handler is not called

impl<A: Adapter> FromDisconnectParts<A> for SocketIo<A>

type Error = Infallible

The error type returned by the extractor

fn from_disconnect_parts( s: &Arc<Socket<A>>, _: DisconnectReason, ) -> Result<Self, Self::Error>

Extract the arguments from the disconnect event. If it fails, the handler is not called

impl<A: Adapter> FromMessageParts<A> for SocketIo<A>

type Error = Infallible

The error type returned by the extractor

fn from_message_parts( s: &Arc<Socket<A>>, _: &mut Value, _: &Option<i64>, ) -> Result<Self, Self::Error>

Extract the arguments from the message event. If it fails, the handler is not called.