Socket.IO

This page shows how to set up Socket.IO in a Dox app with dox_socketio.

It can also run on its own, see Without Dox.

dox_socketio is a Socket.IO v4 server for Dart. The official clients connect to it with no changes: socket.io-client (JavaScript v4) and the Dart socket_io_client package. It has acks, rooms, namespaces, binary data and long-polling. Socket.IO is the way to do realtime in Dox.

In a Dox app

Socket.IO runs on the same port as your HTTP routes, at /socket.io/. You add a provider, a config, a namespace and one env key.

1. Install

dart pub add dox_socketio

2. The provider

The provider starts Socket.IO in every isolate. Put it in lib/providers/, so the generated container finds it:

lib/providers/socketio_provider.dart

import 'package:dox_socketio/dox_socketio.dart';
import 'package:my_app/config/socketio.dart';

class AppSocketIoProvider extends SocketIoProvider {
  AppSocketIoProvider() : super(socketIoConfig);
}

3. The config

The config says which namespaces to serve:

lib/config/socketio.dart

import 'package:dox_socketio/dox_socketio.dart';
import 'package:my_app/app/socket/chat.namespace.dart';
import 'package:my_app/config/env.dart';

SocketIoConfig socketIoConfig = SocketIoConfig(
  enabled: env.socketioEnabled,
  namespaces: <NamespaceHandler>[ChatNamespace()],
);

`SocketIoConfig` option	Default	What it does
`enabled`	`true`	serve Socket.IO at `options.path`
`options`	`ServerOptions()`	see Options
`namespaces`	`[]`	`NamespaceHandler` classes to register
`adapter`	`null` (in memory with one isolate, `IsolateAdapter` with many)	makes one adapter in each isolate, see Many isolates

4. A namespace

A namespace handles the events of one path, like /chat:

lib/app/socket/chat.namespace.dart

import 'package:dox_socketio/dox_socketio.dart';

class ChatNamespace extends NamespaceHandler {
  @override
  String get path => '/chat';

  @override
  void onConnection(Socket socket) {
    socket.emit('welcome', <String, dynamic>{'id': socket.id});
    socket.join('lobby');
    socket.on('message', onMessage);
  }

  void onMessage(SocketEvent event) {
    event.socket.to('lobby').emit('message', event.data); // everyone in lobby but me
    event.ack('sent'); // answer the client callback
  }
}

See Namespaces for middleware and dependencies.

5. Turn it on

Add a getter for the SOCKETIO_ENABLED key to lib/config/env.dart:

lib/config/env.dart

  /// serve Socket.IO on the app port
  bool get socketioEnabled => false;

.env

SOCKETIO_ENABLED=true

Then run dart run build_runner build and dox serve.

6. Connect a client

import { io } from 'socket.io-client';

const socket = io('http://localhost:3000/chat', { transports: ['websocket'] });
socket.on('welcome', (data) => console.log(data.id));
socket.emit('message', 'hi', (answer) => console.log(answer)); // 'sent'

Use transports: ['websocket']. See Many isolates for why.

Send from a controller or a service

SocketIoProvider.server is the Socket.IO server of this isolate:

await SocketIoProvider.server?.of('/chat').to('lobby').emit('news', data);

From a queue job or another program, use the emitter.

Requests to /socket.io/ go to Socket.IO before the router. Middleware and body reading are skipped. To log users in, see Log in a socket.
With no provider, or with enabled: false, there is no extra work per request.
Need a plain WebSocket with no Socket.IO protocol? Route.websocket(path, handler) in dox_core gives you the raw socket.

Many isolates

Dox runs one isolate per CPU core, and each isolate has its own Socket.IO server. A client is connected to one of them.

With many isolates and no adapter, the provider uses the IsolateAdapter. It sends each broadcast to the other isolates of the process (through IsolateBus of dox_core), so it reaches the clients of every isolate. socketsJoin, fetchSockets, serverSideEmit and broadcast acks work across isolates too. You need no Redis.

Adapter	Reaches
`InMemoryAdapter` (one isolate)	the clients of this isolate
`IsolateAdapter` (many isolates, the default)	the clients of every isolate of this process
`RedisAdapter`	the clients of every isolate of every server

Clients should connect with websocket only

One websocket stays on one isolate. HTTP long-polling sends many requests, and they can reach different isolates. Long-polling only works with totalIsolate: 1, or with sticky sessions in front of many single-isolate servers. Dox logs this hint at startup when it runs many isolates and polling is allowed.

With more than one server process, or next to Node.js servers, use the Redis adapter.

Without Dox

dox_socketio also runs on its own, with no Dox app:

import 'package:dox_socketio/dox_socketio.dart';

Future<void> main() async {
  SocketIoServer io = SocketIoServer();
  io.register(ChatNamespace());
  await io.listen(3001);
}

io.listen(port) starts a native server (dox_http_native) on its own port. Run it in its own small program, for example bin/socket.dart, and put it behind the same proxy as the HTTP server.
io.attach(httpServer) uses a server you already have: a NativeHttpServer or a dart:io one.
In your own server, call io.handle(request) when io.matches(request) is true.
io.on('connection', (Socket socket) {...}) handles the default namespace / with no class.

Namespaces

A namespace is a class. Its dependencies come in through the constructor.

class ChatNamespace extends NamespaceHandler {
  ChatNamespace(this.messages);

  final MessageService messages;

  @override
  String get path => '/chat';

  @override
  List<Middleware> get middleware => <Middleware>[checkToken];

  void checkToken(Socket socket) {
    if (socket.handshake.auth['token'] != 'secret') {
      throw SocketIoException('unauthorized', data: <String, dynamic>{'reason': 'bad token'});
    }
  }

  @override
  void onConnection(Socket socket) {
    socket.on('message', onMessage);
  }

  void onMessage(SocketEvent event) {
    event.ack(messages.save(event.data));
  }
}

// in lib/config/socketio.dart
namespaces: <NamespaceHandler>[ChatNamespace(MessageService())],

When a middleware throws a SocketIoException, the client gets connect_error with the message and the data. Any other error is written to stderr. The client only gets {"message": "Internal error"}, so it can not see private text like SQL.

A client that is refused is still closed after connectTimeout, unless it joins another namespace.

Log in a socket

Socket.IO gets its requests before the router, so route middleware (session, auth) does not run for it. Use SocketAuth as a namespace middleware. Your function gets the handshake and gives the user, or null.

lib/app/socket/chat.namespace.dart

class ChatNamespace extends NamespaceHandler {
  @override
  String get path => '/chat';

  @override
  List<Middleware> get middleware => <Middleware>[SocketAuth(findUser).handle];

  static Future<User?> findUser(Handshake handshake) async {
    String? token = handshake.token;
    if (token == null) return null;
    return User().where('api_token', token).getFirst();
  }

  @override
  void onConnection(Socket socket) {
    User user = socket.user! as User;
    socket.join('user:${user.id}');
  }
}

The client sends the token in auth:

const socket = io('http://localhost:3000/chat', {
  transports: ['websocket'],
  auth: { token: 'the-user-token' },
});
socket.on('connect_error', (err) => console.log(err.message)); // 'Unauthorized'

With a user, it is kept in socket.user. It stays on this server. fetchSockets() on other servers does not get it.
With null, the client gets a connect_error with the message Unauthorized, and onConnection does not run. Change the message with SocketAuth(findUser, message: '...').
handshake.token is auth.token from the client, else the Authorization: Bearer header. Browsers can not set headers on a websocket, so use auth.
handshake.cookie('name') reads a cookie and decrypts it with the APP_KEY, like req.cookie('name'). Use decrypt: false for a plain cookie.
await handshake.session() gives the Dox session of the client, from its session cookie. See below.
io.use(SocketAuth(findUser).handle) does the same for the default namespace.

With dox_auth

dox_auth can check a JWT or the web login session for you:

static Future<User?> findUser(Handshake handshake) async {
  // an API client or a mobile app sends a JWT
  String? token = handshake.token;
  if (token != null) return Auth(guard: 'api').userFromToken(token);

  // a browser that logged in on a web page sends the session cookie
  Session session = await handshake.session();
  return Auth(guard: 'web').userFromSession(session);
}

userFromToken runs the same checks as AuthMiddleware: signature, guard, exp and revocation. A bad, expired, revoked or changed token gives null, so the client is refused. With userCacheTtl on, a deleted or banned user can still connect until the cache time ends, unless forgetUser runs.
handshake.session() runs the same cookie checks as SessionMiddleware. A changed cookie, a cookie made with another APP_KEY or an expired session gives an empty session, so userFromSession gives null.
handshake.session() uses AppConfig.session. Pass another SessionConfig like handshake.session(config).
The session is read once, when the socket connects. A logout later does not close the socket. Call socket.disconnect() yourself if you need that.
The session is read only. Changes to it are not saved.
With no allowedOrigins, or with '*' in it, handshake.session() prints a warning once.
With the cookie session driver, an old copy of the session cookie still works after logout, until the session lifetime ends. If a logout must stop old cookies, use the cache session driver, like SessionConfig(driver: 'cache', store: 'redis').

Cookie login needs allowedOrigins

A browser sends its cookies on a websocket from any website. So if you log users in with a cookie, set allowedOrigins to your own sites. Without it, another website can open a socket as your user. See Allowed origins.

Node.js to Dart

Node.js	Dart
`socket.on('e', (a, cb) => ...)`	`socket.on('e', (SocketEvent e) => ...)`, then `e.data`, `e.args`, `e.ack(x)`
`socket.emit('e', a, b)`	`socket.emit('e', a)` or `socket.emitArgs('e', [a, b])`
`await socket.timeout(5000).emitWithAck('e', a)`	`await socket.emitWithAck('e', a, timeout: Duration(seconds: 5))`
`socket.join(r)` / `socket.leave(r)`	the same
`socket.to(r).emit(...)`	the same
`socket.broadcast.emit(...)`	the same
`io.to(r).except(x).emit(...)`	the same
`io.of('/chat')`, `io.use(fn)`	the same
`socket.handshake`	`.headers`, `.query`, `.auth`, `.address`, `.token`, `.cookie(name)`, `.session()`
`socket.data`	`socket.data` (a `Map`)
`await io.in(r).fetchSockets()`	`await io.to(r).fetchSockets()` (all servers)
`await io.timeout(5000).emitWithAck('e', a)`	`await io.timeout(Duration(seconds: 5)).emitWithAck('e', a)`
`io.serverSideEmit('e', a)`	`await io.serverSideEmit('e', [a])`
`io.on('e', (a, cb) => ...)` (from other servers)	`io.onServerSideEmit('e', (ServerSideEvent e) => ...)`, then `e.data`, `e.ack(x)`

disconnecting and disconnect give the reason in event.data, like transport close or ping timeout.
Binary data: send a Uint8List. It can be anywhere in lists and maps.

Options

SocketIoServer(
  options: ServerOptions(
    path: '/socket.io/',
    pingInterval: Duration(seconds: 25),
    pingTimeout: Duration(seconds: 20),
    maxHttpBufferSize: 1000000,
    transports: <String>['polling', 'websocket'],
    cors: CorsOptions(origins: <String>['https://my.app'], credentials: true),
  ),
);

Option	Default	What it does
`path`	`/socket.io/`	the URL path
`pingInterval`	25 s	how often the server sends a ping
`pingTimeout`	20 s	how long to wait for the pong before the client is gone
`upgradeTimeout`	10 s	how long an upgrade from polling to websocket may take
`connectTimeout`	45 s	how long the client has to join a namespace
`maxHttpBufferSize`	1,000,000 bytes	the biggest message. A bigger websocket message closes the connection
`maxAttachments`	`10`	the most binary parts in one packet. A packet with more closes that client
`transports`	`['polling', 'websocket']`	allowed transports
`allowUpgrades`	`true`	let polling clients move to websocket
`cors`	`null`	CORS headers for polling requests. It also checks the origin of websockets
`allowedOrigins`	`null`	the sites that may connect, see Allowed origins

cors with a list of origins: a websocket from another origin gets 403. A request with no Origin header (not a browser) is allowed.
cors: null, or * in origins: websockets from any origin are allowed. If you log users in with cookies, list your origins.
* together with credentials: true throws an ArgumentError. List your origins instead.
Websocket compression is off. A message bigger than maxHttpBufferSize closes with code 1009.
On a plain dart:io server (io.attach), the whole message is read before the size check. Also set a size limit in your proxy.

Allowed origins

allowedOrigins says which websites may open a socket. Set it when you log users in with cookies.

ServerOptions(allowedOrigins: <String>['https://app.example.com']);

null (the default): no extra check. Websockets are checked against cors when it is set. Dox logs a warning at start.
[]: only pages of the same host can connect.
A list of sites: the same host and those sites can connect.
['*']: every site can connect.
Other sites get 403. A client with no Origin header (not a browser) is always allowed.

Many servers: Redis adapter

With more than one server process, or next to Node.js servers, use the Redis adapter. Then a broadcast on one server reaches the clients of all servers, and of all their isolates.

Add the ioredis package:

dart pub add ioredis

In a Dox app, make the adapter in each isolate with a factory:

lib/config/socketio.dart

import 'package:dox_socketio/dox_socketio.dart';
import 'package:ioredis/ioredis.dart';
import 'package:my_app/app/socket/chat.namespace.dart';
import 'package:my_app/config/env.dart';

SocketIoConfig socketIoConfig = SocketIoConfig(
  enabled: env.socketioEnabled,
  namespaces: <NamespaceHandler>[ChatNamespace()],
  adapter: const RedisSocketAdapterFactory(),
);

/// makes the Redis adapter inside each isolate.
/// A Redis connection can not be sent to another isolate.
class RedisSocketAdapterFactory extends SocketAdapterFactory {
  const RedisSocketAdapterFactory();

  @override
  Adapter create() {
    Redis redis = Redis(); // 127.0.0.1:6379. Or Redis(RedisOptions(host: ..., port: ...))
    return RedisAdapter(subscriber: redis.duplicate(), publisher: redis);
  }
}

Without Dox, pass the adapter to the server:

Redis redis = Redis(RedisOptions(host: '127.0.0.1', port: 6379));
SocketIoServer io = SocketIoServer(
  adapter: RedisAdapter(subscriber: redis.duplicate(), publisher: redis),
);

The subscriber needs its own connection. A Redis connection in subscribe mode can not run other commands.
It uses the format of @socket.io/redis-adapter. Node.js servers with that adapter can share one Redis with Dox servers.
It reads messages in msgpack and in JSON. It writes msgpack, like the Node.js adapter. If your Node.js servers use a JSON parser, set format: RedisFormat.json.
fetchSockets, socketsJoin, socketsLeave and disconnectSockets act on the sockets of all servers.
Broadcasts with an ack get the acks of clients on all servers.
Servers can send events to each other with serverSideEmit. Clients do not get them.
A request waits at most requestsTimeout for the other servers (5 seconds by default, like Node.js).
Long-polling needs sticky sessions, the same as with many isolates. Or use transports: ['websocket'] on the client.

Act on all servers

// sockets of every server. Sockets of this server are `Socket`s.
List<RemoteSocket> sockets = await io.to('room1').fetchSockets();
for (RemoteSocket socket in sockets) {
  socket.emit('hello', socket.data['name']);
}

await io.to('room1').socketsJoin('room2');
await io.to('banned').disconnectSockets();

// one ack per client, from all servers. A timeout is needed.
List<Object?> answers = await io
    .timeout(const Duration(seconds: 5))
    .emitWithAck('ping', 'are you there?');

fetchSockets throws a TimeoutException when a server does not answer in time.

Events between servers

serverSideEmit sends an event to the other servers. Clients do not get it.

// on every server
io.onServerSideEmit('reload-config', (ServerSideEvent event) {
  reloadConfig(event.data);
  event.ack('done'); // only when the sender waits for an answer
});

// on one server
await io.serverSideEmit('reload-config', <Object?>['v2']);

// wait for one answer from each other server
List<Object?> answers = await io.serverSideEmitWithAck('reload-config', <Object?>['v2']);

When a server does not answer in time, serverSideEmitWithAck throws an AckTimeoutException. Its responses holds the answers that came back. onServerSideEmit is also on namespaces: io.of('/chat').onServerSideEmit(...).

Emit from other programs

An emitter sends events to your clients without a Socket.IO server. Use it in a queue worker, a cron job or a service in another language. It writes to Redis, and every server with the Redis adapter sends the event to its clients.

Dart (for example in a Dox queue job or a CLI command):

import 'package:dox_socketio/dox_socketio.dart';
import 'package:ioredis/ioredis.dart';

Future<void> main() async {
  Redis redis = Redis(RedisOptions(host: '127.0.0.1', port: 6379));
  SocketIoEmitter emitter = SocketIoEmitter(redis);

  await emitter.emit('news', <String, dynamic>{'title': 'hello'}); // all clients of `/`
  await emitter.to('room1').except('room2').emit('news', 'hi room1');
  await emitter.of('/chat').to('vip').emit('notice', 'for vip');

  await emitter.to('room1').socketsJoin('room3'); // sockets in room1 join room3
  await emitter.to('room3').socketsLeave('room1');
  await emitter.to('banned').disconnectSockets();

  await redis.disconnect();
}

Node.js, with @socket.io/redis-emitter:

import { Emitter } from '@socket.io/redis-emitter';
import { createClient } from 'redis';

const redis = createClient({ url: 'redis://127.0.0.1:6379' });
await redis.connect();

const emitter = new Emitter(redis);
emitter.to('room1').emit('news', { title: 'hello' });
emitter.of('/chat').emit('notice', 'for everyone in /chat');

The emitters of other languages (PHP, Python, Go) use the same format, so they work too.
Binary data (Uint8List in Dart, Buffer in Node.js) arrives as binary.
The emitter does not wait for an answer. It can not know if a client got the event.
key must be the same as the key of the adapter. Both default to socket.io.