SocketCluster

16.0.1

AGExchange

Inherits from

AsyncStreamEmitter

Events

'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 exchange becomes unsubscribed from a channel. The object produced by the listener will have a channel property.

Methods

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

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 exchange instance. 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.
channelGetOutputConsumerStatsList(channelName) Get the list of consumers which are consuming output data from the specified channel. This method returns a list of objects which have an id and backpressure property.
channelGetAllListenersConsumerStatsList(channelName) Get the list of all consumers which are consuming events from any listener on the specified channel. This method returns a list of objects which have an id and backpressure property.
getAllChannelOutputsConsumerStatsList() Get the list of all consumers which are consuming output data from any channel which is bound to the exchange instance. This method returns a list of objects which have an id and backpressure property.
getAllChannelListenersConsumerStatsList() Get the list of all consumers which are consuming events from any listener on all channels associated with the exchange instance. This method returns a list of objects which have an id and backpressure property.
killChannelOutputConsumer(consumerId) This will cause the target channel output consumer's for-await-of loop to break immediately.
killChannelListenerConsumer(consumerId) This will cause the target channel listener consumer's for-await-of loop to break immediately.
← AGServerSocketAGAction →