fix(client): XCLAIM & XAUTOCLAIM after a TRIM might return nils

redis · Jul 21, 2023 · eb2c37e · eb2c37e
1 parent a7d5bc7
commit eb2c37e
Show file tree

Hide file tree

Showing 6 changed files with 158 additions and 15 deletions.
diff --git a/packages/client/lib/commands/XAUTOCLAIM.spec.ts b/packages/client/lib/commands/XAUTOCLAIM.spec.ts
@@ -23,7 +23,7 @@ describe('XAUTOCLAIM', () => {
         });
     });
 
-    testUtils.testWithClient('client.xAutoClaim', async client => {
+    testUtils.testWithClient('client.xAutoClaim without messages', async client => {
         await Promise.all([
             client.xGroupCreate('key', 'group', '$', {
                 MKSTREAM: true
@@ -39,4 +39,66 @@ describe('XAUTOCLAIM', () => {
             }
         );
     }, GLOBAL.SERVERS.OPEN);
+
+    testUtils.testWithClient('client.xAutoClaim with messages', async client => {
+        const [,,id,] = await Promise.all([
+            client.xGroupCreate('key', 'group', '$', {
+                MKSTREAM: true
+            }),
+            client.xGroupCreateConsumer('key', 'group', 'consumer'),
+            client.xAdd('key', '*', { foo: 'bar' }),
+            client.xReadGroup('group', 'consumer', { key: 'key', id: '>' })
+        ]);
+
+        assert.deepEqual(
+            await client.xAutoClaim('key', 'group', 'consumer', 1, '0-0'),
+            {
+                nextId: '0-0',
+                messages: [{
+                    id,
+                    message: Object.create(null, { 'foo': {
+                        value: 'bar',
+                        configurable: true,
+                        enumerable: true
+                    } })
+                }]
+            }
+        );
+    }, GLOBAL.SERVERS.OPEN);
+
+    testUtils.testWithClient('client.xAutoClaim with trimmed messages', async client => {
+        const [,,,,,id2,] = await Promise.all([
+            client.xGroupCreate('key', 'group', '$', {
+                MKSTREAM: true
+            }),
+            client.xGroupCreateConsumer('key', 'group', 'consumer'),
+            client.xAdd('key', '*', { foo: 'bar' }),
+            client.xReadGroup('group', 'consumer', { key: 'key', id: '>' }),
+            client.xTrim('key', 'MAXLEN', 0),
+            client.xAdd('key', '*', { bar: 'baz' }),
+            client.xReadGroup('group', 'consumer', { key: 'key', id: '>' }),
+        ]);
+
+        assert.deepEqual(
+            await client.xAutoClaim('key', 'group', 'consumer', 1, '0-0'),
+            {
+                nextId: '0-0',
+                messages: testUtils.isVersionGreaterThan([7, 0]) ? [{
+                    id: id2,
+                    message: Object.create(null, { 'bar': {
+                        value: 'baz',
+                        configurable: true,
+                        enumerable: true
+                    } })
+                }] : [null, {
+                    id: id2,
+                    message: Object.create(null, { 'bar': {
+                        value: 'baz',
+                        configurable: true,
+                        enumerable: true
+                    } })
+                }]
+            }
+        );
+    }, GLOBAL.SERVERS.OPEN);
 });
diff --git a/packages/client/lib/commands/XAUTOCLAIM.ts b/packages/client/lib/commands/XAUTOCLAIM.ts
@@ -1,5 +1,5 @@
 import { RedisCommandArgument, RedisCommandArguments } from '.';
-import { StreamMessagesReply, transformStreamMessagesReply } from './generic-transformers';
+import { StreamMessagesNullReply, transformStreamMessagesNullReply } from './generic-transformers';
 
 export const FIRST_KEY_INDEX = 1;
 
@@ -28,12 +28,12 @@ type XAutoClaimRawReply = [RedisCommandArgument, Array<any>];
 
 interface XAutoClaimReply {
     nextId: RedisCommandArgument;
-    messages: StreamMessagesReply;
+    messages: StreamMessagesNullReply;
 }
 
 export function transformReply(reply: XAutoClaimRawReply): XAutoClaimReply {
     return {
         nextId: reply[0],
-        messages: transformStreamMessagesReply(reply[1])
+        messages: transformStreamMessagesNullReply(reply[1])
     };
 }
diff --git a/packages/client/lib/commands/XCLAIM.spec.ts b/packages/client/lib/commands/XCLAIM.spec.ts
@@ -87,4 +87,44 @@ describe('XCLAIM', () => {
             []
         );
     }, GLOBAL.SERVERS.OPEN);
+
+    testUtils.testWithClient('client.xClaim with a message', async client => {
+        const [,,id,] = await Promise.all([
+            client.xGroupCreate('key', 'group', '$', {
+                MKSTREAM: true
+            }),
+            client.xGroupCreateConsumer('key', 'group', 'consumer'),
+            client.xAdd('key', '*', { foo: 'bar' }),
+            client.xReadGroup('group', 'consumer', { key: 'key', id: '>' })
+        ]);
+
+        assert.deepEqual(
+            await client.xClaim('key', 'group', 'consumer', 1, id),
+            [{
+                id,
+                message: Object.create(null, { 'foo': {
+                    value: 'bar',
+                    configurable: true,
+                    enumerable: true
+                } })
+            }]
+        );
+    }, GLOBAL.SERVERS.OPEN);
+
+    testUtils.testWithClient('client.xClaim with a trimmed message', async client => {
+        const [,,id,,,] = await Promise.all([
+            client.xGroupCreate('key', 'group', '$', {
+                MKSTREAM: true
+            }),
+            client.xGroupCreateConsumer('key', 'group', 'consumer'),
+            client.xAdd('key', '*', { foo: 'bar' }),
+            client.xReadGroup('group', 'consumer', { key: 'key', id: '>' }),
+            client.xTrim('key', 'MAXLEN', 0),
+        ]);
+
+        assert.deepEqual(
+            await client.xClaim('key', 'group', 'consumer', 1, id),
+            testUtils.isVersionGreaterThan([7, 0]) ? []: [null]
+        );
+    }, GLOBAL.SERVERS.OPEN);
 });
diff --git a/packages/client/lib/commands/XCLAIM.ts b/packages/client/lib/commands/XCLAIM.ts
@@ -45,4 +45,4 @@ export function transformArguments(
     return args;
 }
 
-export { transformStreamMessagesReply as transformReply } from './generic-transformers';
+export { transformStreamMessagesNullReply as transformReply } from './generic-transformers';
diff --git a/packages/client/lib/commands/generic-transformers.spec.ts b/packages/client/lib/commands/generic-transformers.spec.ts
@@ -9,6 +9,7 @@ import {
     transformStringNumberInfinityArgument,
     transformTuplesReply,
     transformStreamMessagesReply,
+    transformStreamMessagesNullReply,
     transformStreamsMessagesReply,
     transformSortedSetWithScoresReply,
     pushGeoCountArgument,
@@ -219,6 +220,38 @@ describe('Generic Transformers', () => {
         );
     });
 
+    it('transformStreamMessagesNullReply', () => {
+        assert.deepEqual(
+            transformStreamMessagesNullReply([null, ['0-0', ['0key', '0value']]]),
+            [null, {
+                id: '0-0',
+                message: Object.create(null, {
+                    '0key': {
+                        value: '0value',
+                        configurable: true,
+                        enumerable: true
+                    }
+                })
+            }]
+        );
+    });
+
+    it('transformStreamMessagesNullReply', () => {
+        assert.deepEqual(
+            transformStreamMessagesNullReply([null, ['0-1', ['11key', '11value']]]),
+            [null, {
+                id: '0-1',
+                message: Object.create(null, {
+                    '11key': {
+                        value: '11value',
+                        configurable: true,
+                        enumerable: true
+                    }
+                })
+            }]
+        );
+    });
+
     describe('transformStreamsMessagesReply', () => {
         it('null', () => {
             assert.equal(

diff --git a/packages/client/lib/commands/generic-transformers.ts b/packages/client/lib/commands/generic-transformers.ts
@@ -92,19 +92,27 @@ export interface StreamMessageReply {
     message: Record<string, RedisCommandArgument>;
 }
 
-export type StreamMessagesReply = Array<StreamMessageReply>;
+export function transformStreamMessageReply([id, message]: Array<any>): StreamMessageReply {
+    return {
+        id,
+        message: transformTuplesReply(message)
+    };
+}
 
-export function transformStreamMessagesReply(reply: Array<any>): StreamMessagesReply {
-    const messages = [];
+export function transformStreamMessageNullReply(reply: Array<any>): StreamMessageReply | null {
+    if (reply === null) return null;
+    return transformStreamMessageReply(reply);
+}
 
-    for (const [id, message] of reply) {
-        messages.push({
-            id,
-            message: transformTuplesReply(message)
-        });
-    }
 
-    return messages;
+export type StreamMessagesReply = Array<StreamMessageReply>;
+export function transformStreamMessagesReply(reply: Array<any>): StreamMessagesReply {
+    return reply.map(transformStreamMessageReply);
+}
+
+export type StreamMessagesNullReply = Array<StreamMessageReply | null>;
+export function transformStreamMessagesNullReply(reply: Array<any>): StreamMessagesNullReply {
+    return reply.map(transformStreamMessageNullReply);
 }
 
 export type StreamsMessagesReply = Array<{