Node: add XLEN command

node/src/BaseClient.ts

@@ -99,6 +99,7 @@ import {
     createZRemRangeByScore,
     createZScore,
     createSUnionStore,
+    createXLen,
 } from "./Commands";
 import {
     ClosingError,
@@ -2271,6 +2272,26 @@ export class BaseClient {
         return this.createWritePromise(createXRead(keys_and_ids, options));
     }
 
+    /**
+     * Returns the number of entries in the stream stored at `key`.
+     *
+     * See https://valkey.io/commands/xlen/ for more details.
+     *
+     * @param key - The key of the stream.
+     * @returns The number of entries in the stream. If `key` does not exist, returns `0`.
+     *
+     * @example
+     * ```typescript
+     * const numEntries = await client.xlen("my_stream");
+     * console.log(numEntries); // Output: 2 - "my_stream" contains 2 entries.
+     * ```
+     */
+    public xlen(
+        key: string,
+    ): Promise<number> {
+        return this.createWritePromise(createXLen(key));
+    }
+
     private readonly MAP_READ_FROM_STRATEGY: Record<
         ReadFrom,
         connection_request.ReadFrom

node/src/Commands.ts

@@ -1418,6 +1418,13 @@ export function createXRead(
     return createCommand(RequestType.XRead, args);
 }
 
+/**
+ * @internal
+ */
+export function createXLen(key: string): redis_request.Command {
+    return createCommand(RequestType.XLen, [key]);
+}
+
 /**
  * @internal
  */

node/src/Transaction.ts

@@ -104,6 +104,7 @@ import {
     createZRemRangeByScore,
     createZScore,
     createSUnionStore,
+    createXLen,
 } from "./Commands";
 import { redis_request } from "./ProtobufMessage";
 
@@ -1346,6 +1347,19 @@ export class BaseTransaction<T extends BaseTransaction<T>> {
         return this.addAndReturn(createXRead(keys_and_ids, options));
     }
 
+    /**
+     * Returns the number of entries in the stream stored at `key`.
+     *
+     * See https://valkey.io/commands/xlen/ for more details.
+     *
+     * @param key - The key of the stream.
+     *
+     * Command Response - The number of entries in the stream. If `key` does not exist, returns `0`.
+     */
+    public xlen(key: string): T {
+        return this.addAndReturn(createXLen(key));
+    }
+
     /**
      * Renames `key` to `newkey`.
      * If `newkey` already exists it is overwritten.

node/tests/SharedTests.ts

@@ -2467,10 +2467,12 @@ export function runBaseTests<Context>(config: {
     );
 
     it.each([ProtocolVersion.RESP2, ProtocolVersion.RESP3])(
-        `streams add and trim test_%p`,
-        async () => {
+        `streams add, trim, and len test_%p`,
+        async (protocol) => {
             await runTest(async (client: BaseClient) => {
                 const key = uuidv4();
+                const nonExistingKey = uuidv4();
+                const stringKey = uuidv4();
                 const field1 = uuidv4();
                 const field2 = uuidv4();
 
@@ -2501,7 +2503,7 @@ export function runBaseTests<Context>(config: {
                         [field2, "bar2"],
                     ]),
                 ).not.toBeNull();
-                expect(await client.customCommand(["XLEN", key])).toEqual(2);
+                expect(await client.xlen(key)).toEqual(2);
 
                 // this will trim the first entry.
                 const id = await client.xadd(
@@ -2519,7 +2521,7 @@ export function runBaseTests<Context>(config: {
                     },
                 );
                 expect(id).not.toBeNull();
-                expect(await client.customCommand(["XLEN", key])).toEqual(2);
+                expect(await client.xlen(key)).toEqual(2);
 
                 // this will trim the 2nd entry.
                 expect(
@@ -2538,7 +2540,7 @@ export function runBaseTests<Context>(config: {
                         },
                     ),
                 ).not.toBeNull();
-                expect(await client.customCommand(["XLEN", key])).toEqual(2);
+                expect(await client.xlen(key)).toEqual(2);
 
                 expect(
                     await client.xtrim(key, {
@@ -2547,8 +2549,39 @@ export function runBaseTests<Context>(config: {
                         exact: true,
                     }),
                 ).toEqual(1);
-                expect(await client.customCommand(["XLEN", key])).toEqual(1);
-            }, ProtocolVersion.RESP2);
+                expect(await client.xlen(key)).toEqual(1);
+
+                expect(
+                    await client.xtrim(key, {
+                        method: "maxlen",
+                        threshold: 0,
+                        exact: true,
+                    }),
+                ).toEqual(1);
+                // Unlike other Redis collection types, stream keys still exist even after removing all entries
+                expect(await client.exists([key])).toEqual(1);
+                expect(await client.xlen(key)).toEqual(0);
+
+                expect(
+                    await client.xtrim(nonExistingKey, {
+                        method: "maxlen",
+                        threshold: 1,
+                        exact: true,
+                    }),
+                ).toEqual(0);
+                expect(await client.xlen(nonExistingKey)).toEqual(0);
+
+                // key exists, but it is not a stream
+                expect(await client.set(stringKey, "foo")).toEqual("OK");
+                await expect(
+                    client.xtrim(stringKey, {
+                        method: "maxlen",
+                        threshold: 1,
+                        exact: true,
+                    }),
+                ).rejects.toThrow();
+                await expect(client.xlen(stringKey)).rejects.toThrow();
+            }, protocol);
         },
         config.timeout,
     );

node/tests/TestUtilities.ts

@@ -387,6 +387,8 @@ export async function transactionTest(
     args.push("0-2");
     baseTransaction.xadd(key9, [["field", "value3"]], { id: "0-3" });
     args.push("0-3");
+    baseTransaction.xlen(key9);
+    args.push(3);
     baseTransaction.xread({ [key9]: "0-1" });
     args.push({
         [key9]: {