fix: handle added stream group creation

Author: calebboyd

.vscode/settings.json

@@ -32,4 +32,4 @@
   "editor.codeActionsOnSave": {
     "source.fixAll.eslint": true
   }
-}
+}

package-lock.json

@@ -23,8 +23,8 @@
     "commitlint": "commitlint -f $(git rev-list --tags --max-count=1) -t $(git log --pretty=%H | head -1)",
     "lint": "eslint \"src/**/*\" --fix --ext \"*.ts\"",
     "prepublishOnly": "npm run build",
-    "coverage": "npm run test -- --collectCoverage",
-    "test": "npm run commitlint && vitest run",
+    "coverage": "npm run test -- --coverage",
+    "test": "npm run commitlint && vitest run --no-isolate",
     "watch:test": "vitest",
     "watch:typescript": "tsc -w"
   },
@@ -50,10 +50,13 @@
     "@calebboyd/semantic-release-config": "^1.0.1",
     "@commitlint/cli": "^17.3.0",
     "@commitlint/config-angular": "^17.3.0",
+    "@types/chance": "^1.1.3",
     "@types/debug": "^4.1.7",
     "@types/node": "^18.11.9",
-    "@typescript-eslint/eslint-plugin": "^5.44.0",
-    "@typescript-eslint/parser": "^5.44.0",
+    "@typescript-eslint/eslint-plugin": "^5.45.0",
+    "@typescript-eslint/parser": "^5.45.0",
+    "@vitest/coverage-c8": "^0.25.3",
+    "chance": "^1.1.9",
     "eslint": "^8.28.0",
     "eslint-config-prettier": "~8.5.0",
     "eslint-plugin-prettier": "^4.2.1",
@@ -110,7 +113,16 @@
       "*.d.ts"
     ],
     "rules": {
-      "@typescript-eslint/no-use-before-define": 0
+      "@typescript-eslint/no-use-before-define": 0,
+      "@typescript-eslint/no-unused-vars": [
+        "error",
+        {
+          "ignoreRestSiblings": true,
+          "destructuredArrayIgnorePattern": "^_",
+          "varsIgnorePattern": "^_",
+          "argsIgnorePattern": "^_"
+        }
+      ]
     }
   },
   "contributors": [

package.json

@@ -1,7 +1,7 @@
 import Redis, { ChainableCommander, RedisOptions } from 'ioredis'
 import { RedisStream } from './stream.js'
 import mkDebug from 'debug'
-import { XStreamResult, env } from './types.js'
+import { XStreamResult, env, RedisStreamOptions } from './types.js'
 
 const debug = mkDebug('redis-x-stream')
 
@@ -67,7 +67,17 @@ export function ack(
   pendingAcks.clear()
 }
 
-function xgroup(client: ChainableCommander, { group, streams, first }: RedisStream): void {
+function xgroup(client: ChainableCommander, stream: RedisStream): void {
+  const { group, streams, first, addedStreams } = stream
+  if (addedStreams) {
+    for (const [key, start] of addedStreams) {
+      streams.set(key, start)
+      if (group && !first) {
+        client.xgroup('CREATE', key, group, start, 'MKSTREAM')
+      }
+    }
+    stream.addedStreams = null
+  }
   if (!first || !group) return
   for (const [key, start] of streams) {
     debug(`xgroup create ${key} ${group} ${start} mkstream`)
@@ -104,10 +114,25 @@ function xreadgroup(
   ;(client as KindaAny)[buffers ? 'xreadgroupBuffer' : 'xreadgroup'](...args)
 }
 
+export function* initStreams(
+  streams: RedisStreamOptions['streams'] | string
+): Iterable<[string, string]> {
+  if (typeof streams === 'string') streams = [streams]
+  if (Array.isArray(streams)) {
+    for (const stream of streams) {
+      yield [stream, '0']
+    }
+  } else if (streams) {
+    yield* Object.entries(streams)
+  } else {
+    return
+  }
+}
+
 export function createClient(options?: Redis | string | RedisOptions) {
   let client: Redis,
     created = true
-  if (typeof options === 'object') {
+  if (options && typeof options === 'object') {
     if ('pipeline' in options) {
       client = options
       created = false

src/redis.ts

@@ -1,7 +1,7 @@
 import { describe, it, expect } from 'vitest'
 import Redis from 'ioredis'
-import redisStream from './stream'
 import { rand } from './test.util.spec'
+import redisStream from './stream'
 
 //eslint-disable-next-line @typescript-eslint/no-explicit-any
 type JustForTests = any

src/stream.spec.ts

@@ -1,5 +1,5 @@
 import { hostname } from 'node:os'
-import { ack, createClient, readAckDelete } from './redis.js'
+import { ack, createClient, initStreams, readAckDelete } from './redis.js'
 import {
   RedisStreamOptions,
   RedisClient,
@@ -11,6 +11,7 @@ import {
 export { RedisStreamOptions }
 
 const allowedKeys = {
+  stream: 1,
   streams: 1,
   group: 1,
   consumer: 1,
@@ -57,8 +58,9 @@ export class RedisStream {
   public done = false
   public first = false
   public draining = false
+  public reading = false
+  public addedStreams: null | Iterable<[string, string]> = null
   private unblocked = false
-  private reading = false
 
   private readerId: number | null = null
   private pendingId: Promise<number | null> | null = null
@@ -94,6 +96,12 @@ export class RedisStream {
       this.block = options.block
     }
 
+    this.streams = new Map([
+      ...initStreams(streams),
+      ...initStreams(options.streams),
+      ...initStreams(options.stream),
+    ])
+
     if (this.block === 0 || this.block === Infinity) {
       this.blocked = true
       const { client, created } = createClient(options.redisControl)
@@ -128,14 +136,6 @@ export class RedisStream {
       this.first = true
     }
 
-    if (Array.isArray(options.streams)) {
-      this.streams = options.streams.reduce((keys, key) => {
-        return keys.set(key, '0')
-      }, new Map<string, string>())
-    } else {
-      this.streams = new Map(Object.entries(options.streams))
-    }
-
     if (typeof options.count === 'number') {
       this.count = options.count
     }
@@ -247,8 +247,10 @@ export class RedisStream {
     }
   }
 
-  public async addStream(streamName: string) {
-    this.streams.set(streamName, '0')
+  public async addStream(streams: RedisStreamOptions['streams'] | string) {
+    this.addedStreams = this.addedStreams
+      ? [...this.addedStreams, ...initStreams(streams)]
+      : initStreams(streams)
     await this.maybeUnblock()
   }
 

src/stream.ts

@@ -1,7 +1,13 @@
 import { RedisClient, StreamEntry, XEntryResult } from './types.js'
+import { afterAll } from 'vitest'
 import mkDebug from 'debug'
+import { randomInt } from 'node:crypto'
+import Chance from 'chance'
 
-const debug = mkDebug('redis-x-stream')
+const seed = Number(process.env.TEST_SEED) || randomInt(Date.now())
+const chance = new Chance(seed)
+
+const debug = mkDebug('test-redis-x-stream')
 
 const delay = (ms: number): Promise<void> => new Promise((resolve) => setTimeout(resolve, ms)),
   times = <T>(count: number, fn: (_: undefined, i: number) => T): Array<T> =>
@@ -10,8 +16,8 @@ const delay = (ms: number): Promise<void> => new Promise((resolve) => setTimeout
     await client.quit()
     return new Promise((resolve) => client.once('end', resolve))
   },
-  randNum = (min: number, max: number) => Math.floor(Math.random() * (max - min) + min),
-  rand = (): string => Math.random().toString(36).slice(6),
+  randNum = (min: number, max: number) => chance.integer({ min, max }),
+  rand = (): string => chance.string({}),
   drain = async (iterable: AsyncIterable<XEntryResult>): Promise<Map<string, StreamEntry[]>> => {
     const results = new Map<string, StreamEntry[]>()
     for await (const [streamName, entry] of iterable) {
@@ -34,4 +40,8 @@ async function hydrateForTest(writer: RedisClient, stream: string, ...values: st
   return values
 }
 
-export { delay, times, quit, hydrateForTest, rand, testEntries, redisIdRegex, drain }
+afterAll(() => {
+  console.log(`Seed set to: ${seed}`)
+})
+
+export { delay, times, quit, hydrateForTest, rand, randNum, testEntries, redisIdRegex, drain }

src/test.util.spec.ts

@@ -27,10 +27,16 @@ if (typeof process !== 'undefined' && process.env) {
   env.REDIS_X_STREAM_URL = process.env.REDIS_X_STREAM_URL
 }
 export interface RedisStreamOptions {
+  /**
+   * Redis stream keys to be read. If a Record is provided each value is the starting id for that stream
+   *
+   *  *alias* for stream
+   */
+  streams?: string[] | Record<string, string> | string
   /**
    * Redis stream keys to be read. If a Record is provided each value is the starting id for that stream
    */
-  streams: string[] | Record<string, string>
+  stream?: string[] | Record<string, string> | string
   /**
    * The consumer group.
    * Note: if only a group is provided a consumer is created automatically

src/types.ts

@@ -1,8 +1,15 @@
 import Redis from 'ioredis'
 import { describe, it, expect, beforeAll, afterAll, beforeEach, afterEach } from 'vitest'
+import {
+  delay,
+  hydrateForTest,
+  quit,
+  testEntries,
+  redisIdRegex,
+  randNum,
+} from './test.util.spec.js'
 import { RedisStream } from './stream.js'
 import { RedisClient } from './types.js'
-import { delay, hydrateForTest, quit, times, testEntries, redisIdRegex } from './test.util.spec.js'
 
 describe('redis-x-stream xread', () => {
   let writer!: RedisClient, reader: RedisClient, prefix: string
@@ -28,8 +35,8 @@ describe('redis-x-stream xread', () => {
     return quit(reader)
   })
   it('should dispense entries', async () => {
-    const streamName = key('my-straam'),
-      iterable = new RedisStream(streamName)
+    const streamName = key('my-stream'),
+      iterable = new RedisStream({ stream: streamName, count: randNum(1, 20) })
     await hydrateForTest(writer, streamName, ...testEntries)
     let entryIdx = 0,
       asserted = false
@@ -47,33 +54,25 @@ describe('redis-x-stream xread', () => {
   it('should block waiting for new entries', async () => {
     let entries = 0
     const streamName = key('my-stream'),
-      redisIdRegex = /\d+-\d/,
       block = 200,
       iterable = new RedisStream({
         block,
+        count: randNum(1, 50),
         streams: [streamName],
-      }),
-      hydrate = () => hydrateForTest(writer, streamName),
-      iterate = async () => {
-        for await (const [str, entry] of iterable) {
-          entries++
-          expect(str).toEqual(streamName)
-          expect(entry[0]).toMatch(redisIdRegex)
-        }
-      }
+      })
+    const hydrate = () => hydrateForTest(writer, streamName)
     await hydrate()
-    const consuming = iterate()
-    await delay(block)
-    times(9, hydrate)
-    await delay(block)
-    times(10, hydrate)
-    await consuming
-    expect(entries).toEqual(testEntries.length * 20)
+    for await (const _ of iterable) {
+      if (entries++ === testEntries.length - 1) {
+        delay(block - 20).then(hydrate)
+      }
+    }
+    expect(entries).toEqual(testEntries.length * 2)
   })
 
   it('should throw if ack is called without a group or consumer', async () => {
     const streamName = key('my-stream')
-    const stream = new RedisStream(streamName)
+    const stream = new RedisStream({ stream: [streamName], count: randNum(1, 500) })
     const values = await hydrateForTest(writer, streamName)
     let ackAttempts = 0
     for await (const [_, [id]] of stream) {
@@ -96,25 +95,29 @@ describe('redis-x-stream xread', () => {
     const laterStream = key('later-stream')
     await hydrateForTest(writer, myStream)
     await hydrateForTest(writer, laterStream)
-    const stream = new RedisStream({ streams: [myStream], block: Infinity })
+    const stream = new RedisStream({
+      streams: [myStream],
+      block: Infinity,
+      count: randNum(300, 400),
+    })
     let i = 0
     for await (const [streamName, _] of stream) {
       i++
       if (i === testEntries.length) {
         expect(streamName).toEqual(myStream)
         setTimeout(() => {
-          //TODO: expect stream is blocked?
+          expect(stream.reading).toBe(true)
           stream.addStream(laterStream)
         })
       }
       if (i > testEntries.length) {
         expect(streamName).toEqual(laterStream)
       }
-      if (i === testEntries.length * 2) {
+      if (i === testEntries.length * 2 - 1) {
         setTimeout(() => {
           i++
           stream.end() //break;
-        })
+        }, 100)
       }
     }
     //stream will block indefinitely (i++ in the future to assert after loop)
@@ -123,7 +126,7 @@ describe('redis-x-stream xread', () => {
 
   it('should not allow re-iteration (done is set)', async () => {
     const streamName = key('my-stream'),
-      iterable = new RedisStream(streamName)
+      iterable = new RedisStream({ stream: streamName, count: randNum(1, 2) })
     let entries = 0
     const values = await hydrateForTest(writer, streamName)
     const iterate = async () => {

src/xread.spec.ts

@@ -1,9 +1,9 @@
+import Redis from 'ioredis'
 import { describe, it, expect, beforeAll, afterAll, beforeEach, afterEach } from 'vitest'
-import redisStream, { RedisStream } from './stream.js'
 import { hostname } from 'os'
 import { drain, hydrateForTest, quit, rand, redisIdRegex, testEntries } from './test.util.spec.js'
+import redisStream, { RedisStream } from './stream.js'
 import { RedisClient } from './types.js'
-import Redis from 'ioredis'
 
 describe('redis-x-stream xreadgroup', () => {
   let writer!: RedisClient, reader: RedisClient, prefix: string
@@ -143,7 +143,7 @@ describe('redis-x-stream xreadgroup', () => {
       deleteOnAck: true,
     })
     let i = 0
-    for await (const [streamName, [id, keyvals]] of stream) {
+    for await (const _ of stream) {
       if (++i === 3) {
         void stream.drain()
       }