sseries-of-tubes

Takes Express/Connect routes and creates stream.Writable endpoints for Server Sent Events.

Introduction

SSEriesOfTubes attaches to an http.Server and efficiently manages each endpoint for Server Sent Events by creating a writeable StringTube Stream that is piped to every http.Response. StringTube is a utf8 String Stream subclassed from Through2.

Additionally, SSEriesOfTubes can wrap a standard Express/Connect route controller (a function that takes arguments (req, res, next)), poll that controller for data, and push that data to the SSE stream.

This way, any API can easily support REST/polling and Server Sent Events with the same route controller.

SSEriesOfTubes is an EventEmitter based on chrisdickinson/sse-stream, so it handles SSE headers, keep-alive messaging, and emits events around the connection and polling lifecycle.

Installation

npm install --save sseries-of-tubes

Usage

Example

# on server 
http           = require "http"
express        = require "express"
SSEriesOfTubes = require "sseries-of-tubes"
 
server         = new http.Server
app            = express()
sseriesOfTubes = new SSEriesOfTubes server
 
sendHello = (req, res, next) ->
    res.json hello: "world"
 
# Sends {"hello": "world"} every 3s to every client connected to /sse/hello 
app.get "/sse/hello", sseriesOfTubes.plumb sendHello, 3
 
...
 
# on client 
source = new EventSource "/sse/hello"
# log {"hello": "world"} as received every 3s 
source.onmessage = (data) -> console.log data
 

Interface

SSEriesOfTubes.StringTube

Pointer to class StringTube, a Through2 Readable/Writable stream.

SSEriesOfTubes::constructor(server, keepAliveInterval)

new SSEriesOfTubes http.Server server, int keepAliveInterval

returns sseriesOfTubes instance

Creates a new SSEriesOfTubes instance that reacts to http.Server "listening" and "close" events.

sseriesOfTubes.plumb(fn, interval)

sseriesOfTubes.plumb Function route, int interval

returns Function route

Creates a new StringTube source which polls a route to write to all clients for an endpoint.

sseriesOfTubes.plumb(fn, interval, event)

sseriesOfTubes.plumb Function route, int interval, String event

returns Function route

Creates a new StringTube source which polls a route to write to all clients for an endpoint, and will send the event event (in EventSource syntax) before sending data.

sseriesOfTubes.plumb()

sseriesOfTubes.plumb()

returns Function route

Creates a new StringTube source which writes to all clients for an endpoint.

sseriesOfTubes.source(url)

sseriesOfTubes.source String url

returns StringTube source

Retrieves the StringTube source for an endpoint.

sseriesOfTubes.combine(router, paths...)

sseriesOfTubes.combine express.Router router, String paths...

returns Function route

Creates a new StringTube source by combining existing StringTube sources referenced by their path. This way different combinations of streams can be provided without duplicating pollers. Requires a router conforming to the express.Router interface for method handle.

sseriesOfTubes.multiplex(router, param = "streams")

sseriesOfTubes.multiplex express.Router router, String param

returns Function route

Creates new StringTube sources by dynamically combining existing StringTube sources, referenced by their path in an comma-separated list provided by the client in the query or body as parameter param. This method is similar to combine, but works with paths that have route parameters. Requires a router conforming to the express.Router interface for method handle.

sseriesOfTubes.destroy()

sseriesOfTubes.destroy()

Ends all SSE client connections and destroys all clients.

Events

connection

sseriesOfTubes.on "connection", (client) ->

On client connection, calls bound function with an instance of Client.

poll

sseriesOfTubes.on "poll", (url) ->

On route start poll, calls bound function with the url endpoint.

plumb

sseriesOfTubes.on "plumb", (url) ->

On route start plumb (creates StringTube source), calls bound function with the url endpoint.

stop

sseriesOfTubes.on "stop", (url) ->

On last client disconnect, calls bound function with the url endpoint.

License

MIT