socketcluster Documentation

1. Project Overview & Quickstart (SocketCluster/socketcluster) Copy Chapter

File: README.md Copy Chapter

SocketCluster

Toolset and boilerplate for quickly creating systems using SocketCluster.
See the client and server repos for documentation:

• https://github.com/SocketCluster/socketcluster-client
• https://github.com/SocketCluster/socketcluster-server

Documentation for SCC (horizontally scalable cluster) is available at https://github.com/SocketCluster/socketcluster/blob/master/scc-guide.md

Installation Copy Chapter

First, set up Node.js on your machine.

Then set up the socketcluster command:

npm install -g socketcluster

or:

sudo npm install -g socketcluster

then:

socketcluster create myapp

Once it's installed, go to your new myapp/ directory and launch with:

node server

Access at URL http://localhost:8000/

Node.js v10.0.0 or above is recommended but you can also use SocketCluster with older Node.js versions if you use while loops instead of for-await-of loops.

Compatibility mode Copy Chapter

For compatibility with existing SocketCluster clients, set the protocolVersion to 1 and make sure that the path matches your old client path:

let agServer = socketClusterServer.attach(httpServer, {
  protocolVersion: 1,
  path: '/socketcluster/'
});

Change log Copy Chapter

See the 'releases' section for changes: https://github.com/SocketCluster/socketcluster/releases

License Copy Chapter

(The MIT License)

Copyright (c) 2013-2023 SocketCluster.io

Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the 'Software'), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED 'AS IS', WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

File: package.json Copy Chapter

{
"name": "socketcluster",
"version": "21.0.1",
"description": "Highly scalable realtime framework with support for async/await",
"scripts": {
"test": "echo "Error: no test specified" && exit 1"
},
"repository": {
"type": "git",
"url": "git+https://github.com/SocketCluster/socketcluster.git"
},
"keywords": [
"socketcluster",
"async",
"await",
"realtime"
],
"author": "Jonathan Gros-Dubois",
"license": "MIT",
"bugs": {
"url": "https://github.com/SocketCluster/socketcluster/issues"
},
"homepage": "https://socketcluster.io/",
"dependencies": {
"@maartennnn/cli-builder": "^2.3.0",
"fs-extra": "^7.0.1",
"yaml": "^2.9.1"
},
"bin": {
"socketcluster": "bin/cli.js"
}
}

File: scc-guide.md Copy Chapter

SCC Guide

SCC is a collection of services which allow you to easily deploy and scale SocketCluster to any number of machines.
SCC is designed to scale linearly and is optimized for running on Kubernetes but it can be setup without using an orchestrator.

SCC is made up of the following services:

• scc-worker (SocketCluster) https://github.com/SocketCluster/socketcluster
• scc-broker https://github.com/SocketCluster/scc-broker
• scc-state https://github.com/SocketCluster/scc-state
• scc-ingress (Kubernetes only)

How it works Copy Chapter

The scc-worker service can be made up of any number of regular SocketCluster instances - The main difference between running SocketCluster as a single instance vs running it as a cluster is that in cluster mode, you need to point each scc-worker instance to a working scc-state (state server) instance.

The scc-broker service can be made up of any number of scc-broker instances - This is a special backend-only service which is designed to broker
messages between multiple frontend-facing scc-worker instances. All the pub/sub channels in your entire system will be sharded evenly across available scc-broker instances.
Just like with the scc-worker instances above, each scc-broker instance needs to point to a state server in order to work.

The scc-state service is made up of a single instance - Its job is to dispatch the state of the cluster to all interested services to allow them to reshard themselves. The scc-state instance will notify all frontend scc-worker instances whenever a new backend scc-broker joins the cluster. This allows scc-worker instances to rebalance their pub/sub channels evenly across available brokers whenever a new scc-broker instance joins the cluster.
Note that SCC can continue to operate without any disruption of service while the scc-state instance is down/unavailable (see notes at the bottom of this page).

Running on Kubernetes (recommended) Copy Chapter

Running on Kubernetes (K8s) is easy; you just need to run all the .yaml files from the kubernetes/ directory from the SocketCluster repo (https://github.com/SocketCluster/socketcluster/tree/master/app/kubernetes) using the kubectl command (one at a time):

kubectl create -f <service-deployment-or-ingress-definition.yaml>

By default, you should also add a TLS/SSL key and cert pair to your provider (Rancher has a page were you can just paste them in).
Or if you don't want to use a certificate (not recommended), you can just delete these lines from scc-ingress.yaml before you create it with kubectl:

tls:
  - secretName: scc-tls-credentials

Note that the step above is crucial if you don't want to use TLS/SSL - Otherwise the ingress load balancer service will not show up on your Rancher control panel until you add some credentials with the name scc-tls-credentials to your Rancher control panel (See Infrastructure > Certificates page).

Running using Node.js directly Copy Chapter

You can also run SCC using only Node.js version >= 10.x.x.
For simplicity, we will show you how to run everything on your localhost (127.0.0.1), but in practice, you will need to change 127.0.0.1 to an appropriate IP, host name or domain name.

First, you need to download each of these repositories to your machine(s):

• git clone https://github.com/SocketCluster/scc-broker
• git clone https://github.com/SocketCluster/scc-state

Then inside each repo, you should run npm install without any arguments to install dependencies.

Then you need to setup a new SocketCluster project to use as your user-facing instance.

Once you have the two repos mentioned earlier and your SocketCluster project setup, you should launch the state server first by
going inside your local scc-state repo and then running the command:

node server

Next, to launch a broker, you should navigate to your scc-broker repo and run the command:

SCC_STATE_SERVER_HOST='127.0.0.1' SCC_BROKER_SERVER_PORT='8888' node server

Finally, to run a frontend-facing SocketCluster instance, you can navigate to your project directory and run:

SCC_STATE_SERVER_HOST='127.0.0.1' SOCKETCLUSTER_PORT='8000' node server

You can add a second frontend-facing server by running (this time running on port 8001):

SCC_STATE_SERVER_HOST='127.0.0.1' SOCKETCLUSTER_PORT='8001' node server

Now if you navigate to either localhost:8000 or localhost:8001 in your browser, you should see that your pub/sub channels are shared between the two scc-worker instances.

Note that you can provide additional environment variables to various instances to set custom port numbers, passwords etc...
For more info, you can look inside the code in the server.js file in each repo and see what process.env vars are used.

When running multiples instances of any service on the same machine, make sure that the ports don't clash - Modify the SCC_BROKER_SERVER_PORT or SOCKETCLUSTER_PORT environment variable for each instance to make sure that they are unique.

CAP theorem Copy Chapter

User-facing instances in SCC are highly available (this was done to ensure that SCC is resilient against DDoS attacks).
Back end instances, on the other hand, are optimized for consistency and efficiency. SCC is designed to quickly recover from failure of back end instances (a few seconds of partial missed messages at worst). Note that it's still possible to implement guaranteed delivery of messages by assigning a unique ID to each published message and resending messages which have not been acknowledged.

Notes Copy Chapter

You should only ever run a single scc-state per cluster - Note that the cluster can continue to operate without any disruption while scc-state is down.
scc-state only needs to be available for a few seconds while SCC is in the process of scaling itself up or down. Even if scc-state crashes while in the middle of scaling up, SCC will wait for scc-state to become available again (still without disruptions to the existing service) and will resume the scaling operation as soon as the scc-state instance becomes available again.
In the event of a crash, K8s will respawn scc-state within a few seconds so a failure of scc-state will only delay your scale up/down operation at worst.
Nevertheless, it is recommended that you run the scc-state instance inside your datacenter/AWS availability zone and do not expose it to the public internet.

The scc-state instance does not handle any pub/sub messages and so it is not a bottleneck with regards to the scalability of your cluster (SCC scales linearly).

Note that you can launch the services in any order you like but if your state server is not available, you may get harmless Socket hung up warnings on other instances (while they keep trying to reconnect) until scc-state becomes available again.

File: SECURITY.md Copy Chapter

Security Policy

Reporting a Vulnerability Copy Chapter

Please report security issues to [email protected]

File: socketcluster-protocol.md Copy Chapter

SocketCluster Protocol Overview Copy Chapter

SocketCluster protocol is implemented on top of the WebSockets protocol and consists of multiple components:

• Handshake
• Connection health check (ping/pong)
• Event layer
• Pub/Sub layer
• Authentication layer

Minimal requirements for a simple SocketCluster compatible client are to implement:
Handshake, ping/pong and (at least partially) the Event layer.

Pub/Sub and Authentication layers are completely optional.

Contrast between Protocol V1 and V2 Copy Section

• SocketCluster <=v14 uses Protocol v1.
  SocketCluster >=v15 uses Protocol v2 by default and supports protocolVersion configuration option, which allows it to work with Protocol v1.

• SocketCluster <=v14 doesn't send back Handshake event response, if cid is not specified in Handshake event.
  SocketCluster >=v15 always sends back Handshake event response, regardles of cid presence. If cid is not present in Handshake event, rid is omitted from Handshake event response.

• In SocketCluster >=v15 #disconnect event is deprecated and no longer in use.

• Protocol V1 uses '#1' and '#2' for ping/pong
  Protocol V2 uses empty strings '' for both.

• In Protocol V1 all event names starting with symbol # are considered reserved for special control events.
  In Protocol V2 only a handful of event names starting with symbol # are considered reserved.

Reserved event names Copy Section

Protocol V1:

• All event names starting with symbol #

Protocol V2:

• #handshake
• #publish
• #subscribe
• #unsubscribe
• #kickOut
• #authenticate
• #setAuthToken
• #removeAuthToken

Call ID & Response ID Copy Section

cid - Call ID
rid - Response ID

Some events require acknowledgement from another side of communication, in other words they expect event responses.
In order to track which event responses belong to which events, cid and rid exist in SocketCluster Protocol.

cid must be unique for each event sent, during the whole socket connection lifetime.
Call IDs originated from server and Call IDs originated from client are two different sets of ids and are being appointed and tracked separately.

cid, included in events sent from socketcluster-server to a SocketCluster client, for each new socket connection, will always start with number 1 and will be incremented with each event sent.

In your custom SocketCluster client you could use something like UUID strings for cid but, for efficiency, it's recommended to also use number 1 and increment it with each subsequent event sent.

Some special events expect no response, hence cid for them is not required and ignored if present.

Handshake Copy Chapter

As soon as you establish a WebSocket connection, you are to send a special Handshake event for socketcluster-server to initiate the socket.
Clients are not allowed to interact with the server before Handshake.

**Handshake event** is a JSON-encoded string with the following structure:

{
  event: '#handshake'

  // [optional] A JSON-compatible Object
  data: { },

  // [optional] Call ID
  cid: 1
}

**Handshake event response** is a JSON-encoded string with the following structure:

{
  data: {
    // A unique ID, assigned to this socket connection by the server
    id: 'Y7gRvz-hVW_uXx5qAAH',

    // Value of `pingTimeout` configuration option of the server
    pingTimeout: 20000, // ms

    // Look at the Authentication layer overview for more information
    isAuthenticated: false
  }

  // [optional] Response ID
  rid: 1,
}

socketcluster-server <=v14:

• If cid was specified in the Handshake event, socketcluster-server will send back Handshake event response with matching rid.
  If cid was not specified, no Handshake event response will be sent.

socketcluster-server >=v15:

• Whether or not cid was specified in the Handshake event, socketcluster-server will always send back Handshake event response.
  If cid was specified in the Handshake event, Handshake event response will include matching rid.
  If cid was not specified, rid will be omitted.

Connection health check (ping/pong) Copy Chapter

socketcluster-server periodically sends ping messages to connected clients to check whether or not a connection is still alive.
A SocketCluster client has to answer every ping message with pong message as soon as possible.

**Protocol V1:**

Ping message (from server) is a String: '#1'
Pong message (from client) is a String: '#2'

**Protocol V2:**

Ping message (from server) is an empty String: ''
Pong message (from client) is an empty String: ''

Ping/pong mechanism is required to account for cases when a connection might be closed without sending a proper SocketCluster #disconnect* event or WebSockets Close control frame.
For example, if a user's internet drops out suddenly, there would be no way to tell that the socket is no longer connected otherwise.

* - In Protocol V2 #disconnect event is deprecated and no longer in use.

Event layer Copy Chapter

Event layer is responsible for one-to-one communication between a particular socket connection and the server.

Basic part Copy Section

Basic part of the Event layer is responsible for transmitting and receiving user-defined events.
API example from JavaScript socketcluster-client v17:

// Transmit an event to the server
socket.transmit('eventName', data)

// Receive events from the server
for await (const data of socket.receiver('eventName')) {
  console.info('received data', data)
}

For more in depth knowledge on API visit https://socketcluster.io/docs/basic-usage

**Transmitted event** is a JSON-encoded string with the following structure:

{
  // Arbitrary name of the event*
  event: 'eventName',

  // [optional] Any JSON-compatible data
  data: eventData,
}

* - Some event names starting with '#' are reserved for special control events in SocketCluster Protocol.

Transmitted events never expect responses.
Even if action.TRANSMIT was blocked within agServer.MIDDLEWARE_INBOUND, no response will be sent.

Advanced part Copy Section

Advanced part of the Event layer is responsible for invoking and processing Remote Procedure Calls.
API example from JavaScript socketcluster-client v17:

// Invoke a remote procedure on the server
const responseData = await socket.invoke('procedureName', data)

// Process remote procedure calls from the server
for await (const request of socket.procedure('procedureName')) { 
  console.info('received data', request.data)
  request.end(dataToReturnToServer)
}

Transmitted events and RPC are similar in structure. They share the same property event for their names, but they are different entities.
In order to invoke a RPC, a SocketCluster client should send a RPC request.
Unlike transmitted events, every RPC request must include a unique cid (Call ID), because every RPC request expects a RPC response with matching rid (Response ID) from another side of communication.

**RPC request** is a JSON-encoded string with the following structure:

{
  // Call ID
  cid: 12345,

  // Arbitrary name of the procedure*
  event: 'procedureName',

  // [optional] Any JSON-compatible data
  data: procedureData
}

* - Some procedure names starting with '#' are reserved for special control events in SocketCluster Protocol.

When the RPC request is processed, a RPC response should be sent back.

**Successful RPC response** is a JSON-encoded string with the following structure:

{
  // Response ID
  rid: 12345,

  // [optional] Any JSON-compatible data
  data: responseData
}

If the RPC request was blocked within agServer.MIDDLEWARE_INBOUND, then the argument, which was provided to the action.block(err) method, will be included into the RPC response as error property.
If no argument was provided to the action.block() method, the error will contain default SocketCluster SilentMiddlewareBlockedError:

**Unsuccessful RPC response** is a JSON-encoded string with the following structure:

{
  rid: 12345,

  error: {
    message: 'The invoke AGAction was blocked by inbound middleware',
    name: 'SilentMiddlewareBlockedError',
    type: 'inbound'
  }
}

Every RPC response must include rid exactly matching cid of the respective RPC.
As an example, let's invoke a RPC from the server side.
API example from socketcluster-server v17:

try {
  const responseData = await socket.invoke('procedureName', data)
} catch (err) {
  if (err.name === 'TimeoutError') {
    // ...
  }
}

If no response with matching rid will be received from a client, the socket.invoke method will throw TimeoutError, after time interval specified in ackTimeout configuration option of the server.

Most of the SocketCluster clients follow the same logic.
A SocketCluster client sets a timer (alike setTimeout) for each RPC sent, with consideration of cid. Those timers expose a TimeoutError when are finished. And if the client receives a RPC response with rid matching cid of one of the ongoing timers, the client destroys the timer before it fires up.

Pub/Sub layer Copy Chapter

Pub/Sub layer is responsible for one-to-many communication between a particular socket connection or a particular SocketCluster worker and unlimited amount of connected sockets, which are subscribed to a Pub/Sub channel.

Subscribe Copy Section

In order to subscribe a socket connection to a Pub/Sub channel, a SocketCluster client should send to server a subscription request.
API example from JavaScript socketcluster-client v17:

const channel = socket.subscribe('channelName')
for await (const message of channel) {
  console.info(message)
}

For more in depth knowledge on API visit https://socketcluster.io/docs/basic-usage

**Subscription request** is a JSON-encoded string with the following structure:

{
  event: '#subscribe',

  data: {
    // Arbitrary name of a Pub/Sub channel to subscribe to
    channel: 'channelName'
  },

  cid: 12345
}

When subscription request will be processed, socketcluster-server will send back subscription response with matching rid. So the client would know it's successfully subscribed to the channel.

**Successful subscription response** is a JSON-encoded string with the following structure:

{
  rid: 12345
}

If the subscription request was blocked within agServer.MIDDLEWARE_INBOUND, then the argument, which was provided to the action.block(err) method, will be included into the subscription response as error property.
If no argument was provided to the action.block() method, the error will contain default SocketCluster SilentMiddlewareBlockedError:

**Unsuccessful subscription response** is a JSON-encoded string with the following structure:

{
  rid: 12345,

  error: {
    message: 'The subscribe AGAction was blocked by inbound middleware',
    name: 'SilentMiddlewareBlockedError',
    type: 'inbound'
  }
}

Publish Copy Section

In order to publish a message to a Pub/Sub channel, a SocketCluster client should send to server a publish request.
API example from JavaScript socketcluster-client v17:

// transmitPublish includes `cid`
socket.transmitPublish('channelName', messageData)

// invokePublish does not include `cid`
const responseData = await socket.invokePublish('channelName', messageData)

**Publish request** is a JSON-encoded string with the following structure:

{
  event: '#publish',

  data: {
    // Name of the channel to publish message to
    channel: 'channelName',
    // [optional] Any JSON-compatible data
    data: messageData
  },

  // [optional] Call ID
  cid: 12345
}

If cid was specified, socketcluster-server will send back publish response with matching rid. So the client would know the sent message was successfully published to the channel.

**Successful publish response** is a JSON-encoded string with the following structure:

{
  rid: 12345
}

If cid was specified and the publish request was blocked within agServer.MIDDLEWARE_INBOUND, then the argument, which was provided to the action.block(err) method, will be included into the publish response as error property.
If no argument was provided to the action.block() method, the error will contain default SocketCluster SilentMiddlewareBlockedError:

**Unsuccessful publish response** is a JSON-encoded string with the following structure:

{
  rid: 12345,

  error: {
    message: 'The publishIn AGAction was blocked by inbound middleware',
    name: 'SilentMiddlewareBlockedError',
    type: 'inbound'
  }
}

If cid was not specified in the publish request, no publish response will be sent even if the request was unsuccessful.

Unsubscribe Copy Section

In order to unsubscribe a socket connection from a Pub/Sub channel, a SocketCluster client should send to server an unsubscription event.

**Unsubscription event** is a JSON-encoded string with the following structure:

{
  event: '#unsubscribe',

  // Name of a Pub/Sub channel to unsubscribe from
  data: 'channelName',

  // [optional] Call ID
  cid: 12345
}

When unsubscription event will be processed, socketcluster-server will send back unsubscription event response with matching rid. So the client would know it's successfully unsubscribed from the channel.

**Unsubscription event response** is a JSON-encoded string with the following structure:

{
  rid: 12345
}

If no cid was specified in the unsubscription event, no unsubscription event response will be sent back.

Kick out Copy Section

It's possible, from server side, to forcibly unsubscribe a socket connection from one or more particular Pub/Sub channels or from all Pub/Sub channels at once.
API example from socketcluster-server v17:

socket.kickOut(['channelName', 'channelName2'], 'custom message')

In that case a SocketCluster client will receive a special #kickOut event. Or multiple events, if it was kicked from multiple channels. One per each channel it was kicked from.

**kickOut event** is a JSON-encoded string with the following structure:

{
  event: '#kickOut',

  data: {
    // Name of the Pub/Sub channel the socket connection was kicked from
    channel: 'channelName',

    // [optional] If a message has been provided to the socket.kickOut method
    message: 'custom message'
  }
}

Authentication layer Copy Chapter

Authentication layer is responsible for means of acquiring and storing an authentication token on client side, as well as for transfering previously acquired authentication token to server for processing.

Authentication process in SocketCluster is deeply customizable. You could implement and use in your SocketCluster client virtually any authentication strategy for your application.
Here let's review the default SocketCluster authentication strategy, which uses JWT as the authentication token and transfers it to server via WebSockets connection within Handshake event.

Token acquisition Copy Section

For more in depth knowledge on authentication process visit https://socketcluster.io/docs/authentication
API example from socketcluster-server v17:

socket.setAuthToken({username: 'Alice', channels: []})

When setAuthToken method is called on the server side, a SocketCluster client will receive a JSON-encoded string with the following structure:

{
  event: '#setAuthToken',

  data: {
    token: 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VybmFtZSI6IkFsaWNlIiwiY2hhbm5lbHMiOltdLCJUaGFua1lvdUZvckNob29zaW5nU29ja2V0Q2x1c3RlciI6dHJ1ZSwiaWF0IjoxNjc0NzMxODc3LCJleHAiOjE2NzQ4MTgyNzd9.MzQ0QQzofbtlnzvPbeTgtpcvg8Sh6cY8EwXqNXHj5ns'
  }
}

Receiving #setAuthToken event on the client side means the socket connection is now authenticated. A SocketCluster client is supposed to store the acquired token somewhere locally for later use. If the client was already authenticated, it's supposed to replace previously stored token with the new one.

Transfer token to server for processing Copy Section

In the default authentication strategy this could be done in two ways:

• By including authentication token within Handshake event
• By sending Authentication event

Authentication within Handshake Copy Section

The most practical way is to include previously acquired authentication token within the Handshake event.
This way socketcluster-server will automatically pick up the token and your client will become authenticated right away.

**Handshake event** is a JSON-encoded string with the following structure:

{
  event: '#handshake',

  data: {
    authToken: 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VybmFtZSI6IkFsaWNlIiwiY2hhbm5lbHMiOltdLCJUaGFua1lvdUZvckNob29zaW5nU29ja2V0Q2x1c3RlciI6dHJ1ZSwiaWF0IjoxNjc0NzMxODc3LCJleHAiOjE2NzQ4MTgyNzd9.MzQ0QQzofbtlnzvPbeTgtpcvg8Sh6cY8EwXqNXHj5ns'
  },

  // [optional] Call ID
  cid: 1
}

Depending on success of authentication process, Handshake event response may include different information.
If provided authToken is valid and not expired, a SocketCluster client will receive Handshake event response with no authError property included.

**Handshake event response** is a JSON-encoded string with the following structure:

{
  // [optional] Response ID
  rid: 1,

  data: {
    // A unique ID, assigned to this socket connection by the server
    id: 'ZnY2picItxGyKbzeAAAE',

    // Value of `pingTimeout` configuration option of the server
    pingTimeout: 20000, // ms

    // If true, then `#setAuthToken` event will be received immediately after
    isAuthenticated: true
  }
}

If any errors will occur during authentication process, for example the provided authToken is invalid or expired, or Handshake was blocked in the server middleware, then Handshake event response will include authError property describing the occured error. More about authentication errors

**Handshake event response** is a JSON-encoded string with the following structure:

{
  // [optional] Response ID
  rid: 1,

  data: {
    // A unique ID, assigned to this socket connection by the server
    id: 'Y7Uw-jHCJP_gld4QAAAA',

    // Value of `pingTimeout` configuration option of the server
    pingTimeout: 20000, // ms

    // If false, then `#removeAuthToken` event will be received immediately after
    isAuthenticated: false,

    authError: {
      name: 'ErrorName',
      message: 'error message',
      isBadToken: true
    }
  }
}

Authentication event Copy Section

Instead of including authentication token within Handshake event, a SocketCluster client could send special Authentication event any time after Handshake.
This could be useful if authentication token somehow becomes available in your client after Handshake.
As an example, if you use JavaScript socketcluster-client in a web browser and get multiple tabs opened, you could acquire authentication token in one tab and, when authToken becomes available in the localStorage, other tabs would send to server the authToken within Authentication event, for their socket connections to also become authenticated, without need for page reload.

**Authentication event** is a JSON-encoded string with the following structure:

{
  event: '#authenticate',

  // A String with an authentication token
  data: 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VybmFtZSI6IkFsaWNlIiwiY2hhbm5lbHMiOltdLCJUaGFua1lvdUZvckNob29zaW5nU29ja2V0Q2x1c3RlciI6dHJ1ZSwiaWF0IjoxNjc0NzMxODc3LCJleHAiOjE2NzQ4MTgyNzd9.MzQ0QQzofbtlnzvPbeTgtpcvg8Sh6cY8EwXqNXHj5ns',

  // [optional] Call ID
  cid: 12345
}

Depending on success of authentication process, Authentication event response may include different information.
If authToken is valid and not expired, a SocketCluster client will receive successful Authentication event response, which means socket connection is now authenticated.
If cid was specified in Authentication event, Authentication event response will also include matching rid.

**Successful Authentication event response** is a JSON-encoded string with the following structure:

{
  // [optional] Response ID
  rid: 12345,

  data: {
    isAuthenticated: true,
    authError: null
  }
}

If any errors will occur during authentication process, for example authentication token is invalid or expired, Authentication event response will include error property describing the occured error. More about authentication errors

**Unsuccessful Authentication event response** is a JSON-encoded string with the following structure:

{
  // [optional] Response ID
  rid: 12345,

  // If error is present, then `#removeAuthToken` event will be received immediately after
  error: {
    name: 'ErrorName',
    message: 'error message',
    isBadToken: true
  }
}

Deathentication Copy Section

When you deauthenticate a socket connection from server side, a SocketCluster client will receive special #removeAuthToken event. It means the socket connection is no longer authenticated.
Also Deauthentication event could be received immediately after sending Handshake or Authentication event, if authToken provided within Handshake or Authentication event is invalid.
API example from socketcluster-server v17:

socket.deauthenticate()

**Deauthentication event** is a JSON-encoded string with the following structure:

{
  event: '#removeAuthToken'
}

No response is expected for Deauthentication event.

If a SocketCluster client intends to deauthenticate, it should send to server identical Deauthentication event.

In both cases the client should remove previously stored authentication token, as well as to perform any other operations you deem necessary, like routing client to login screen to start over the token acquisition process

Authentication errors Copy Section

By default authentication errors consist of name, message and isBadToken. But they also could include additional information depending on the error type. Or to have various content in case of user-defined errors.

When you block an authentication action (action.type === action.AUTHENTICATE) within the server middleware (agServer.MIDDLEWARE_INBOUND), the error object you pass to the action.block(err) method becomes the authentication error, which your client will receive.
For more in depth knowledge on API visit https://socketcluster.io/docs/middleware-and-authorization

name:
Could be either one of the SocketCluster error names or a user-defined error name.

All possible SocketCluster errors are listed in SC-errors module.
List of the names which are relevant to the authentication process, in order of frequency occurring are:

• AuthTokenExpiredError
• AuthTokenInvalidError
• AuthTokenError
• AuthTokenNotBeforeError

message:
Could be originated from SocketCluster or from underlying libraries like jsonwebtoken. It also could be a user-defined error message.
List of all the possible error messages is hard to come by, but some of them you could find in JWT Readme

isBadToken:
If an authentication error was caused by an expired or invalid or malformed JWT or by incorrect JWT signature, isBadToken will be true. If the authentication error was caused by any other reason, it will be false.

As an example, the two most frequently occurring errors are:

authError: {
  name: 'AuthTokenExpiredError',
  message: 'jwt expired',
  expiry: '2023-01-01T01:23:45.000Z',
  isBadToken: true
}

authError: {
  name: 'AuthTokenInvalidError',
  message: 'invalid token',
  isBadToken: true
}

Any authentication error will always cause a SocketCluster client to receive #removeAuthToken event immediately after the event response, which contained the error.

Thank you for your time implementing custom SocketCluster client! :heart:
If you have completed one, please, open a pull request to add your client to the list of SocketCluster clients
If you encounter any errors or have any questions, feel free to ask for help in SocketCluster Gitter chat room

File: app/nodemon.json Copy Chapter

{
"ignore": ["public/*"]
}

File: app/package.json Copy Chapter

{
"name": "socketcluster-sample",
"description": "A sample SocketCluster app",
"version": "1.0.1",
"type": "module",
"contributors": [
{
"name": "Jonathan Gros-Dubois",
"email": "[email protected]"
}
],
"dependencies": {
"eetase": "^7.1.0",
"express": "^4.16.3",
"morgan": "^1.7.0",
"nodemon": "^3.1.10",
"scc-broker-client": "^11.0.0",
"serve-static": "^1.13.2",
"socketcluster-client": "^21.0.0",
"socketcluster-server": "^21.0.2",
"uuid": "^14.0.2"
},
"keywords": [
"websocket",
"server",
"realtime",
"cluster",
"scalable"
],
"readmeFilename": "README.md",
"scripts": {
"start": "node server.js",
"start:watch": "./node_modules/nodemon/bin/nodemon.js server.js",
"start:docker": "./node_modules/nodemon/bin/nodemon.js /usr/src/app/server.js"
}
}

File: app/kubernetes/scc-broker-deployment.yaml Copy Chapter

apiVersion: apps/v1
kind: Deployment
metadata:
name: scc-broker
spec:
replicas: 1
selector:
matchLabels:
component: scc-broker
template:
metadata:
labels:
component: scc-broker
spec:
containers:
-
name: scc-broker
image: 'socketcluster/scc-broker:v11.0.0'
ports:
-
name: scc-broker
containerPort: 8888
env:
-
name: SCC_STATE_SERVER_HOST
value: scc-state
-
name: SCC_INSTANCE_IP
valueFrom:
fieldRef:
fieldPath: status.podIP
-
name: SCC_BROKER_SERVER_LOG_LEVEL
value: '2'
readinessProbe:
httpGet:
path: /health-check
port: scc-broker
initialDelaySeconds: 5
timeoutSeconds: 10
livenessProbe:
httpGet:
path: /health-check
port: scc-broker
initialDelaySeconds: 15
timeoutSeconds: 20

File: app/kubernetes/scc-broker-service.yaml Copy Chapter

apiVersion: v1
kind: Service
metadata:
name: scc-broker
spec:
type: ClusterIP
ports:
-
port: 8888
targetPort: 8888
selector:
component: scc-broker

File: app/kubernetes/scc-ingress.yaml Copy Chapter

apiVersion: extensions/v1beta1
kind: Ingress
metadata:
name: scc-ingress
annotations:
kubernetes.io/ingress.allow-http: 'false'
spec:

You should upload a certificate to your Kubernetes platform and refer to it here by secretName.

If you do not want to serve SCC over TLS, you can remove or comment out the

kubernetes.io/ingress.allow-http annotation above and the tls section below.

Note that raw (unencrypted) WebSockets may not work if you are behind a corporate proxy so

it is advised that you provide a cert for production and access the SCC service via https:// and wss://

tls:
-
secretName: scc-tls-credentials
backend:
serviceName: scc-worker
servicePort: 8000

File: app/kubernetes/scc-state-deployment.yaml Copy Chapter

apiVersion: apps/v1
kind: Deployment
metadata:
name: scc-state
spec:
replicas: 1
selector:
matchLabels:
component: scc-state
template:
metadata:
labels:
component: scc-state
spec:
containers:
-
name: scc-state
image: 'socketcluster/scc-state:v11.0.0'
ports:
-
name: scc-state
containerPort: 7777
readinessProbe:
httpGet:
path: /health-check
port: scc-state
initialDelaySeconds: 5
timeoutSeconds: 10
livenessProbe:
httpGet:
path: /health-check
port: scc-state
initialDelaySeconds: 15
timeoutSeconds: 20

File: app/kubernetes/scc-state-service.yaml Copy Chapter

apiVersion: v1
kind: Service
metadata:
name: scc-state
spec:
type: ClusterIP
ports:
-
port: 7777
targetPort: 7777
selector:
component: scc-state

File: app/kubernetes/scc-worker-deployment.yaml Copy Chapter

apiVersion: apps/v1
kind: Deployment
metadata:
name: scc-worker
spec:
replicas: 1
selector:
matchLabels:
component: scc-worker
template:
metadata:
labels:
component: scc-worker
spec:
containers:
-
name: scc-worker
image: 'socketcluster/socketcluster:v21.0.0'
ports:
-
name: scc-worker
containerPort: 8000
env:
-
name: SCC_STATE_SERVER_HOST
value: scc-state
-
name: SCC_INSTANCE_IP
valueFrom:
fieldRef:
fieldPath: status.podIP
-
name: ENV
value: prod
readinessProbe:
httpGet:
path: /health-check
port: scc-worker
initialDelaySeconds: 5
timeoutSeconds: 10
livenessProbe:
httpGet:
path: /health-check
port: scc-worker
initialDelaySeconds: 15
timeoutSeconds: 20

File: app/kubernetes/scc-worker-service.yaml Copy Chapter

apiVersion: v1
kind: Service
metadata:
name: scc-worker
spec:
type: NodePort
ports:
-
port: 8000
targetPort: 8000
selector:
component: scc-worker

File: .github/FUNDING.yml Copy Chapter

These are supported funding model platforms

github: jondubois

2. Official Technical Reference & Guides (SocketCluster/socketcluster-website) Copy Chapter

File: docs/api.md Copy Chapter

id: api title: API Reference sidebar_label: API Copy Chapter

Core API Copy Section

• socketClusterClient
• AGClientSocket
• AGChannel
• AGServer
• AGServerSocket
• AGExchange
• AGAction

Stream API Copy Section

• ConsumableStream
• WritableConsumableStream
• AsyncStreamEmitter

File: docs/api-ag-action.md Copy Chapter

id: api-ag-action title: AGAction sidebar_label: AGAction Copy Chapter

Properties Copy Chapter

Methods Copy Chapter

File: docs/api-ag-channel.md Copy Chapter

id: api-ag-channel title: AGChannel sidebar_label: AGChannel Copy Chapter

Inherits from Copy Chapter

WritableConsumableStream

Properties Copy Chapter

`waitForAuth`: A boolean value which indicates whether or not this channel will wait for socket authentication before subscribing to the server. It will remain in the 'pending' state until the socket becomes authenticated.
    `priority`: In case of a resubscribe (e.g. after recovering from a lost connection), this value determines the order in which channel subscriptions will be processed. Higher priority channels will be processed first. This value may be undefined.
    `data`: Data which was passed along with the the channel subscription (if specified). This value can be accessed by the `SUBSCRIBE` action on the `MIDDLEWARE_INBOUND` middleware on the server side as part of the channel subscription. |

| SUBSCRIBED | A string constant which is used to indicate that this channel is in a subscribed state. See the state property above. |
| PENDING | A string constant which is used to indicate that this channel is in a pending state. See the state property above. |
| UNSUBSCRIBED | A string constant which is used to indicate that this channel is in an unsubscribed state. See the state property above. |

Events Copy Chapter

Methods Copy Chapter

The data argument can be any JSON-compatible object/array or primitive. |

| invokePublish(data) | Publish data to this channel. Expect a response from the server.
The data argument can be any JSON-compatible object/array or primitive.
This method returns a Promise which will be rejected if the operation fails.
For example, it can be rejected if the MIDDLEWARE_INBOUND middleware blocks the action on the server side.
The promise will resolve once the server has processed the publish action. |
| listener(eventName) | This method returns an event listener stream for the specified eventName. This object is an asyncIterable which can be consumed with a for-await-of loop. |
| closeListener(eventName) | This method will signal to all consuming for-await-of loops (for the eventName listener) to break after they have finished iterating over their current backlogs of events. |
| closeAllListeners() | This method will signal to all consuming for-await-of loops for all listeners to break after they have finished consuming their respective backlogs of events. |
| killListener(eventName) | This method will signal to all consuming for-await-of loops for the eventName listener to break immediately and will reset the backpressure for that listener to 0. |
| killAllListeners() | This method will signal to all consuming for-await-of loops for all listeners to break immediately and will reset the aggregate backpressure for all listeners to 0. |
| closeOutput() | This method will signal to all consuming for-await-of loops for the channel's output stream to break after they have finished consuming their backlogs of data. |
| killOutput() | This method will signal to all consuming for-await-of loops for the channel's output stream to break immediately and will reset the aggregate backpressure for the channel's output stream to 0. |
| close() | This method will signal to all consuming for-await-of loops for all of the channel's streams (including all listener streams) to break after they have finished consuming their backlogs of data. |
| kill() | This method will signal to all consuming for-await-of loops for all of the channel's streams (including all listener streams) to break immediately and will reset the aggregate backpressure for the channel to 0. |
| getBackpressure() | Get the aggregate backpressure for all streams on the current channel. The aggregate backpressure represents the highest backpressure of all consumers. |
| getOutputBackpressure() | Get the aggregate backpressure for the main output stream of the current channel. The aggregate backpressure represents the highest backpressure of all consumers. |
| getAllListenersBackpressure() | Get the aggregate backpressure for all listener streams on the current channel. The aggregate backpressure represents the highest backpressure of all consumers. |
| getListenerBackpressure(eventName) | Get the aggregate backpressure for the eventName listener stream on the current channel. The aggregate backpressure represents the highest backpressure of all consumers for that listener. |

Stream management methods Copy Chapter

These methods should only be used for advanced use cases when you need more control over stream management; for example, when you want to break out of a specific consumer loop without affecting any other consumer.
These methods can also be useful to check that consumers are being cleaned up properly and to selectively kill specific consumers which are causing backpressure to build up.
For most use cases, you should try to stick to the methods in the table above as it will lead to cleaner logic.

File: docs/api-ag-client-socket.md Copy Chapter

id: api-ag-client-socket title: AGClientSocket sidebar_label: AGClientSocket Copy Chapter

Inherits from Copy Chapter

AsyncStreamEmitter

Properties Copy Chapter

Events Copy Chapter

The object produced by the listener will have an `id`, `isAuthenticated`, `authToken`, `pingTimeout` and `processPendingSubscriptions` property. If an issue occurred as part of the handshake, there will also be an `authError` property which will hold an error object.
  
  
    The `processPendingSubscriptions` property represents a callback function which, when called, will send all pending channel subscriptions to the server (to activate pending channels).
    Note that this `processPendingSubscriptions` callback will only work if the client `socket.options.autoSubscribeOnConnect` option is set to `false`. See [SocketCluster client API](/docs/api-socket-cluster-client). |

| 'disconnect' | Happens when this socket becomes disconnected from the server. The object produced by the listener will have a code and reason property. |
| 'connectAbort' | Triggers when a new connection is aborted for whatever reason - This could be caused by a failure during the connection phase or it may be triggered intentionally by calling socket.disconnect() while the socket is connecting. The object produced by the listener will have a code and reason property. |
| 'close' | Triggers when a socket is disconnected or the connection is aborted - This is a catch-all event for both 'disconnect' and 'connectAbort'. The object produced by the listener will have a code and reason property. |
| 'connecting' | Triggers whenever the socket initiates a connection to the server - This includes reconnects.
In order capture the very first 'connecting' event, you will need to set the initial autoConnect option
to false when you create the socket with socketClusterClient.create(...) - You will need to register the handler
before you call socket.connect(). |
| 'raw' | This gets triggered whenever the server socket on the other side calls socket.send(...). The object produced by the listener will have a message property. |
| 'kickOut' | Occurs when this socket is kicked out of a channel by the backend. The object produced by the listener will have a channel and message property. |
| 'subscribe' | When the subscription succeeds. The object produced by the listener will have a channel and subscriptionOptions property. |
| 'subscribeFail' | Happens when the subscription fails. The object produced by the listener will have an error, channel and subscriptionOptions property. |
| 'unsubscribe' | When the socket becomes unsubscribed from a channel. The object produced by the listener will have a channel property. |
| 'authStateChange' | Triggers whenever the client's authState changes between socket.AUTHENTICATED and socket.UNAUTHENTICATED states.
The object produced by the listener will have at least two properties: oldAuthState and newAuthState.
If newAuthState is 'authenticated', the object will also have an additional signedAuthToken property which
will be the base64 signed JWT auth token as a string and an authToken property which will represent the token as a plain Object. |
| 'subscribeStateChange' | Triggers whenever a pub/sub channel's state transitions between 'pending', 'subscribed' and 'unsubscribed' states.
The object produced by the listener will have a channel, oldChannelState, newChannelState and subscriptionOptions property. |
| 'subscribeRequest' | Emits the channel name when a subscribe action is invoked by the client. The object produced by the listener will have a channel and subscriptionOptions property. |
| 'authenticate' | Triggers whenever the client is successfully authenticated by the server.
The object produced by the listener will have a signedAuthToken and authToken property. |
| 'deauthenticate' | Triggers whenever the client becomes unauthenticated.
The object produced by the listener will have a oldSignedAuthToken and oldAuthToken property. |
| 'message' | All data that arrives on this socket is emitted through this event as a string.
The object produced by the listener will have a message property. |

Errors Copy Chapter

For the list of all SocketCluster errors (and their properties) see sc-errors.
To check the type of an error in SocketCluster, you should use the name property of the error (do not use the instanceof statement).
Errors which are sent to the client from the server will be dehydrated on the server and rehydrated on the client - As a result, they will be cast
to plain Error objects.

Methods Copy Chapter

See [basic usage guide](/docs/basic-usage) for examples of how to consume listener streams. For more advanced usage, see [StreamDemux](https://github.com/SocketCluster/stream-demux#usage) (this is the library which SocketCluster uses to implement listener streams). |

| closeListener(eventName) | This method will signal to all consuming for-await-of loops (for the eventName listener) to break after they have finished iterating over their current backlogs of events. |
| closeAllListeners() | This method will signal to all consuming for-await-of loops for all listeners to break after they have finished consuming their respective backlogs of events. |
| killListener(eventName) | This method will signal to all consuming for-await-of loops for the eventName listener to break immediately and will reset the backpressure for that listener to 0. |
| killAllListeners() | This method will signal to all consuming for-await-of loops for all listeners to break immediately and will reset the aggregate backpressure for all listeners to 0. |
| receiver(receiverName) | This method returns a receiver stream for the specified receiverName. This object is an asyncIterable which can be consumed with a for-await-of loop. |
| closeReceiver(receiverName) | This method will signal to all consuming for-await-of loops (for the receiverName receiver) to break after they have finished iterating over their current backlogs of data. |
| closeAllReceivers() | This method will signal to all consuming for-await-of loops for all receivers to break after they have finished consuming their respective backlogs of data. |
| killReceiver(receiverName) | This method will signal to all consuming for-await-of loops for the receiverName receiver to break immediately and will reset the backpressure for that receiver to 0. |
| killAllReceivers() | This method will signal to all consuming for-await-of loops for all receivers to break immediately and will reset the aggregate backpressure for all receivers to 0. |
| procedure(procedureName) | This method returns a procedure stream for the specified procedureName. This object is an asyncIterable which can be consumed with a for-await-of loop. |
| closeProcedure(procedureName) | This method will signal to all consuming for-await-of loops (for the procedureName procedure) to break after they have finished iterating over their current backlogs of data. |
| closeAllProcedures() | This method will signal to all consuming for-await-of loops for all procedures to break after they have finished consuming their respective backlogs of data. |
| killProcedure(procedureName) | This method will signal to all consuming for-await-of loops for the procedureName procedure to break immediately and will reset the backpressure for that procedure to 0. |
| killAllProcedures() | This method will signal to all consuming for-await-of loops for all procedures to break immediately and will reset the aggregate backpressure for all procedures to 0. |
| disconnect([code, data]) | Disconnect this socket from the server.
This function accepts two optional arguments:
A custom error code (a Number - Ideally between 4100 to 4500) and data (which can be either reason message String or an Object). |
| transmit(receiverName, data) | Transmit the specified event to the corresponding server side socket receiver. You can pass any JSON-compatible object as data.
This method doesn't return anything and doesn't throw or reject. |
| invoke(procedureName, data) | Invoke the specified procedure (RPC) on the corresponding server side socket. You can pass any JSON-compatible object as data. This method returns a Promise. Note that there is a default timeout of 10 seconds to receive a response from the server. You can change this limit by setting ackTimeout when instantiating the client. If the client does not receive a response in time, the returned Promise will reject with a TimeoutError. |
| send(data, [options]) | Send some raw data to the server. This will trigger a 'raw' event on the server side which will carry the provided data. |
| authenticate(encryptedTokenString) | Perform client-initiated authentication - This is useful if you already have a valid encrypted auth token string and would like to use
it to authenticate directly with the server (without having to ask the user for login details).
Typically, you should perform server-initiated authentication using the socket.setAuthToken() method from the server side.
This method is useful if, for example, you received the token from a different browser tab via localStorage and you want to immediately
authenticate the current socket without having to reconnect the socket. It may also be useful if you're getting the token from a third-party
JWT-based system and you're using the same authKey (see the authKey option passed to the AGServer constructor).
This method returns a Promise which resolves on success and rejects with an Error on failure (e.g. if the token was invalid). |
| deauthenticate() | Perform client-initiated deauthentication - Deauthenticate (logout) the current socket. |
| getBackpressure() | Get the aggregate backpressure for all streams on the current socket. The aggregate backpressure represents the highest backpressure of all consumers. |
| getAllListenersBackpressure() | Get the aggregate backpressure of all listener streams on the current socket. |
| getListenerBackpressure(eventName) | Get the aggregate backpressure for the eventName listener stream on the current socket. The aggregate backpressure represents the highest backpressure of all consumers for that listener. |
| getAllReceiversBackpressure() | Get the aggregate backpressure of all receiver streams on the current socket. |
| getAllProceduresBackpressure() | Get the aggregate backpressure of all procedure streams on the current socket. |
| getAllChannelsBackpressure() | Get the aggregate backpressure of all channel streams on the current socket. |
| All of the following methods are related to pub/sub features of SocketCluster.
SocketCluster lets you interact with channels either directly through the socket or through
AGChannel objects. | |
| transmitPublish(channelName, data) | Publish data to the specified channelName. Do not expect a response from the server.
The channelName argument must be a string.
The data argument can be any JSON-compatible object/array or primitive. |
| invokePublish(channelName, data) | Publish data to the specified channelName. Expect a response from the server.
The channelName argument must be a string.
The data argument can be any JSON-compatible object/array or primitive.
This method returns a Promise which will be rejected if the operation fails.
For example, it can be rejected if the MIDDLEWARE_INBOUND middleware blocks the action on the server side.
The promise will resolve once the server has processed the publish action. |
| subscribe(channelName, [options]) | Subscribe to a channel.
This function returns an AGChannel instance - This object is an asyncIterable and lets you consume data that is published to the channel.
You can provide an optional options object in the form {waitForAuth: true, data: someCustomData} (all properties are optional) - If waitForAuth is true, the channel will wait for the socket to become authenticated before trying to subscribe to the server - These kinds of channels are called "private channels" - Note that in this case, "authenticated" means that the client socket has received a valid JWT authToken - Read about the server side socket.setAuthToken(tokenData) function here for more details. The data property can be used to pass data along with the subscription.

To consume a channel, it is recommended to use a for-await-of loop like this:

for await (
  let data of socket.channel('myChannel')
) {
  // Consume channel data...
}

Because the socket.subscribe(...) method returns an AGChannel instance, you can also consume it directly like this:

for await (
  let data of socket.subscribe('myChannel')
) {
  // Consume channel data...
}

Note that socket.subscribe(...) can be called multiple times for the same channel; if already subscribed, the method will only return the channel. |
| unsubscribe(channelName) | Unsubscribe from the specified channel. This makes any associated AGChannel object inactive.
You can reactivate the AGChannel object by calling subscribe(channelName) again at a later time. |
| channel(channelName) | Returns an AGChannel instance - This object is an asyncIterable.
This method is different from socket.subscribe(...) in that it will not try to subscribe to that channel.
The returned channel will be inactive initially.
You can call channel.subscribe() later to activate that channel when required. |
| closeChannel(channelName) | This method will signal to all consuming for-await-of loops (for the channelName channel and all of its listeners) to break after they have finished iterating over their current backlogs of events. |
| channelCloseOutput(channelName) | This method is like closeChannel(channelName) except that it only closes the main channel (output) stream. Listener consumers on the channel will not be affected. |
| channelCloseAllListeners(channelName) | This method is like closeChannel(channelName) except that it only closes listener streams on the channel. The main channel output stream will not be affected.
To close specific listeners (by eventName) on a specific channel, it's recommended that you use the AGChannel API. |
| closeAllChannels() | This method will signal to all consuming for-await-of loops for all channels (and all of their listeners) to break after they have finished consuming their respective backlogs of events. |
| killChannel(channelName) | This method will signal to all consuming for-await-of loops for the channelName channel (and all of its listeners) to break immediately. This will reset the backpressure for that channel (and all of its event listeners) to 0. |
| channelKillOutput(channelName) | This method is like killChannel(channelName) except that it only kills the main channel (output) stream. Listener consumers on the channel will not be affected. |
| channelKillAllListeners(channelName) | This method is like killChannel(channelName) except that it only kills listener streams on the channel. The main channel output stream will not be affected.
To kill specific listeners (by eventName) on a specific channel, it's recommended that you use the AGChannel API. |
| killAllChannels() | This method will signal to all consuming for-await-of loops for all channels (and all of of their listeners) to break immediately. This will reset the aggregate backpressure for all channels (and all of their event listeners) to 0. |
| killAllChannelOutputs() | This method is similar to killAllChannels() except that it only kills channel output streams; channel event listeners will not be affected. This will reset the aggregate backpressure for all channel output streams to 0. |
| killAllChannelListeners() | This method is similar to killAllChannels() except that it only kills channel listener streams; channel output streams will not be affected. This will reset the aggregate backpressure for all channel listener streams to 0. |
| subscriptions(includePending) | Returns an array of active channel subscriptions which this socket is bound to.
If includePending is true, pending subscriptions will also be included in the list. |
| isSubscribed(channelName, [includePending]) | Check if socket is subscribed to channelName.
If includePending is true, pending subscriptions will also be included in the list. |

Stream management methods Copy Chapter

These methods should only be used for advanced use cases when you need more control over stream management; for example, when you want to break out of a specific consumer loop without affecting any other consumer.
These methods can also be useful to check that consumers are being cleaned up properly and to selectively kill specific consumers which are causing backpressure to build up.
For most use cases, you should try to stick to the methods in the table above as it will lead to cleaner logic.

File: docs/api-ag-exchange.md Copy Chapter

id: api-ag-exchange title: AGExchange sidebar_label: AGExchange Copy Chapter

Inherits from Copy Chapter

AsyncStreamEmitter

Events Copy Chapter

Methods Copy Chapter

| transmitPublish(channel, data) | Publish data to the specified channelName. Do not expect a response from the back end broker.
The channelName argument must be a string.
The data argument can be any JSON-compatible object/array or primitive. |

| --- | --- |
| invokePublish(channel, data) | Publish data to the specified channelName. Expect a response from the broker.
The channelName argument must be a string.
The data argument can be any JSON-compatible object/array or primitive.
This method returns a Promise which will be rejected if the operation fails.
The promise will resolve once the server has processed the publish action. |
| subscribe(channelName) | Subscribe to a channel.
This function returns an AGChannel instance - This object is an asyncIterable and lets you consume data that is published to the channel.

To consume a channel, it is recommended to use a for-await-of loop like this:

for await (
  let data of exchange.channel('myChannel')
) {
  // Consume channel data...
}

Because the agServer.exchange.subscribe(...) method returns an AGChannel instance, you can also consume it directly like this:

for await (
  let data of exchange.subscribe('myChannel')
) {
  // Consume channel data...
}

Note that agServer.exchange.subscribe(...) can be called multiple times for the same channel; if already subscribed, the method will only return the channel. |
| unsubscribe(channelName) | Unsubscribe from the specified channel. This makes any associated AGChannel object inactive.
You can reactivate the AGChannel object by calling subscribe(channelName) again at a later time. |
| channel(channelName) | Returns an AGChannel instance - This object is an asyncIterable.
This method is different from exchange.subscribe(...) in that it will not try to subscribe to that channel.
The returned channel will be inactive initially.
You can call channel.subscribe() later to activate that channel when required. |
| closeChannel(channelName) | This method will signal to all consuming for-await-of loops (for the channelName channel and all of its listeners) to break after they have finished iterating over their current backlogs of events. |
| channelCloseOutput(channelName) | This method is like closeChannel(channelName) except that it only closes the main channel (output) stream. Listener consumers on the channel will not be affected. |
| channelCloseAllListeners(channelName) | This method is like closeChannel(channelName) except that it only closes listener streams on the channel. The main channel output stream will not be affected.
To close specific listeners (by eventName) on a specific channel, it's recommended that you use the AGChannel API. |
| closeAllChannels() | This method will signal to all consuming for-await-of loops for all channels (and all of their listeners) to break after they have finished consuming their respective backlogs of events. |
| killChannel(channelName) | This method will signal to all consuming for-await-of loops for the channelName channel (and all of its listeners) to break immediately. This will reset the backpressure for that channel (and all of its event listeners) to 0. |
| channelKillOutput(channelName) | This method is like killChannel(channelName) except that it only kills the main channel (output) stream. Listener consumers on the channel will not be affected. |
| channelKillAllListeners(channelName) | This method is like killChannel(channelName) except that it only kills listener streams on the channel. The main channel output stream will not be affected.
To kill specific listeners (by eventName) on a specific channel, it's recommended that you use the AGChannel API. |
| killAllChannels() | This method will signal to all consuming for-await-of loops for all channels (and all of of their listeners) to break immediately. This will reset the aggregate backpressure for all channels (and all of their event listeners) to 0. |
| killAllChannelOutputs() | This method is similar to killAllChannels() except that it only kills channel output streams; channel event listeners will not be affected. This will reset the aggregate backpressure for all channel output streams to 0. |
| killAllChannelListeners() | This method is similar to killAllChannels() except that it only kills channel listener streams; channel output streams will not be affected. This will reset the aggregate backpressure for all channel listener streams to 0. |
| subscriptions(includePending) | Returns an array of active channel subscriptions which this exchange client is bound to.
If includePending is true, pending subscriptions will also be included in the list. |
| isSubscribed(channelName, [includePending]) | Check if exchange client is subscribed to channelName.
If includePending is true, pending subscriptions will also be included in the list. |
| getBackpressure() | Get the aggregate backpressure for all streams on the exchange instance. The aggregate backpressure represents the highest backpressure of all consumers. |
| getAllListenersBackpressure() | Get the aggregate backpressure of all listener streams on the exchange instance. |
| getListenerBackpressure(eventName) | Get the aggregate backpressure for the eventName listener stream on the exchange instance. The aggregate backpressure represents the highest backpressure of all consumers for that listener. |
| getAllChannelsBackpressure() | Get the aggregate backpressure of all channel streams on the exchange instance. |

Stream management methods Copy Chapter

These methods should only be used for advanced use cases when you need more control over stream management; for example, when you want to break out of a specific consumer loop without affecting any other consumer.
These methods can also be useful to check that consumers are being cleaned up properly and to selectively kill specific consumers which are causing backpressure to build up.
For most use cases, you should try to stick to the methods in the table above as it will lead to cleaner logic.

File: docs/api-ag-server.md Copy Chapter

id: api-ag-server title: AGServer sidebar_label: AGServer Copy Chapter

Inherits from Copy Chapter

AsyncStreamEmitter

Properties Copy Chapter

| exchange | A client which lets you interact with pub/sub channels and make RPC requests from the server side.
(See API section on the AGExchange object for details). |

| --- | --- |
| clients | An object which holds all fully connected clients on the current server (only those which have completed the handshake).
Keys are socket IDs and the values are AGServerSocket instances. |
| clientsCount | The number of clients currently connected to this server. |
| pendingClients | An object which holds all pending clients on the current server.
Keys are socket IDs and the values are AGServerSocket instances.
Pending clients are those whose socket.state is 'connecting'; this means
all sockets which are in the middle of the handshake phase.
Once a socket completes its handshake, it will be removed from the pendingClients object
and it will be added to the clients object. |
| pendingClientsCount | The number of pending clients currently connected to this server. |
| sourcePort | The public port on which clients connect to SocketCluster (e.g. 80 for
HTTP). |
| httpServer | An instance of the Node.js HTTP server to which this SC server is attached. |
| secure | Whether or not this server uses SSL |
| host | The host name for this server |
| ackTimeout | The timeout in milliseconds in which clients have to receive a response to events
which require acknowledgement. For example, when you call
socket.invoke(...) or socket.invokePublish(...) on the client. |

Events Copy Chapter

the beginning of the client handshake, before the 'connection' event.
  You should not try to send events to the socket while it is in this state. The object produced by the listener will have a `socket` property. |

| 'connectionAbort' | Emitted whenever a socket becomes disconnected during the handshake phase. The object produced by the listener will have a socket, code and reason property. |
| 'connection' | Emitted whenever a new socket connection is established with the server (and the handshake has completed). The object produced by the listener will have a socket, id, pingTimeout and isAuthenticated property. If an authentication error occured during the socket handshake phase, the event object will also have an authError property. The socket object of type AGServerSocket and can be used to interact with the corresponding client. |
| 'disconnection' | Emitted whenever a connected socket becomes disconnected (after the handshake phase). Note that if the socket connection was not fully established (e.g. during the SocketCluster handshake phase), then the 'connectionAbort' event will be triggered instead. The object produced by the listener will have a socket, code and reason property. |
| 'closure' | Emitted whenever a connected socket becomes disconnected (at any stage of the handshake/connection cycle). Note that this event is a catch-all for both 'disconnection' and 'connectionAbort' events. The object produced by the listener will have a socket, code and reason property. |
| 'subscription' | Emitted whenever a socket connection which is attached to the server becomes subscribed to a channel. The object produced by the listener will have a socket (AGServerSocket), a channel and channelOptions property. |
| 'unsubscription' | Emitted whenever a socket connection which is attached to the server becomes unsubscribed from a channel. The object produced by the listener will have a socket and channel property. |
| 'authentication' | Emitted whenever a socket connection which is attached to the server becomes authenticated. The object produced by the listener will have a socket and authToken property. |
| 'deauthentication' | Emitted whenever a socket connection which is attached to the server becomes deauthenticated. The object produced by the listener will have a socket and oldAuthToken property. |
| 'authenticationStateChange' | Triggers whenever the authState of a socket which is attached to the server changes (e.g. transitions between authenticated and unauthenticated states). The object produced by the listener will have a socket, oldAuthState and newAuthState property. |
| 'badSocketAuthToken' | Emitted when a client which is attached to the server tries to authenticate itself with an invalid (or expired) token.
The object produced by the listener will have a socket, authError and signedAuthToken property. |
| 'ready' | Emitted when the server is ready to accept connections. |

Methods Copy Chapter

| constructor(options) | Create the server instance (with config options).
Note that you can also create the server using the attach method of the top-level socketClusterServer instance.
The options object can have any of the following properties (sample showing defaults):

{
// An instance of a Node.js HTTP server.
// https://nodejs.org/api/http.html#http_class_http_server
// This option should not be set if the server is created
// with socketClusterServer.attach(...).
httpServer: null,

// This can be the name of an npm module or a path to a
// Node.js module to use as the WebSocket server engine.
wsEngine: 'ws',

// Custom options to pass to the wsEngine when it is being
// instantiated.
wsEngineServerOptions: {},

// The key which SC will use to encrypt/decrypt authTokens,
// defaults to a 256 bits cryptographically random hex
// string. The default JWT algorithm used is 'HS256'.
// If you want to use RSA or ECDSA, you should provide an
// authPrivateKey and authPublicKey instead of authKey.
authKey: null,

// perMessageDeflate compression. Note that this option is
// passed directly to the wsEngine's Server object.
// So if you're using 'ws' as the engine, you can pass an
// object instead of a boolean.
// Note that by default, per-message deflate only kicks in
// for messages > 1024 bytes.
perMessageDeflate: false,

// If using an RSA or ECDSA algorithm to sign the
// authToken, you will need to provide an authPrivateKey
// and authPublicKey in PEM format (string or Buffer).
authPrivateKey: null,
authPublicKey: null,

// The default expiry for auth tokens in seconds
authDefaultExpiry: 86400,

// The algorithm to use to sign and verify JWT tokens.
authAlgorithm: 'HS256',

// Can be 1 or 2. Version 1 is for maximum backwards
// compatibility with SocketCluster clients.
protocolVersion: 2,

// In milliseconds - If the socket handshake hasn't been
// completed before this timeout is reached, the new
// connection attempt will be terminated.
handshakeTimeout: 10000,

// In milliseconds, the timeout for receiving a response
// when using invoke() or invokePublish().
ackTimeout: 10000,

// Origins which are allowed to connect to the server.
origins: '*:*',

// The maximum number of unique channels which a single
// socket can subscribe to.
socketChannelLimit: 1000,

// The interval in milliseconds on which to
// send a ping to the client to check that
// it is still alive.
pingInterval: 8000,

// How many milliseconds to wait without receiving a ping
// before closing the socket.
pingTimeout: 20000,

// Whether or not an error should be emitted on
// the socket whenever an action is blocked by a
// middleware function
middlewareEmitFailures: true,

// The URL path reserved by SocketCluster clients to
// interact with the server.
path: '/socketcluster/',

// Whether or not clients are allowed to publish messages
// to channels.
allowClientPublish: true,

// Whether or not to batch all socket messages
// for some time immediately after completing
// a handshake. This can be useful in failure-recovery
// scenarios (e.g. batch resubscribe).
batchOnHandshake: false,

// If batchOnHandshake is true, this lets you specify
// How long to enable batching (in milliseconds) following
// a successful socket handshake.
batchOnHandshakeDuration: 400,

// If batchOnHandshake is true, this lets you specify
// the size of each batch in milliseconds.
batchInterval: 50,

// Lets you specify the default cleanup behaviour for
// when a socket becomes disconnected.
// Can be either 'kill' or 'close'. Kill mode means
// that all of the socket's streams will be killed and
// so consumption will stop immediately.
// Close mode means that consumers on the socket will
// be able to finish processing their stream backlogs
// bebfore they are ended.
socketStreamCleanupMode: 'kill'
}
``` |
| --- | --- |
| close() | Close the server and terminate all sockets. |
| setMiddleware(type, middlewareFn) | Lets you set a middleware functions which can be used to manage client access control
      to various features of the AGServer. This is useful for monitoring
      real-time data flows and also to block access to restricted channels
      and resources. Note that only actions from clients pass through middleware.
      Server side calls on the [AGExchange](/docs/api-ag-exchange) client such as `agServer.exchange.transmitPublish(...)` or `agServer.exchange.invokePublish(...)` do not. See [Middleware and authorization](/docs/middleware-and-authorization) guide for a list of all available middleware types and actions. |
| removeMiddleware(type, middlewareFn) | Lets you remove a middleware function which was previously added using the `setMiddleware(...)` method. |
| listener(eventName) | This method returns an event listener stream for the specified `eventName`. This object is an [asyncIterable](https://jakearchibald.com/2017/async-iterators-and-generators/) which can be consumed with a [for-await-of loop](https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Statements/for-await...of).
      
      
        See [basic usage guide](/docs/basic-usage) for examples of how to consume listener streams. For more advanced usage, see [StreamDemux](https://github.com/SocketCluster/stream-demux#usage) (this is the library which SocketCluster uses to implement listener streams). |
| closeListener(eventName) | This method will signal to all consuming `for-await-of` loops (for the `eventName` listener) to `break` after they have finished iterating over their current backlogs of events. |
| closeAllListeners() | This method will signal to all consuming `for-await-of` loops for all listeners to `break` after they have finished consuming their respective backlogs of events. |
| killListener(eventName) | This method will signal to all consuming `for-await-of` loops for the `eventName` listener to `break` immediately and will reset the backpressure for that listener to 0. |
| killAllListeners() | This method will signal to all consuming `for-await-of` loops for all listeners to `break` immediately and will reset the aggregate backpressure for all listeners to 0. |
| getAllListenersBackpressure() | Returns the aggregate backpressure for the server's event listener streams. |
| getListenerBackpressure(eventName) | Get the aggregate backpressure for the `eventName` listener stream on the server instance. The aggregate backpressure represents the highest backpressure of all consumers for that listener. |



## Stream management methods

These methods should only be used for advanced use cases when you need more control over stream management; for example, when you want to `break` out of a specific consumer loop without affecting any other consumer.
These methods can also be useful to check that consumers are being cleaned up properly and to selectively kill specific consumers which are causing backpressure to build up.
For most use cases, you should try to stick to the methods in the table above as it will lead to cleaner logic.



| getListenerConsumerStatsList(eventName) | Get the list of consumers which are consuming data from the specified event listener. This method returns a list of objects which have an `id` and `backpressure` property. |
| --- | --- |
| getAllListenersConsumerStatsList() | Get the list of all consumers which are consuming data from any listener on the server. This method returns a list of objects which have an `id` and `backpressure` property. |
| killListenerConsumer(consumerId) | This will cause the target listener consumer's `for-await-of` loop to `break` immediately. |

---

## File: docs/api-ag-server-socket.md

---
id: api-ag-server-socket
title: AGServerSocket
sidebar_label: AGServerSocket
---

## Inherits from

[AsyncStreamEmitter](https://github.com/SocketCluster/async-stream-emitter#async-stream-emitter)

## Properties



| id | The socket id. |
| --- | --- |
| request | The Node.js HTTP request which initiated the AGServerSocket handshake - This object is useful for accessing HTTP headers (including cookies) which relate to the current socket. |
| remoteAddress | This socket's client IP address. |
| state | The current state of the socket as a string - Can be `socket.CONNECTING`, `socket.OPEN` or `socket.CLOSED`. |
| authState | The authentication state of the socket as a string. Can be either `socket.AUTHENTICATED` or `socket.UNAUTHENTICATED`. |
| authToken | The auth token (as a plain Object) currently associated with the socket. This property will be null if no token is associated with this socket. |
| CONNECTING | **'connecting'** - A string constant which represents this socket's connecting state. See state property. |
| OPEN | **'open'** - A string constant which represents this socket's open/connected state. See state property. |
| CLOSED | **'closed'** - A string constant which represents this socket's closed state. See state property. |
| AUTHENTICATED | **'authenticated'** - A string constant which represents this socket's authenticated authState. |
| UNAUTHENTICATED | **'unauthenticated'** - A string constant which represents this socket's unauthenticated authState. |



## Events



| 'error' | This gets triggered when an error occurs on this socket. The object produced by the listener will have an `error` property |
| --- | --- |
| 'raw' | This gets triggered whenever the client socket on the other side calls `socket.send(...)`. The object produced by the listener will have a `message` property. |
| 'connect' | Triggers when the socket completes the SocketCluster handshake phase and is fully connected.
      Note that if you capture a socket using the `agServer.listener('connection')` listener, then the socket-level 'connect' event will already have occurred by that point so you won't be able to capture it on the socket. The 'connect' event can only be handled from a socket which was captured from an `agServer.listener('handshake')` listener instead.
      The object produced by the listener will have an `id`, `pingTimeout` and `isAuthenticated` property. If an authentication error occured during the socket handshake phase, the event object will also have an `authError` property. |
| 'disconnect' | Happens when the client becomes disconnected from the server. Note that if the socket becomes disconnected during the SocketCluster handshake stage,
      then the `'connectAbort'` event will be triggered instead. The object produced by the listener will have a `code` and `reason` property. |
| 'connectAbort' | Happens when the client disconnects from the server before the SocketCluster handshake has completed (I.e. while `socket.state` was 'connecting').
      Note that the `'connectAbort'` event can only be triggered during the socket's handshake phase before the server's `'connection'` event is triggered.
      The object produced by the listener will have a `code` and `reason` property. |
| 'close' | Happens when the client disconnects from the server at any stage of the handshake/connection cycle.
      Note that this event is a catch-all for both `'disconnect'` and `'connectAbort'` events.
      The object produced by the listener will have a `code` and `reason` property. |
| 'subscribe' | Emitted when the matching client socket successfully subscribes to a channel. The object produced by the listener will have a `channel` and `subscriptionOptions` property. |
| 'unsubscribe' | Occurs whenever the matching client socket unsubscribes from a channel - This includes automatic unsubscriptions triggered by disconnects. The object produced by the listener will have a `channel` property. |
| 'badAuthToken' | Emitted when the client tries to authenticate itself with an invalid (or expired) token.
      Note that this event will typically be triggered as part of the socket handshake (before the socket is connected); therefore, it's generally
      better to listen for the `'badSocketAuthToken'` event directly on the server instance instead.
      The object produced by the listener will have an `authError` and `signedAuthToken` property. |
| 'authenticate' | Triggers whenever the client becomes authenticated. The object produced by the listener will have an `authToken` property. |
| 'deauthenticate' | Triggers whenever the client becomes unauthenticated. The object produced by the listener will have an `oldAuthToken` property. |
| 'authStateChange' | Triggers whenever the socket's `authState` changes (e.g. transitions between authenticated and unauthenticated states).
      The object produced by the listener will have an `oldAuthState` and `newAuthState` property. |
| 'message' | All data that arrives on this socket is emitted through this event as a string. The object produced by the listener will have a `message` property. |



## Methods



| disconnect([code, data]) | Disconnect this socket client.
      This function accepts two optional arguments:
      A custom error code (a Number - Ideally between 4100 to 4500) and data (which can be either reason message String or an Object). |
| --- | --- |
| listener(eventName) | This method returns an event listener stream for the specified `eventName`. This object is an [asyncIterable](https://jakearchibald.com/2017/async-iterators-and-generators/) which can be consumed with a [for-await-of loop](https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Statements/for-await...of).
      
      
        See [basic usage guide](/docs/basic-usage) for examples of how to consume listener streams. For more advanced usage, see [StreamDemux](https://github.com/SocketCluster/stream-demux#usage) (this is the library which SocketCluster uses to implement listener streams). |
| closeListener(eventName) | This method will signal to all consuming `for-await-of` loops (for the `eventName` listener) to `break` after they have finished iterating over their current backlogs of events. |
| closeAllListeners() | This method will signal to all consuming `for-await-of` loops for all listeners to `break` after they have finished consuming their respective backlogs of events. |
| killListener(eventName) | This method will signal to all consuming `for-await-of` loops for the `eventName` listener to `break` immediately and will reset the backpressure for that listener to 0. |
| killAllListeners() | This method will signal to all consuming `for-await-of` loops for all listeners to `break` immediately and will reset the aggregate backpressure for all listeners to 0. |
| receiver(receiverName) | This method returns a receiver stream for the specified `receiverName`. This object is an [asyncIterable](https://jakearchibald.com/2017/async-iterators-and-generators/) which can be consumed with a [for-await-of loop](https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Statements/for-await...of). |
| closeReceiver(receiverName) | This method will signal to all consuming `for-await-of` loops (for the `receiverName` receiver) to `break` after they have finished iterating over their current backlogs of data. |
| closeAllReceivers() | This method will signal to all consuming `for-await-of` loops for all receivers to `break` after they have finished consuming their respective backlogs of data. |
| killReceiver(receiverName) | This method will signal to all consuming `for-await-of` loops for the `receiverName` receiver to `break` immediately and will reset the backpressure for that receiver to 0. |
| killAllReceivers() | This method will signal to all consuming `for-await-of` loops for all receivers to `break` immediately and will reset the aggregate backpressure for all receivers to 0. |
| procedure(procedureName) | This method returns a procedure stream for the specified `procedureName`. This object is an [asyncIterable](https://jakearchibald.com/2017/async-iterators-and-generators/) which can be consumed with a [for-await-of loop](https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Statements/for-await...of). |
| closeProcedure(procedureName) | This method will signal to all consuming `for-await-of` loops (for the `procedureName` procedure) to `break` after they have finished iterating over their current backlogs of data. |
| closeAllProcedures() | This method will signal to all consuming `for-await-of` loops for all procedures to `break` after they have finished consuming their respective backlogs of data. |
| killProcedure(procedureName) | This method will signal to all consuming `for-await-of` loops for the `procedureName` procedure to `break` immediately and will reset the backpressure for that procedure to 0. |
| killAllProcedures() | This method will signal to all consuming `for-await-of` loops for all procedures to `break` immediately and will reset the aggregate backpressure for all procedures to 0. |
| transmit(receiverName, data) | Transmit the specified event to the corresponding client side socket `receiver`. You can pass any JSON-compatible object as data.
      This method doesn't return anything and doesn't throw or reject. |
| invoke(procedureName, data) | Invoke the specified `procedure` (RPC) on the corresponding client side socket. You can pass any JSON-compatible object as data. This method returns a `Promise`. Note that there is a default timeout of 10 seconds to receive a response from the server. You can change this limit by setting `ackTimeout` when instantiating the client. If the client does not receive a response in time, the returned `Promise` will reject with a `TimeoutError`. |
| send(data, [options]); | Send some raw data to the client. This will trigger a 'raw' event on the client which will carry the provided data. |
| setAuthToken(data, [options]) | Authenticate the current socket's client by providing it some token data which you can use to validate their identity and/or access rights.
        Generally, you should call this method in response to a successful login attempt by the client (I.e. The client has provided
        a username and password combination which exists in your database).
      
      
        When you call this method, a JWT (token) containing the specified data will be created and attached to both the current server socket and the client socket on the other side of the connection.
        This method can be called multiple times to update the token with new data - Avoid modifying the socket.authToken property directly because otherwise, the change will not propagate to the client.
      
      
        In the simplest case, the data argument to this method would be an object in the form `{username: 'bob123'}`.
        For a more complex use case, you could also store a list of authorized channel names inside the token (that way, you will be able to check access privileges directly from the socket's token instead of having to read the database every time).
      
        The optional **options** argument is an Object which can be used to modify the token's behavior - Valid properties include any
        option accepted by the node-jsonwebtoken library's sign method.
        [See the list of options here](https://github.com/auth0/node-jsonwebtoken#jwtsignpayload-secretorprivatekey-options-callback).
        Note that if you use a different algorithm than the default 'HS256', you may need to provide an authPrivateKey and authPublicKey instead of the default authKey.
        See `authKey`, `authPrivateKey` and `authPublicKey` options in `AGServer` constructor [here](/docs/api-ag-server#methods).
      
      
        Using auth tokens has the following advantages:
        
          The user doesn't need to login again if their socket loses its connection - Works well with SocketCluster's auto-reconnect feature.
          
            An auth token is shared between sockets across multiple tabs in a browser (under the same domain name).
            If a user logs in on one tab, they will automatically be authenticated on any other tab that they have open.
            Tokens therefore allow you to attach a user's identity to multiple sockets. |
| deauthenticate() | Disassociate the current socket from its auth token.
      This method will also tell the client socket to remove the token from the browser's cookies.
      This is useful for logging out the user. |
| kickOut([channel, message]) | Forcibly unsubscribe this socket from the specified channel. It's also possible to provide an array of channel names.
      All arguments are optional - If no channel name is provided, it will unsubscribe this socket from all channels.
      In addition, the client side of this socket will emit a 'kickOut' event to signify that it was kicked out of the channel. Note that this doesn't prevent the client from resubscribing to that channel later. You will need to update your middleware
      if you want to achieve this. |
| subscriptions() | Returns an array of active channel subscriptions which this socket's client is bound to. |
| isSubscribed(channelName) | Check if socket is subscribed to channelName. |
| getBackpressure() | Returns the aggregate backpressure for all streams associated with the socket. |
| getInboundBackpressure() | Returns the aggregate backpressure for the socket's inbound stream. |
| getOutboundBackpressure() | Returns the aggregate backpressure for the socket's outbound stream. |
| getAllListenersBackpressure() | Returns the aggregate backpressure for the socket's event listener streams. |
| getListenerBackpressure(eventName) | Get the aggregate backpressure for the `eventName` listener stream on the current socket. The aggregate backpressure represents the highest backpressure of all consumers for that listener. |
| getAllReceiversBackpressure() | Returns the aggregate backpressure for the socket's receiver streams. |
| getAllProceduresBackpressure() | Returns the aggregate backpressure for the socket's procedure streams. |



## Stream management methods

These methods should only be used for advanced use cases when you need more control over stream management; for example, when you want to `break` out of a specific consumer loop without affecting any other consumer.
These methods can also be useful to check that consumers are being cleaned up properly and to selectively kill specific consumers which are causing backpressure to build up.
For most use cases, you should try to stick to the methods in the table above as it will lead to cleaner logic.



| getListenerConsumerStatsList(eventName) | Get the list of consumers which are consuming data from the specified event listener. This method returns a list of objects which have an `id` and `backpressure` property. |
| --- | --- |
| getAllListenersConsumerStatsList() | Get the list of all consumers which are consuming data from any listener on the socket. This method returns a list of objects which have an `id` and `backpressure` property. |
| killListenerConsumer(consumerId) | This will cause the target listener consumer's `for-await-of` loop to `break` immediately. |
| getReceiverConsumerStatsList(receiverName) | Get the list of consumers which are consuming data from the specified receiver. This method returns a list of objects which have an `id` and `backpressure` property. |
| getAllReceiversConsumerStatsList() | Get the list of all consumers which are consuming data from all receivers on the socket. This method returns a list of objects which have an `id` and `backpressure` property. |
| killReceiverConsumer(consumerId) | This will cause the target receiver consumer's `for-await-of` loop to `break` immediately. |
| getProcedureConsumerStatsList(procedureName) | Get the list of consumers which are consuming data from the specified procedure. This method returns a list of objects which have an `id` and `backpressure` property. |
| getAllProceduresConsumerStatsList() | Get the list of all consumers which are consuming data from all procedures on the socket. This method returns a list of objects which have an `id` and `backpressure` property. |
| killProcedureConsumer(consumerId) | This will cause the target procedure consumer's `for-await-of` loop to `break` immediately. |

---

## File: docs/api-socket-cluster-client.md

---
id: api-socket-cluster-client
title: socketClusterClient
sidebar_label: socketClusterClient
---

## Properties


| version | The SocketCluster client version number. |
| --- | --- |
| AGClientSocket | A reference to the SocketCluster client socket class. |



## Methods


| create([options]) | Creates and returns a new socket connection to the specified host (or origin if not
        specified). The options argument is optional - If omitted, the socket
        will try to connect to the origin server on the current port. For cross
        domain requests, a typical options object might look like this (example
        over HTTPS/WSS):
      

```js
{
  hostname: 'securedomain.com',
  secure: true,
  port: 443,
  // Only necessary during debug if using a self-signed certificate
  wsOptions: { rejectUnauthorized: false }
}

Here are some properties that you can provide to the options object:

**hostname:** String - Defaults to the current host (ready from
      the URL).
    **secure:** Boolean - Defaults to false
    **port**: Number - Defaults to 80 if !secure otherwise defaults
      to 443.
    **path**: String - The URL which SocketCluster uses to make the initial handshake
      for the WebSocket. Defaults to '/socketcluster/'.
    **query**: Object - A map of key-value pairs which will be used as query
      parameters for the initial HTTP handshake which will initiate the WebSocket connection.
    **ackTimeout**: Number (milliseconds) - This is the timeout for
      getting a response to a AGClientSocket invoke action.
    **autoConnect**: Boolean - Whether or not to automatically connect
      the socket as soon as it is created. Default is true.
    **autoReconnect**: Boolean - Whether or not to automatically reconnect
      the socket when it loses the connection.
    **autoReconnectOptions**: Object - Valid properties are: `initialDelay`
      (milliseconds), `randomness` (milliseconds), `multiplier` (decimal;
      default is 1.5) and `maxDelay` (milliseconds).
    **disconnectOnUnload**: Boolean - Whether or not a client automatically disconnects on page unload.
      If enabled, the client will disconnect when a user navigates away from the page. This can happen when a user closes the tab/window,
      clicks a link to leave the page, or types a new URL into the address bar.
      Defaults to true.
    **wsOptions**: Object - This object will be passed to the constructor of the `ws` `WebSocket` instance.
      The list of supported properties is [here](https://github.com/websockets/ws/blob/master/doc/ws.md#new-websocketaddress-protocols-options).
    **batchOnHandshake**: Boolean - Whether or not to start batching messages immediately after the connection handshake completes. This is useful for handling connection recovery when the client tries to resubscribe to a large number of channels in a very short amount of time. Defaults to false.
    **batchOnHandshakeDuration**: Number - The amount of time in milliseconds after the handshake completes during which all socket messages will be batched. Defaults to 100.
    **batchInterval**: Number - The amount of milliseconds to wait before flushing each batch of messages. Defaults to 50.
    **timestampRequests**: Boolean - Whether or not to add a timestamp
      to the WebSocket handshake request.
    **timestampParam**: String - The query parameter name to use to
      hold the timestamp.
    **authEngine**: Object - A custom engine to use for storing and
      loading JWT auth tokens on the client side.
    **authTokenName**: String - The name of the JWT auth token (provided
      to the authEngine - By default this is the localStorage variable
      name); defaults to 'socketcluster.authToken'.
    **binaryType**: String - The type to use to represent binary on
      the client. Defaults to 'arraybuffer'.
    **cloneData**: Boolean - If you set this to true, any data/objects/arrays that you pass to the client socket
      will be cloned before being sent/queued up. If the socket is disconnected and you emit an event, it will be added to a queue
      which will be processed upon reconnection. The cloneData option is false by default; this means that if you emit/publish an object
      and that object changes somewhere else in your code before the queue is processed, then the changed version of that object
      will be sent out to the server.
    **autoSubscribeOnConnect**: Boolean - This is true by default. If you set this to false, then the socket will not automatically
      try to subscribe to pending subscriptions on connect - Instead, you will have to manually invoke the `processSubscriptions` callback
      from inside the `'connect'` event handler on the client side. See [AGClientSocket API](/docs/api-ag-client-socket).
      This gives you more fine-grained control with regards to when pending subscriptions are processed after the socket connection is established (or re-established).
    **codecEngine**: Object - Lets you set a custom codec engine. This allows you to specify how
      data gets encoded before being sent over the wire and how it gets decoded once it reaches the other side.
      The codecEngine must be an object which exposes an `encode(object)` and a `decode(encodedData)` function.
      The encode function can return any data type - Commonly a string or a Buffer/ArrayBuffer.
      The decode function needs to return a JavaScript object which adheres to the
      [SC protocol](https://github.com/SocketCluster/socketcluster/blob/master/socketcluster-protocol.md).
      The idea of using a custom codec is that it allows you to compress SocketCluster packets in any format you like (optimized for any use case) -
      By decoding these packets back into their original protocol form, SocketCluster will be able process them appropriately.
      Note that if you provide a codecEngine when creating a client socket, you will need to make sure that the server uses the same codec
      by passing the same engine to the `AGServer` constructor (using the `codecEngine` option).
      The default codec engine used by SocketCluster is [here](https://github.com/SocketCluster/sc-formatter/blob/master/index.js). |

| --- | --- |

File: docs/api-socket-cluster-server.md Copy Chapter

id: api-socket-cluster-server title: socketClusterServer sidebar_label: socketClusterServer Copy Chapter

Properties Copy Chapter

Methods Copy Chapter

| attach(httpServer, options) | Create a new SocketCluster server and attach it to the specified Node.js HTTP server instance. This method returns an AGServer instance.
The list of possible options which will be passed to the SocketCluster server instance can be found here. |

| --- | --- |
| listen(port, options, fn) | Create a SocketCluster server instance which is already attached to an HTTP server instance and ready to use. This method returns an AGServer instance. |

File: docs/authentication.md Copy Chapter

id: authentication title: Authentication sidebar_label: Authentication Copy Chapter

Feature overview Copy Chapter

The default authentication mechanism in SocketCluster is JWT; it is essentially the same as it was previous versions of SocketCluster; the main difference is that with version 15+ any function which accepted a callback will now return a Promise instead. You can use either the HTTP-based flow or the WebSocket-based flow for doing authentication.

The custom authKey string which you pass as an option to the AGServer constructor will be used to sign and verify JWT tokens from client sockets.
The initial authentication phase occurs automatically as part of the socket handshake. Authentication can also be done at runtime on a connected socket using the client's socket.authenticate(...) method.

API overview Copy Chapter

HTTP flow Copy Section

The best (and also simplest) way to do authentication in SocketCluster over HTTP is by creating and signing a new JWT token from express and then sending it to the client side (as a string in the HTTP response). Once you have the token on the client side, you need to add it to localStorage under the key 'socketcluster.authToken' (this is the default JWT localStorage key for SocketCluster). The SocketCluster client socket will automatically pick up the JWT from local storage when a new connection is created.

When you create and sign the JWT token inside your HTTP express route, you need to use the same key which was specified as authKey when you instantiated the SocketCluster AGServer.

You can create the JWT token however you like (so long as it meets JWT specifications). In Node.js, you can use the jsonwebtoken module directly: https://www.npmjs.com/package/jsonwebtoken

or for convenience you can also use SocketCluster's agServer.auth.signToken(...) function like this:

// Sample token data, don't store any sensitive/secret data here
// it will be signed but not encrypted.
let myTokenData = {
  username: 'bob',
  language: 'English',
  company: 'Google',
  groups: ['engineering', 'science', 'mathematics']
};

// agServer.signatureKey below is the key which SocketCluster uses to sign the token.
// By default, agServer.signatureKey is equal to agServer.options.authKey
// (and also equal to agServer.verificationKey); but these may differ if you
// switch to a different JWT algorithm in the future.

let signedTokenString;
try {
  signedTokenString = await agServer.auth.signToken(myTokenData, agServer.signatureKey);
} catch (error) {
  // ... Handle failure to sign token.
  return;
}

// The signedTokenString variable is the signed JWT token; you can send it
// to your front end however you like (e.g. in HTTP response).

Once you have the JWT token on the front end, you should add it to localStorage under the key 'socketcluster.authToken' - By doing this, the SocketCluster client will automatically pick up the JWT token when doing its handshake/connection.

In the browser, you can add the token to localStorage like this:

localStorage.setItem('socketcluster.authToken', token);

WebSocket flow Copy Section

A sample WebSocket-based authentication flow might look like this:

// Client code
socket = socketClusterClient.create();

// The 'connect' event carries a status object which has a
// boolean 'isAuthenticated' property - It will be true if the client socket carried
// a valid token at the time the connection was established.
// The status object may also have an 'authError' property (which contains an
// Error object) if the JWT authentication failed during the handshake.

let status = await socket.listener('connect').once();

if (status.isAuthenticated) {
  // goToMainScreen();
} else {
  // goToLoginScreen();
}

Assume that the user is not authenticated and so they are sent to the login screen. We log them in by emitting a 'login' event from the client:

// Client code

let credentials = {
  username: 'alice123',
  password: 'thisisapassword654'
};

try {
  // Invoke a custom 'login' procedure (RPC) on our server socket
  // then wait for the socket to be authenticated.
  await Promise.all([
    socket.invoke('login', credentials),
    socket.listener('authenticate').once()
  ]);
} catch (error) {
  // showLoginError(err);
  return;
}

// Socket is authenticated!
// goToMainScreen();

Note that since SocketCluster v6, you typically don't need to wait for the socket authenticate event as shown above (assuming that your login procedure is written as shown in the example below). Nonetheless, using Promise.all as shown above is the surest way to check for authentication regardless of how the login procedure is implemented.

On the server, we would need some code to process the login:

// Server code

// This is a slightly simplified version of what it might look
// like if you were using MySQL as a database.

(async () => {
  for await (let request of socket.procedure('login')) {
    let passwordHash = sha256(credentials.password);

    let userQuery = 'SELECT * FROM Users WHERE username = ?';
    let userData;
    try {
      let rows = await mySQLClient.query(userQuery, [credentials.username]);
      userData = rows[0];
    } catch (error) {
      let loginError = new Error(`Could not find a ${credentials.username} user`);
      loginError.name = 'LoginError';
      request.error(loginError);

      continue;
    }

    let isValidLogin = userData && userData.password === passwordHash;
    if (!isValidLogin) {
      let loginError = new Error('Invalid user credentials');
      loginError.name = 'LoginError';
      request.error(loginError);

      continue;
    }

    // End the 'login' request successfully.
    request.end();

    // This will give the client a token so that they won't
    // have to login again if they lose their connection
    // or revisit the app at a later time.
    socket.setAuthToken({username: credentials.username, channels: userData.channels});
  }
})();

Verify and use the JWT token Copy Section

You can verify and read the JWT token in the same way regardless of whether you used the HTTP or WebSocket auth flow. Once the token has been set/captured by SocketCluster,
you can access it from inside your middleware functions (example using a PUBLISH_IN action inside the MIDDLEWARE_INBOUND middleware):

// Server code

agServer.setMiddleware(agServer.MIDDLEWARE_INBOUND, async (middlewareStream) => {
  for await (let action of middlewareStream) {

    if (action.type === action.PUBLISH_IN) {
      let authToken = action.socket.authToken;
      if (
        !authToken ||
        !Array.isArray(authToken.channels) ||
        authToken.channels.indexOf(action.channel) === -1
      ) {
        let publishError = new Error(
          `You are not authorized to publish to the ${action.channel} channel`
        );
        publishError.name = 'PublishError';
        action.block(publishError);

        continue; // Go to the start of the loop to process the next inbound action.
      }
    }

    // Any unhandled case will be allowed by default.
    action.allow();
  }
});

!! Note that in this case, the token contains all the information that we need to authorize this publish action, but we didn't really need to store the channels list inside the JWT - An alternative approach would have been to fetch the user account details from the database using the username from the JWT (but it would require an extra database lookup; bad for performance). See the section on middleware and authorization for more info about middleware in SocketCluster.

File: docs/basic-usage.md Copy Chapter

id: basic-usage title: Basic usage sidebar_label: Basic usage Copy Chapter

API overview Copy Chapter

[Server] Create a server Copy Section

Similar logic should already be inside server.js if using the default boilerplate:

// --- in server.js ---
const http = require('http');
const socketClusterServer = require('socketcluster-server');

let options = {
  // ...
};

let httpServer = http.createServer();
let agServer = socketClusterServer.attach(httpServer, options);

// port 8000
httpServer.listen(8000);

The httpServer argument must be an instance of a Node.js HTTP server.
The list of possible options which can be passed to the SocketCluster server instance can be found here.
Note that you can also use the ECMAScript import syntax.

[Server] Listen for inbound socket connections Copy Section

Inside server.js, you can find the for-await-of loop which is handling inbound connections. It should look like this:

// --- in server.js ---

// SocketCluster/WebSocket connection handling loop.
(async () => {
  for await (let {socket} of agServer.listener('connection')) {
    // Handle socket connection.
  }
})();

[Client] Connect to the server Copy Section

Inside public/index.html, a client connects to the server like this:

// --- in public/index.html ---

let socket = socketClusterClient.create();

^ If the connection succeeds, this will cause the connection loop on the server side to iterate once.

!! You can pass an options object to the socketClusterClient.create(...) function.

[Server] Listen for inbound messages on a socket Copy Section

You can use a socket.receiver(...) within a for-await-of loop to handle messages on a socket:

// --- in server.js ---

// SocketCluster/WebSocket connection handling loop.
(async () => {
  for await (let {socket} of agServer.listener('connection')) {

    (async () => {
      // Set up a loop to handle remote transmitted events.
      for await (let data of socket.receiver('customRemoteEvent')) {
        // ...
      }
    })();

  }
})();

!! You can also iterate over a socket.receiver(receiverName) stream on a client socket using the same syntax.

[Client] transmit messages through a socket Copy Section

// --- in public/index.html ---

// ... After the socket is created.
socket.transmit('customRemoteEvent', 123);

^ If the message reaches the server, this will cause the customRemoteEvent loop on the server side to iterate once; data will be 123.

!! You can also use the same syntax to transmit from the server socket. Transmit can never fail, so you don’t need to wrap it in a try-catch block.

[Server] Listen for inbound RPCs on a socket Copy Section

Unlike messages, RPCs expect a response from the other side.
You can use a socket.procedure(...) within a for-await-of loop to handle RPCs on a socket:

// --- in server.js ---

// SocketCluster/WebSocket connection handling loop.
(async () => {
  for await (let {socket} of agServer.listener('connection')) {

    (async () => {
      // Set up a loop to handle and respond to RPCs.
      for await (let request of socket.procedure('customProc')) {
        if (request.data && request.data.bad) {
          let badCustomError = new Error('Server failed to execute the procedure');
          badCustomError.name = 'BadCustomError';
          request.error(badCustomError);

          continue;
        }
        request.end('Success');
      }
    })();

  }
})();

!! You can also iterate over a socket.procedure(procedureName) stream on a client socket using the same syntax.

[Client] Invoke RPCs through a socket Copy Section

// --- in public/index.html ---

// ... After the socket is created.
(async () => {
  let result = await socket.invoke('customProc', {foo: 'bar'});
  // result will be 'Success'
})();

!! You should always add a try-catch block around the socket.invoke(...) call to capture async errors:

// --- in public/index.html ---

// ... After the socket is created.
(async () => {
  let result;
  try {
    result = await socket.invoke('customProc', {bad: true});
  } catch (error) {
    // error will throw.
    // error.name will be 'BadCustomError'.
  }
})();

!! You can also use the same syntax on server sockets.

[Client] Subscribe to a channel and watch for messages Copy Section

(async () => {
  let channel = socket.subscribe('foo');
  for await (let data of channel) {
    // ... Handle channel data.
  }
})();

[Client] Publish to a channel without waiting for acknowledgement Copy Section

// Publish data; do not wait for an acknowledgement from the server.
socket.transmitPublish('foo', 'This is some data');

!! You can also call transmitPublish on an AGChannel object; in this case, you should omit the first argument, for example:

let fooChannel = socket.channel('foo');
fooChannel.transmitPublish('This is some data');

[Client] Publish to a channel and wait for acknowledgement Copy Section

(async () => {
  try {
    // Publish data; wait for an acknowledgement from the server.
    await socket.invokePublish('foo', 'This is some more data');
  } catch (error) {
    // ... Handle potential error if server does not acknowledge before timeout.
  }
})();

!! You can also call invokePublish on an AGChannel object; in this case, you should omit the first argument, for example:

(async () => {
  let fooChannel = socket.channel('foo');
  try {
    // Publish data; wait for an acknowledgement from the server.
    await fooChannel.invokePublish('This is some more data');
  } catch (error) {
    // ... Handle potential error if server does not acknowledge before timeout.
  }
})();

[Server] Publish to a channel without waiting for acknowledgement Copy Section

// Publish data; do not wait for an acknowledgement from the back end broker.
agServer.exchange.transmitPublish('foo', 'This is some data');

[Server] Publish to a channel and wait for acknowledgement Copy Section

(async () => {
  try {
    // Publish data; wait for an acknowledgement from the back end broker (if it exists).
    await agServer.exchange.invokePublish('foo', 'This is some more data');
  } catch (error) {
    // ... Handle potential error if broker does not acknowledge before timeout.
  }
})();

External documentation Copy Chapter

• https://github.com/SocketCluster/socketcluster
• https://github.com/SocketCluster/socketcluster-server
• https://github.com/SocketCluster/socketcluster-client

File: docs/consumables-and-consumers.md Copy Chapter

id: consumables-and-consumers title: Consumables and Consumers sidebar_label: Consumables and Consumers Copy Chapter

In SocketCluster, streams returned by functions such as socket.listener(...), socket.receiver(...), socket.procedure(...) and socket.channel(...) are all async iterables.

This means that you can iterate over any stream directly like this:

for await (let data of socket.receiver('foo')) {
  // ...
}

Whenever you start iterating over a stream, a new async iterator will be created implicitly with a new (empty) message/event buffer.
This feature allows you to iterate over a single stream using multiple loops in parallel; each loop can await and process messages/events at its own pace.

Consumables Copy Chapter

While the pattern described above is succinct and convenient, sometimes, you need more flexibility.
For example, you may want to await for a back end operation to complete before processing events from a receiver.

Consider this naive approach:

// Back end.

(async () => {
  for await (let {socket} of agServer.listener('connection')) {

    (async () => {
      await doSomethingWhichTakesAFewSeconds();

      for await (let data of socket.receiver('foo')) {
        // ...
      }
    })();

  }
})();

Assume that your client has this logic:

// Front end.

let socket = socketClusterClient.create();

for await (let event of socket.listener('connect')) {
  socket.transmit('foo', 123);
}

In this case, the socket.receiver('foo') stream on the back end will not receive our client's 123 message; this is because we added an await doSomethingWhichTakesAFewSeconds() statement before the for await (let data of socket.receiver('foo')) {...} loop.
Any message which arrives on the socket before the loop begins iterating will be ignored/missed (and not buffered).

Note that simply instantiating the stream sooner and putting it inside a variable will not fix the issue:

// Back end.

(async () => {
  for await (let {socket} of agServer.listener('connection')) {

    (async () => {
      // This will not work because the iterator is not yet created at this point.
      let fooStream = socket.receiver('foo');

      // If any messages arrive during this time, they will be ignored!
      await doSomethingWhichTakesAFewSeconds();

      // The iterator gets created (and starts buffering) here!
      for await (let data of fooStream) {
        // ...
      }
    })();

  }
})();

To fix this issue, you need to explicitly tell the SocketCluster stream when to create the consumer/consumable. This can be done using the stream.createConsumable() method like this:

// Back end.

(async () => {
  for await (let {socket} of agServer.listener('connection')) {

    (async () => {
      // This will create a consumable which will start buffering messages immediately.
      let fooStreamConsumable = socket.receiver('foo').createConsumable();

      await doSomethingWhichTakesAFewSeconds();

      // This loop will start from the beginning of the buffer.
      for await (let data of fooStreamConsumable) {
        // ...
      }
    })();

  }
})();

Note that a consumable is an async iterable; therefore, it should only be consumed by a single loop. If multiple loops try to iterate over the same consumable, they will end up competing for each message (which can lead to unexpected behavior).

Consumer Copy Chapter

A Consumer is an async iterator. Using a Consumer directly is the most flexible, but most verbose way of consuming messages in SocketCluster.

Using Consumer instances allows you to iterate over streams using regular for and while loops and to handle messages in more advanced ways.

For example, the code above could be rewritten as:

// Back end.

(async () => {
  for await (let {socket} of agServer.listener('connection')) {

    (async () => {
      // This will create a consumer which will start buffering messages immediately.
      let fooStreamConsumer = socket.receiver('foo').createConsumer();

      await doSomethingWhichTakesAFewSeconds();

      // This loop will start from the beginning of the buffer.
      while (true) {
        let packet = await fooStreamConsumer.next();
        if (packet.done) break;
        console.log('Packet:', packet.value);
      }
    })();

  }
})();

Note that a Consumer is an async iterator; therefore, it should only be consumed by a single loop.