Crate socketioxide_mongodb

A mongodb adapter implementation for the socketioxide crate.

The adapter is used to communicate with other nodes of the same application. This allows to broadcast messages to sockets connected on other servers, to get the list of rooms, to add or remove sockets from rooms, etc.

To achieve this, the adapter uses change streams on a collection. The message expiration process is either handled with TTL-indexes or a capped collection. If you change the message expiration strategy, make sure to first drop the collection. MongoDB doesn’t support switching from capped to TTL-indexes on an existing collection.

The Driver abstraction allows the use of any mongodb client. One implementation is provided:

• MongoDbDriver for the mongodb crate.

You can also implement your own driver by implementing the Driver trait.

The provided driver implementation is using change streams. They are only available on replica sets and sharded clusters. Make sure your mongodb server/cluster is configured accordingly.

Socketioxide-mongodb is not compatible with @socketio/mongodb-adapter and @socketio/mongodb-emitter. They use completely different protocols and cannot be used together. Do not mix socket.io JS servers with socketioxide rust servers.

Example with the default mongodb driver

async fn on_connect<A: Adapter>(socket: SocketRef<A>) {
    socket.join("room1");
    socket.on("event", on_event);
    let _ = socket.broadcast().emit("hello", "world").await.ok();
}
async fn on_event<A: Adapter>(socket: SocketRef<A>, Data(data): Data<String>) {}

const URI: &str = "mongodb://127.0.0.1:27017/?replicaSet=rs0&directConnection=true";
let client = mongodb::Client::with_uri_str(URI).await?;
let adapter = MongoDbAdapterCtr::new_with_mongodb(client.database("test")).await?;
let (layer, io) = SocketIo::builder()
    .with_adapter::<MongoDbAdapter<_>>(adapter)
    .build_layer();
Ok(())

Check the chat example for more complete examples.

How does it work?

The MongoDbAdapterCtr is a constructor for the MongoDbAdapter which is an implementation of the Adapter trait. The constructor takes a mongodb::Database as an argument and will configure a collection according to the chosen message expiration strategy (TTL indexes or capped collection).

Then, for each namespace, an adapter is created and it takes a corresponding CoreLocalAdapter. The CoreLocalAdapter allows to manage the local rooms and local sockets. The default LocalAdapter is simply a wrapper around this CoreLocalAdapter.

Once it is created the adapter is initialized with the MongoDbAdapter::init method. It will listen to changes on the event collection and emit heartbeats, messages are composed of a header (in bson) and a binary payload encoded with msgpack. Headers are used to filter and route messages to server/namespaces/event handlers.

All messages are encoded with msgpack.

There are 7 types of requests:

• Broadcast a packet to all the matching sockets.
• Broadcast a packet to all the matching sockets and wait for a stream of acks.
• Disconnect matching sockets.
• Get all the rooms.
• Add matching sockets to rooms.
• Remove matching sockets to rooms.
• Fetch all the remote sockets matching the options.
• Heartbeat
• Initial heartbeat. When receiving a initial heartbeat all other servers reply a heartbeat immediately.

For ack streams, the adapter will first send a BroadcastAckCount response to the server that sent the request, and then send the acks as they are received (more details in MongoDbAdapter::broadcast_with_ack fn).

On the other side, each time an action has to be performed on the local server, the adapter will first broadcast a request to all the servers and then perform the action locally.

Modules

drivers

Drivers are an abstraction over the pub/sub backend used by the adapter. You can use the provided implementation or implement your own.

Structs

CustomMongoDbAdapter

The mongodb adapter implementation. It is generic over the Driver used to communicate with the mongodb server. And over the SocketEmitter used to communicate with the local server. This allows to avoid cyclic dependencies between the adapter, socketioxide-core and socketioxide crates.

InitRes

The result of the init future.

MongoDbAdapterConfig

The configuration of the MongoDbAdapter.

MongoDbAdapterCtr

The adapter constructor. For each namespace you define, a new adapter instance is created from this constructor.

Enums

Error

Represent any error that might happen when using this adapter.

MessageExpirationStrategy

The strategy used to remove old documents in the mongodb collection. The default mongodb driver supports both TTL indexes and capped collections.

Type Aliases

MongoDbAdaptermongodb

The mongodb adapter with the mongodb driver.