Ioredis: How can you "subscribe" to a Redis stream (5.0)?

[Ventis]

Hi,

I've implemented some of the commands to add (XADD) to and read (XREAD) from a Redis stream. However, what I'm not entirely clear about is how you can use these commands to subscribe to a stream like you would in PubSub?

For example, to get every new message that is added to a stream, should I send the XREAD command on an interval? Like so:

setInterval(async () => {
      const command = new Redis.Command('XREAD', ['STREAMS', stream, lastId || '0']);
      const events = await this.redis.sendCommand(command);
      console.log('Redis events received: ', events);
      if (events) cb(events[0]);
    }, 1000);

[matthax]

This is a feature I would be extremely interested in seeing. When I have some free time I will take a look to see what I can dig up - maybe the solution is as simple as a simple interval.

If enough people are interested maybe we could have this feature sponsored?

[matthax]

So I ended up having to do this when writing something for a test project.

I took what I had done and I've rewritten it and dropped a few of the features so that I could share it here, take a look at let me know if this works for you:

https://gist.github.com/matthax/d87c0b1bc0d381c7549f89cbc3bd1941

[vflopes]

I've developed this script to add stream commands to ioredis (I have this project that uses Redis streams: warshipjs):

'use strict';
const IORedis = require('ioredis');

const parseObjectResponse = (reply, customParser = null) => {
    if (!Array.isArray(reply))
        return reply;
    const data = {};
    for (let i = 0; i < reply.length; i += 2) {
        if (customParser) {
            data[reply[i]] = customParser(reply[i], reply[i+1]);
            continue;
        }
        data[reply[i]] = reply[i+1];
    }
    return data;
};

const parseMessageResponse = (reply) => {
    if (!Array.isArray(reply))
        return [];
    return reply.map((message) => {
        return {id:message[0], data:parseObjectResponse(message[1])};
    });
};

const parseStreamResponse = (reply) => {
    if (!Array.isArray(reply))
        return reply;
    const object = {};
    for (const stream of reply)
        object[stream[0]] = parseMessageResponse(stream[1]);
    return object;
};

const addCommand = {
    xgroup:(target) => target.Command.setReplyTransformer('xgroup', (reply) => reply),
    xadd:(target) => target.Command.setReplyTransformer('xadd', (reply) => reply),
    xread:(target) => target.Command.setReplyTransformer('xread', parseStreamResponse),
    xpending:(target) => target.Command.setReplyTransformer('xpending', (reply) => {
        if (!reply || reply.length === 0)
            return [];
        if (reply.length === 4 && !isNaN(reply[0]))
            return {
                count:parseInt(reply[0]),
                minId:reply[1],
                maxId:reply[2],
                consumers:(reply[3] || []).map((consumer) => {
                    return {
                        name:consumer[0],
                        count:parseInt(consumer[1])
                    };
                })
            };
        return reply.map((message) => {
            return {
                id:message[0],
                consumerName:message[1],
                elapsedMilliseconds:parseInt(message[2]),
                deliveryCount:parseInt(message[3])
            };
        });
    }),
    xreadgroup:(target) => target.Command.setReplyTransformer('xreadgroup', parseStreamResponse),
    xrange:(target) => target.Command.setReplyTransformer('xrange', parseMessageResponse),
    xrevrange:(target) => target.Command.setReplyTransformer('xrevrange', parseMessageResponse),
    xclaim:(target) => target.Command.setReplyTransformer('xclaim', parseMessageResponse),
    xinfo:(target) => target.Command.setReplyTransformer('xinfo', (reply) => parseObjectResponse(reply, (key, value) => {
        switch (key) {
        case 'first-entry':
        case 'last-entry':
            if (!Array.isArray(value))
                return value;
            return {
                id:value[0],
                data:parseObjectResponse(value[1])
            };
        default:
            return value;
        }
    })),
    xack:(target) => target.Command.setReplyTransformer('xack', (reply) => parseInt(reply)),
    xlen:(target) => target.Command.setReplyTransformer('xlen', (reply) => parseInt(reply)),
    xtrim:(target) => target.Command.setReplyTransformer('xtrim', (reply) => parseInt(reply)),
    xdel:(target) => target.Command.setReplyTransformer('xdel', (reply) => parseInt(reply))
};

let isPrepared = false;

module.exports = () => {

    if (isPrepared)
        return void 0;

    isPrepared = true;

    Object.keys(addCommand).forEach((command) => {
        const {string, buffer} = IORedis.prototype.createBuiltinCommand(command);
        IORedis.prototype[command] = string;
        IORedis.prototype[command+'Buffer'] = buffer;
        addCommand[command](IORedis);
    });

};

You just need to require it and execute the function before any call to instantiate a Redis os Cluster class.

require('./ioredis-streams.js')();
// now you have the x* commands

[stale[bot]]

This issue has been automatically marked as stale because it has not had recent activity. It will be closed after 7 days if no further activity occurs, but feel free to re-open a closed issue if needed.

[mcleishk]

I am new to ioredis - is there a reason the stream commands (https://redis.io/commands#stream) aren't directly supported line the other comments (such as redisClient.hset)?

[luin]

I didn't notice this issue before and I'm sorry for that. The support of stream is a must have feature in my opinion. Any ideas about the syntax is welcome 😄 .

[mcleishk]

@luin - any eta for when stream support will be added?

[luin]

@mcleishk Should be released before June 20.

[luin]

In the latest version (4.10.0), all stream-related methods are added. Parameter and reply formats of these methods are kept the same as their corresponding Redis commands:

const Redis = require('ioredis')
const redis = new Redis()

redis.xadd('mystream', '*', 'field', 'value')

Reply transformers are not applied for backward compatibility though, it's not hard to implement your own one. Here's a modification leaving out parts that is not necessary with 4.10.0 based on the awesome implementation from @vflopes :

const Redis = require('ioredis')

function parseObjectResponse(reply, customParser = null) {
  if (!Array.isArray(reply)) {
    return reply
  }
  const data = {}
  for (let i = 0; i < reply.length; i += 2) {
    if (customParser) {
      data[reply[i]] = customParser(reply[i], reply[i + 1])
      continue
    }
    data[reply[i]] = reply[i + 1]
  }
  return data
}

function parseMessageResponse(reply) {
  if (!Array.isArray(reply)) {
    return []
  }
  return reply.map((message) => {
    return {
      id: message[0],
      data: parseObjectResponse(message[1])
    }
  })
}

function parseStreamResponse(reply) {
  if (!Array.isArray(reply)) {
    return reply
  }
  const object = {}
  for (const stream of reply) {
    object[stream[0]] = parseMessageResponse(stream[1])
  }
  return object
}

const transformers = {
  xread: parseStreamResponse,
  xpending(reply) {
    if (!reply || reply.length === 0) {
      return []
    }
    if (reply.length === 4 && !isNaN(reply[0]))
      return {
        count: parseInt(reply[0]),
        minId: reply[1],
        maxId: reply[2],
        consumers: (reply[3] || []).map((consumer) => {
          return {
            name: consumer[0],
            count: parseInt(consumer[1])
          }
        })
      }
    return reply.map((message) => {
      return {
        id: message[0],
        consumerName: message[1],
        elapsedMilliseconds: parseInt(message[2]),
        deliveryCount: parseInt(message[3])
      }
    })
  },
  xreadgroup: parseStreamResponse,
  xrange: parseMessageResponse,
  xrevrange: parseMessageResponse,
  xclaim: parseMessageResponse,
  xinfo(reply) {
    return parseObjectResponse(reply, (key, value) => {
      switch (key) {
        case 'first-entry':
        case 'last-entry':
          if (!Array.isArray(value)) {
            return value
          }
          return {
            id: value[0],
              data: parseObjectResponse(value[1])
          }
          default:
            return value
      }
    })
  }
}

Object.keys(transformers).forEach((commandName) => {
  Redis.Command.setReplyTransformer(commandName, transformers[commandName])
})

For the question that @Ventis raised, you can use the block mode of XREAD to avoid fetching data periodically (The following example doesn't have the transformers above applied):

const Redis = require('ioredis')
const redis = new Redis()

async function subscribeStream(stream, listener) {
  let lastID = '$'

  while (true) {
    // Implement your own `try/catch` logic,
    // (For example, logging the errors and continue to the next loop)
    const reply = await redis.xread('BLOCK', '5000', 'COUNT', 100, 'STREAMS', stream, lastID)
    if (!reply) {
      continue
    }
    const results = reply[0][1]
    const {length} = results
    if (!results.length) {
      continue
    }
    listener(results)
    lastID = results[length - 1][0]
  }
}

subscribeStream('mystream', console.log)

[snekcz]

Hi,
just a question, command const reply = await redis.xread('BLOCK', '5000', 'COUNT', 100, 'STREAMS', stream, lastID) blocks the whole javascript event loop. Is it intentional/expected behaviour?

[hedgepigdaniel]

Hi,
just a question, command const reply = await redis.xread('BLOCK', '5000', 'COUNT', 100, 'STREAMS', stream, lastID) blocks the whole javascript event loop. Is it intentional/expected behaviour?

I just tried this, and it doesn't seem to block the event loop - other IO such as console.log works fine while the command is blocking. It does however block all other redis commands on the same client.

[snekcz]

But, it seems it affects http/websocket communication, at least. I have started WS/http server and reading from streams - during the Redis read block, the incoming ws/http communication is blocked also. Can you try this, please too?

[a-bowers]

But, it seems it affects http/websocket communication, at least. I have started WS/http server and reading from streams - during the Redis read block, the incoming ws/http communication is blocked also. Can you try this, please too?

Seeing what I think is the same issue. As soon as the first message hits the stream to be read, everything stops. The xread never happens or never finishes and my websocket no longer processes any new inputs.

[SalesFeed]

You should duplicate the Redis client. You need a separate connection for xread in blocking mode.

[a-bowers]

You should duplicate the Redis client. You need a separate connection for xread in blocking mode.

It is a separate client that does nothing but read from that one stream. The strange thing is that it seems to stop everything else, whether block is set to 0 or a timeout as in the example above.

EDIT: resolved, was an issue in our own code.