sse-service v1.0.9

Server-Sent Events (SSE) service

Lightweight, fast and framework-agnostic SSE service for NodeJS

Written on TS.

Features

• Could be embedded to Express, Fastify, Nest, etc. implemantation
• Easy to manage connections by providing a groups feature of user conections
• Allows a few connections per user (from different devices for example), so you can send data to every user connection providing the user id
• Allows to pass through original response headers(headers that were added by your framework, for example) to end up response stream

Basic usages examples

Fastify

import { connectSSE } from 'sse-service';
import { fastify } from 'fastify';

const server = fastify({ logger: true });

server.route({
  method: 'GET',
  url: '/sse',
  handler: async (req, res) => {
    connectSSE({
      clientId: 'unique id', // this is a primary key of how your client connection could be retrieved, could be ommited, library will take care of creating unique id  
      // client id could be retrieved from the JWT token for example
      req: req.raw, // here we link to raw NodeJS request object, because its wrapped by Fastify
      res: res.raw, // the same as for request
      originalResHeaders: res.getHeaders(), // sometimes you need to pass through response headers added by Fastify, this headers will be added to raw NodeJS response stream
      onDisconnectCb: () => {
        // ....
      }, // sometimes you need to perform some actions when user disconnects
    });
  },
});

Then you can use it like this in some other place where you receive updates from some message queue for instance:

// someService.ts

import { sendSSEToClient } from 'sse-service';

// ....

// some function that handles update from message queue or whatever
const processUpdate = (update) => {
  const { userId, ...payload } = update;
  
  // userId equals to value you passed as clientId in connectSSE method
  sendSSEToClient(userId, payload); // this will send payload to every connection that user with userId has
  
  // OR send update to all connected users:

  sendSSEToAll(payload);
}; 

Nest

Take a note that Nest has built-in SSE implementation: https://docs.nestjs.com/techniques/server-sent-events

So maybe for you goals it will be enough using it

// sse.controller.ts

import { connectSSE } from 'sse-service';

// ...

export class SseController {

    // ...
  
    public sse(@Req() req, @Res() res) {
      connectSSE({
        clientId: req.user.id, // this is a primary key of how your client connection could be retrieved, could be ommited, library will take care of creating unique id  
        // client id could be retrieved from the JWT token for example
        req,
        res,
        originalResHeaders: res.getHeaders(), // sometimes you need to pass through response headers added by Nest (actually it is Fastify or Express under the hood), this headers will be added to raw NodeJS response stream
        onDisconnectCb: () => {
          // ... 
        }, // sometimes you need to perform some actions when user disconnects
      });
    }
}

Then you can use it like this in some other place where you receive updates from some message queue for instance:

// someService.ts

import { sendSSEToClient } from 'sse-service';

// ....

// some function that handles update from message queue or whatever
const processUpdate = (update) => {
  const { userId, ...payload } = update;
  
  // userId equals to value you passed as clientId in connectSSE method
  sendSSEToClient(userId, payload); // this will send payload to every connection that user with userId has

  // OR send update to all:

  sendSSEToAll(payload);
}; 

Express

var express = require('express')
var app = express()

app.get('/sse', function (req, res) {
  connectSSE({
    clientId: req.user.id, // this is a primary key of how your client connection could be retrieved, could be ommited, library will take care of creating unique id  
    // client id could be retrieved from the JWT token for example
    req,
    res,
    originalResHeaders: res.getHeaders(), // sometimes you need to pass through response headers added by Express, this headers will be added to raw NodeJS response stream
    onDisconnectCb: () => {
      // ... 
    }, // sometimes you need to perform some actions when user disconnects
  });
})

app.listen(3000);

Then you can use it like this in some other place where you receive updates from some message queue for instance:

// someService.ts

import { sendSSEToClient } from 'sse-service';

// ....

// some function that handles update from message queue or whatever
const processUpdate = (update) => {
  const { userId, ...payload } = update;
  
  // userId equals to value you passed as clientId in connectSSE method
  sendSSEToClient(userId, payload); // this will send payload to every connection that user with userId has

  // OR send update to all:

  sendSSEToAll(payload);
}; 

Installation

npm install sse-service

API

connectSSE(params)

Setups SSE connection

params:

{
  req: HttpRequest; // NodeJS raw request object
  
  res: HttpResponse; // NodeJS raw response object
  
  originalResHeaders: HttpResponseHeaders; // response headers from the response object of you framework,
  // for example Fastify wraps original response object and could add additional headers,
  // so if you want to pass them through call .getHeaders() on the framework's response object and pass here 
  
  meta?: Record<string, any>; // any metadata to be persisted in client object that will be created 
  
  clientId?: string; // this is a primary key of how your client connection could be retrieved, 
  // could be ommited, library will take care of creating unique id  
  // client id could be retrieved from the JWT token for example
  
  onDisconnectCb?: () => void; // sometimes you need to perform some actions when user disconnects
  
  groupName?: string; // this param allows you to add this user connection to some specific group(namespace)
  // so later you could send a SSE message for all users connections from this group
  // by default all connections being added to default group name
}

Method returns:

 {
    client: {
        connections: HttpResponse[]; // array of raw NodeJS responses, 
        // will contain more then one if we add few connections by the same clientId param
        meta?: Record<string, any>; // any meta you have added by meta param
        clientId: string; // client id you passed to or the one generated by lib
    };
    currConnection: HttpResponse // raw NodeJS response object that was esatablished by this specific connectSSE method call 
 }
 

sendSSEToClient(clientId, msg [, eventType [, groupName , sseMsgId ] ])

Send message to specific client by id

Params:

sendSSEToConnection(connection, msg [, eventType , sseMsgId ])

Send message to specific connection

Params:

sendSSEToAll(connection, msg [, eventType , sseMsgId ])

Send message to all active clients

Params:

getClientObj(clientId , groupName )

Get user's active connections and his meta

Params:

Method returns:

 {
    connections: HttpResponse[]; // array of raw NodeJS responses,
    // will contain more then one if we add few connections by the same clientId param
    meta?: Record<string, any>; // any meta you have added by meta param
    clientId: string; // client id you passed to or the one generated by lib
 };

getAllClientObjs(groupName)

Get connections and meta of all active clients

Params:

Method returns:

 Array<{
    connections: HttpResponse[]; // array of raw NodeJS responses,
    // will contain more then one if we add few connections by the same clientId param
    meta?: Record<string, any>; // any meta you have added by meta param
    clientId: string; // client id you passed to or the one generated by lib
 }>;

setClientMetadata(clientId , groupName )

Set client's meta

Params:

Method returns:

 {
    connections: HttpResponse[]; // array of raw NodeJS responses,
    // will contain more then one if we add few connections by the same clientId param
    meta?: Record<string, any>; // any meta you have added by meta param
    clientId: string; // client id you passed to or the one generated by lib
 };k

License

MIT