Question posted in
The official documentation can be found here.

Retrieving messages from redis stream

I have a NodeJS application that is using Redis stream (library ‘ioredis’) to pass information around. The problem is that when I add a message to a stream and I try to retrieve it, I have to go down a lot of Arrays level:

      const message = await redis.xreadgroup('GROUP', orderGroup, orderConsumer, 'COUNT', 1, 'STREAMS', orderStream, '>');

      const messageId: string  = message[0][1][0][0];
      const pMsg: Obj = JSON.parse(JSON.parse(message[0][1][0][1][1]));

This is how I create the stream:

    await redis.xgroup('CREATE', orderStream, orderGroup, '0', 'MKSTREAM')
    .catch((err) => {
      console.error(`Group already exists error: ${err}`);
    })

Is this normal? In the Redis doc (https://redis.io/commands/xreadgroup) it shows that the return value is an array with the id of the message at position 0 and the fields at position 1. I feel like I’m missing something…

Tags:

3

Answers


  1. 0

    Here is an example output of XREADGROUP, as you can see the values are at the nested level 5.

    127.0.0.1:6379> XREADGROUP Group g1 c1 COUNT 100 STREAMS s1 >
1) 1) "s1"
   2)  1) 1) "1608445334963-0"
          2) 1) "f1"
             2) "v1"
             3) "f2"
             4) "v2"
       2) 1) "1608445335464-0"
          2) 1) "f1"
             2) "v1"
             3) "f2"
             4) "v2"
       3) 1) "1608445335856-0"
          2) 1) "f1"
             2) "v1"
             3) "f2"
             4) "v2"

    For more details see https://redis.io/commands/xread

    Login or Signup to reply.
  2. 0

    It is normal and expected. XREADGROUP supports reading from multiples stream keys, multiple messages, and messages can have multiple field-value pairs.

    Follow the next example:

    > XGROUP CREATE mystream1 mygroup 0 MKSTREAM
OK
> XGROUP CREATE mystream2 mygroup 0 MKSTREAM
OK
> XADD mystream1 * field1 value1 field2 value2
"1608444656005-0"
> XADD mystream1 * field1 value3 field2 value4
"1608444660566-0"
> XADD mystream2 * field3 value5 field4 value6
"1608444665238-0"
> XADD mystream2 * field3 value7 field4 value8
"1608444670070-0"
> XREADGROUP GROUP mygroup yo COUNT 2 STREAMS mystream1 mystream2 > >
1) 1) "mystream1"
   2) 1) 1) "1608444656005-0"
         2) 1) "field1"
            2) "value1"
            3) "field2"
            4) "value2"
      2) 1) "1608444660566-0"
         2) 1) "field1"
            2) "value3"
            3) "field2"
            4) "value4"
2) 1) "mystream2"
   2) 1) 1) "1608444665238-0"
         2) 1) "field3"
            2) "value5"
            3) "field4"
            4) "value6"
      2) 1) "1608444670070-0"
         2) 1) "field3"
            2) "value7"
            3) "field4"
            4) "value8"

    The structure you get has multiple nested arrays. Using 0-indexed as in node:

    [index of the stream key]
  [0: the key name or 1: an array for messages]
    [index of the message]
      [0: the message ID or 1: an array for fields & values]
        [even for field name or odd for value]
    Login or Signup to reply.
  3. 0

    Where data[0][1] is the root level array (adjust this entry point for your own use).

    Variables

    rd: Return Data

    el: Element

    sel: Sub-element

    rel: Relative-element

    p: Relative-Object

    c: Iterate-Counter for deciding if it is a key or value.

    var rd = []
for(var el of data[0][1]){
  var sel = el[1] 
  var p = {}
  var c = 0
  for(var rel of sel){
    if(c % 2 == 0){
        // Right here is where the return object keys/values are set.
        p[rel] = sel[c + 1]
    }
    c++
  }
  rd.push(p)
}
return rd
    Login or Signup to reply.
Please signup or login to give your own answer.

Back To Top
Search