Feat: add xadd, xlen, xrange, xrevrange and xread for stream support

[critocrito]

Support basic stream operations (https://redis.io/topics/streams-intro) for
the upcoming Redis 5 release. Some features are missing, such as consumer
groups, capped streams or some of the auxiliary commands.

Event ids are auto incremented integers in the form of 1-0, 2-0,
etc. Events are stored internally as an array of arrays and for every event an
additional record is kept whether the event has been already polled or not.

const redis = new MockRedis({
  mystream: [
    ["1-0", ["key1", "val1", "key2", "val2"]],
    ["2-0", ["key1", "val1", "key2", "val2"]]
  ],
  "stream:mystream:1-0": {polled: false},
  "stream:mystream:2-0": {polled: false},
})

To poll events in a blocking manner is supported and the poll happens every
100ms.

// This will resolve once a new event becomes available.
redis.xread("BLOCK", "0", "STREAMS", "mystream", "$").then(console.log);
// After 1 second add a new event, that will resolve the waiting `XREAD`
// command.
setTimeout(() => redis.xadd("mystream", "*", "key", "val"), 1000);

ioredis doesn't support Redis streams yet, but there is pull request for redis-commands to add support. In the meantime ioredis' createBuiltinCommand can be used to add support manually:

import Redis from "ioredis";

const {string: xadd} = Redis.prototype.createBuiltinCommand("xadd");
const {string: xread} = Redis.prototype.createBuiltinCommand("xread");
Redis.prototype.xadd = xadd;
Redis.prototype.xread = xread;
const redis = new Redis();

[stipsan]

Hey thanks for this! I'll merge and prepare a release 😄

[erulabs]

@critocrito Heyo! I can't seem to get this working on redis 5.0-rc3 and redis unstable, and it seems like the client stays blocked on the XREAD and the XADD is never sent to the Redis server. info clients shows a blocked_client, and command logging shows that I don't actually ever get the XADD. Is there some non-mocked trick to getting this working? My understanding is that XREAD is intentionally blocking, and there is no equivalent to "UNSUBSCRIBE" for streams, so in this example case we would need two redis handles (for adding and reading accordingly)

Note: The example / ioredis code all works perfectly fine (against a real redis5.0-rc3), but two Redis handles are required.

[critocrito]

Hello @erulabs, I'm not sure I exactly understand where your problem lies. My application has a discrete process that pushes events on a stream, and a discreet process that polls events from a stream. Those processes really operate independently from each other, so yes, they do have separate handles to Redis.

You can use the same handle within the same process if you use asynchronous IO using Promises. I can only point you to examples how to use this code: the unit tests of ioredis-mock and here again some unit tests in some other project that I work on.

[erulabs]

Just that the example code above of:

// This will resolve once a new event becomes available.
redis.xread("BLOCK", "0", "STREAMS", "mystream", "$").then(console.log);
// After 1 second add a new event, that will resolve the waiting `XREAD`
// command.
setTimeout(() => redis.xadd("mystream", "*", "key", "val"), 1000);

will not work against a real, running redis 5 service, since the xadd will never be executed, as the xread is blocking the redis connection. It doesn't block the node.js thread, so the xadd javascript function is called, but the real redis server would not get that call, and thus the xread promise would not resolve. If, from another process, you executed an xadd, then the xread would resolve.

I can provide some example code if you want (or maybe a container so its easy to replicate against a running redis service), but just wanted to point out that the example and the mock test would not work in the real world.

[critocrito]

Ahh I see. Yeah, the above code snippet relates to a unit test using ioredis-mock. I have never tried the exact same code on a real redis client. Thanks for pointing it out. My real world use case doesn't push and pull to the same stream within the same process or on the same client handle.

[erulabs]

No problem - thanks for the library! Worth noting that I brought this up on the redis mailing lists and Antirez himself came to the rescue, and we now have a "CLIENT UNBLOCK" syntax (here is the discussion: https://groups.google.com/forum/#!topic/redis-db/bfgZ-3QgCmM)! Doesn't exactly resolve the syntax above, but it's a very nice feature!

[stipsan]

Thanks for the detailed info @erulabs and @critocrito!

I would love for this to behave more accurately. The whole point of this library is to be as accurate in its emulation as possible. Right now there is no emulation of things like network connectivity, sentinels or pub/sub. I'll make an issue linking to the information above to ensure it'll be handled correctly in the future.