add command/command info and xread/xreadgroup to v5 support

Author: sjpotter

packages/client/lib/commands/COMMAND.ts

@@ -1,12 +1,13 @@
-// import { RedisCommandArguments } from '.';
-// import { CommandRawReply, CommandReply, transformCommandReply } from './generic-transformers';
+import { ArrayReply, Command, UnwrapReply } from '../RESP/types';
+import { CommandRawReply, CommandReply, transformCommandReply } from './generic-transformers';
 
-// export const IS_READ_ONLY = true;
-
-// export function transformArguments(): RedisCommandArguments {
-//     return ['COMMAND'];
-// }
-
-// export function transformReply(reply: Array<CommandRawReply>): Array<CommandReply> {
-//     return reply.map(transformCommandReply);
-// }
+export default {
+  IS_READ_ONLY: true,
+  transformArguments() {
+    return ['COMMAND'];
+  },
+  // TODO: This works, as we don't currently handle any of the items returned as a map
+  transformReply(reply: UnwrapReply<ArrayReply<CommandRawReply>>): Array<CommandReply> {
+    return reply.map(transformCommandReply);
+  }
+} as const satisfies Command;

packages/client/lib/commands/COMMAND_INFO.ts

@@ -1,12 +1,14 @@
-// import { RedisCommandArguments } from '.';
-// import { CommandRawReply, CommandReply, transformCommandReply } from './generic-transformers';
+import { ArrayReply, Command, UnwrapReply } from '../RESP/types';
+import { CommandRawReply, CommandReply, transformCommandReply } from './generic-transformers';
 
-// export const IS_READ_ONLY = true;
-
-// export function transformArguments(commands: Array<string>): RedisCommandArguments {
-//     return ['COMMAND', 'INFO', ...commands];
-// }
-
-// export function transformReply(reply: Array<CommandRawReply | null>): Array<CommandReply | null> {
-//     return reply.map(command => command ? transformCommandReply(command) : null);
-// }
+export default {
+  FIRST_KEY_INDEX: undefined,
+  IS_READ_ONLY: true,
+  transformArguments(commands: Array<string>) {
+    return ['COMMAND', 'INFO', ...commands];
+  },
+  // TODO: This works, as we don't currently handle any of the items returned as a map
+  transformReply(reply: UnwrapReply<ArrayReply<CommandRawReply>>): Array<CommandReply | null> {
+    return reply.map(command => command ? transformCommandReply(command) : null);
+  }
+} as const satisfies Command;

packages/client/lib/commands/XAUTOCLAIM.ts

@@ -1,13 +1,13 @@
 import { RedisArgument, TuplesReply, BlobStringReply, ArrayReply, NullReply, UnwrapReply, Command } from '../RESP/types';
-import { StreamMessageReply, transformStreamMessageNullReply } from './generic-transformers';
+import { StreamMessageRawReply, transformStreamMessageNullReply } from './generic-transformers';
 
 export interface XAutoClaimOptions {
   COUNT?: number;
 }
 
 export type XAutoClaimRawReply = TuplesReply<[
   nextId: BlobStringReply,
-  messages: ArrayReply<StreamMessageReply | NullReply>,
+  messages: ArrayReply<StreamMessageRawReply | NullReply>,
   deletedMessages: ArrayReply<BlobStringReply>
 ]>;
 

packages/client/lib/commands/XCLAIM.ts

@@ -1,5 +1,5 @@
 import { RedisArgument, ArrayReply, NullReply, UnwrapReply, Command } from '../RESP/types';
-import { RedisVariadicArgument, pushVariadicArguments, StreamMessageReply, transformStreamMessageNullReply } from './generic-transformers';
+import { RedisVariadicArgument, pushVariadicArguments, StreamMessageRawReply, transformStreamMessageNullReply } from './generic-transformers';
 
 export interface XClaimOptions {
   IDLE?: number;
@@ -50,7 +50,7 @@ export default {
 
     return args;
   },
-  transformReply(reply: UnwrapReply<ArrayReply<StreamMessageReply | NullReply>>) {
+  transformReply(reply: UnwrapReply<ArrayReply<StreamMessageRawReply | NullReply>>) {
     return reply.map(transformStreamMessageNullReply);
   }
 } as const satisfies Command;

packages/client/lib/commands/XRANGE.ts

@@ -1,5 +1,5 @@
 import { RedisArgument, ArrayReply, UnwrapReply, Command } from '../RESP/types';
-import { StreamMessageReply, transformStreamMessageReply } from './generic-transformers';
+import { StreamMessageRawReply, transformStreamMessageReply } from './generic-transformers';
 
 export interface XRangeOptions {
   COUNT?: number;
@@ -25,7 +25,7 @@ export default {
   FIRST_KEY_INDEX: 1,
   IS_READ_ONLY: true,
   transformArguments: transformXRangeArguments.bind(undefined, 'XRANGE'),
-  transformReply(reply: UnwrapReply<ArrayReply<StreamMessageReply>>) {
+  transformReply(reply: UnwrapReply<ArrayReply<StreamMessageRawReply>>) {
     return reply.map(transformStreamMessageReply);
   }
 } as const satisfies Command;

packages/client/lib/commands/XREAD.ts

@@ -1,4 +1,5 @@
 import { Command, RedisArgument } from '../RESP/types';
+import { transformStreamsMessagesReplyResp2, transformStreamsMessagesReplyResp3 } from './generic-transformers';
 
 export interface XReadStream {
   key: RedisArgument;
@@ -48,8 +49,9 @@ export default {
 
     return args;
   },
-  // export { transformStreamsMessagesReply as transformReply } from './generic-transformers';
-  // TODO
-  transformReply: undefined as unknown as () => unknown
+  transformReply: {
+    2: transformStreamsMessagesReplyResp2,
+    3: transformStreamsMessagesReplyResp3
+  }
 } as const satisfies Command;
 

packages/client/lib/commands/XREADGROUP.ts

@@ -1,4 +1,5 @@
 import { Command, RedisArgument } from '../RESP/types';
+import { transformStreamsMessagesReplyResp2, transformStreamsMessagesReplyResp3 } from './generic-transformers';
 import XREAD, { XReadStreams, pushXReadStreams } from './XREAD';
 
 export interface XReadGroupOptions {
@@ -40,7 +41,8 @@ export default {
 
     return args;
   },
-  // export { transformStreamsMessagesReply as transformReply } from './generic-transformers';
-  // TODO
-  transformReply: undefined as unknown as () => unknown
+  transformReply: {
+    2: transformStreamsMessagesReplyResp2,
+    3: transformStreamsMessagesReplyResp3
+  }
 } as const satisfies Command;

packages/client/lib/commands/generic-transformers.ts

@@ -88,20 +88,25 @@ export function transformTuplesReply(
   return message;
 }
 
-export type StreamMessageReply = TuplesReply<[
+export type StreamMessageRawReply = TuplesReply<[
   id: BlobStringReply,
   message: ArrayReply<BlobStringReply>
 ]>;
 
-export function transformStreamMessageReply(reply: StreamMessageReply) {
+export type StreamMessageReply = {
+  id: BlobStringReply,
+  message: Record<string, BlobStringReply>,
+};
+
+export function transformStreamMessageReply(reply: StreamMessageRawReply): StreamMessageReply {
   const [ id, message ] = reply as unknown as UnwrapReply<typeof reply>;
   return {
-    id,
+    id: id,
     message: transformTuplesReply(message)
   };
 }
 
-export function transformStreamMessageNullReply(reply: StreamMessageReply | NullReply) {
+export function transformStreamMessageNullReply(reply: StreamMessageRawReply | NullReply) {
   return isNullReply(reply) ? reply : transformStreamMessageReply(reply);
 }
 
@@ -446,3 +451,52 @@ function isPlainKey(key: RedisArgument | ZKeyAndWeight): key is RedisArgument {
 function isPlainKeys(keys: Array<RedisArgument> | Array<ZKeyAndWeight>): keys is Array<RedisArgument> {
   return isPlainKey(keys[0]);
 }
+
+export type StreamMessagesReply = Array<StreamMessageReply>;
+
+export type StreamsMessagesReply = Array<{
+  name: BlobStringReply | string;
+  messages: StreamMessagesReply;
+}> | null;
+
+export function transformStreamMessagesReply(reply: Array<any>): StreamMessagesReply {
+  return reply.map(transformStreamMessageReply);
+}
+
+export function transformStreamsMessagesReplyResp2(reply: Array<any> | null): StreamsMessagesReply | null {
+  if (reply === null) return null;
+
+  return reply.map(([name, rawMessages]) => ({
+      name,
+      messages: transformStreamMessagesReply(rawMessages)
+  }));
+}
+
+export function transformStreamsMessagesReplyResp3(reply: any | null): StreamsMessagesReply | null {
+  if (reply === null) return null;
+  
+  if (reply instanceof Map) {
+    const ret: StreamsMessagesReply = [];
+    for (const [name, rawMessages] of reply) {
+      ret.push({
+        name,
+        messages: transformStreamMessagesReply(rawMessages),
+      })
+    }
+
+    return ret;
+  } else if (reply instanceof Array) {
+    return transformStreamsMessagesReplyResp2(reply);
+  } else {
+    const ret: StreamsMessagesReply = [];
+    for (const [name, rawMessages] of Object.entries(reply)) {
+      const m = rawMessages as Array<any>;
+      ret.push({
+        name,
+        messages: transformStreamMessagesReply(m),
+      })
+    }
+    
+    return ret;
+  }
+}