Redis 5 adds Streams, a powerful new data structure which simplifies the construction of real-time applications. The new XREAD command blocks a redis connection while awaiting new data, allows reading from multiple streams, and can be canceled with the new CLIENT UNBLOCK command. This package wraps XREAD and XADD such that new subscriptions automatically UNBLOCK the existing stream, add their new subscription (and handle offset tracking), and restart the stream. The end effect is that you'll need only two Redis connections open - one for reading and one for writing (including the UNBLOCKing).

With redis-streams-aggregator, you can just call .subscribe() and .add() and leave the stream management to us!

yarn add redis-streams-aggregator

constRedisStreamsAgg=require("redis-streams-aggregator");conststreams=newRedisStreams();// options like:{host: ..., port: ... } etc// Sign up for XREAD, get all messages to 'testId' in real-timestreams.subscribe("testId",messages=>{console.log("I got messages!",{ messages });});// Write data to a stream (very thin wrapper around XADD)streams.add("testId",{value: "foobar"});

Messages events provide arrays of "Message" objects, which have the following properties:

constMessage={offset: string,key: value,key2: value2...}

For example, consider the following:

// Serverstreams.add("testChannel",{foo: "bar"});streams.subscribe("testChannel","$",messages=>{// messages === [// [ "1518951480106-0",{foo: 'bar' } ]// ]});

Where "options" is an object with the following properties:

constserver=newRedisStreamsAggregator({host: "192.168.0.1",port: 6379});

• ready - Connected to Redis
• connect - A new socket has connected
• error - An error has been encountered (connection to Redis failed, etc)