feat: add Uint8Array support for zero-copy worker_threads transfer

packages/client/lib/RESP/decoder.spec.ts

@@ -75,13 +75,13 @@ describe('RESP Decoder', () => {
     toWrite: Buffer.from('_\r\n'),
     replies: [null]
   });
-  
+
   describe('Boolean', () => {
     test('true', {
       toWrite: Buffer.from('#t\r\n'),
       replies: [true]
     });
-  
+
     test('false', {
       toWrite: Buffer.from('#f\r\n'),
       replies: [false]
@@ -290,6 +290,30 @@ describe('RESP Decoder', () => {
       replies: [Buffer.from('OK')]
     });
 
+    test("Simple string 'OK' as Uint8Array", {
+      typeMapping: {
+        [RESP_TYPES.SIMPLE_STRING]: Uint8Array
+      },
+      toWrite: Buffer.from('+OK\r\n'),
+      replies: [new Uint8Array([79, 75])]
+    });
+
+    test("Blob string 'OK' as Uint8Array", {
+      typeMapping: {
+        [RESP_TYPES.BLOB_STRING]: Uint8Array
+      },
+      toWrite: Buffer.from('$2\r\nOK\r\n'),
+      replies: [new Uint8Array([79, 75])]
+    });
+
+    test("Verbatim string 'OK' as Uint8Array", {
+      typeMapping: {
+        [RESP_TYPES.VERBATIM_STRING]: Uint8Array
+      },
+      toWrite: Buffer.from('=6\r\ntxt:OK\r\n'),
+      replies: [new Uint8Array([79, 75])]
+    });
+
     test("'é'", {
       toWrite: Buffer.from('=6\r\ntxt:é\r\n'),
       replies: ['é']

packages/client/lib/RESP/decoder.ts

@@ -492,6 +492,12 @@ export class Decoder {
     }
 
     const slice = chunk.subarray(start, crlfIndex);
+
+    // Add Uint8Array support
+    if (type === Uint8Array) {
+      return new Uint8Array(slice.buffer, slice.byteOffset, slice.byteLength);
+    }
+
     return type === Buffer ?
       slice :
       slice.toString();
@@ -507,6 +513,12 @@ export class Decoder {
 
     chunks.push(chunk.subarray(start, crlfIndex));
     const buffer = Buffer.concat(chunks);
+
+    // Add Uint8Array support
+    if (type === Uint8Array) {
+      return new Uint8Array(buffer.buffer, buffer.byteOffset, buffer.byteLength);
+    }
+
     return type === Buffer ? buffer : buffer.toString();
   }
 
@@ -555,6 +567,12 @@ export class Decoder {
 
     const slice = chunk.subarray(this.#cursor, end);
     this.#cursor = end + skip;
+
+    // Add Uint8Array support
+    if (type === Uint8Array) {
+      return new Uint8Array(slice.buffer, slice.byteOffset, slice.byteLength);
+    }
+
     return type === Buffer ?
       slice :
       slice.toString();
@@ -578,6 +596,12 @@ export class Decoder {
     chunks.push(chunk.subarray(this.#cursor, end));
     this.#cursor = end + skip;
     const buffer = Buffer.concat(chunks);
+
+    // Add Uint8Array support
+    if (type === Uint8Array) {
+      return new Uint8Array(buffer.buffer, buffer.byteOffset, buffer.byteLength);
+    }
+
     return type === Buffer ? buffer : buffer.toString();
   }
 

packages/client/lib/RESP/encoder.spec.ts

@@ -30,4 +30,13 @@ describe('RESP Encoder', () => {
       ['*1\r\n$6\r\n', Buffer.from('string'), '\r\n']
     );
   });
+
+  it('uint8array', () => {
+    const uint8Array = new Uint8Array([115, 116, 114, 105, 110, 103]); // 'string'
+    assert.deepEqual(
+      encodeCommand([uint8Array]),
+      ['*1\r\n$6\r\n', Buffer.from('string'), '\r\n']
+    );
+  });
+
 });

packages/client/lib/RESP/encoder.ts

@@ -11,14 +11,21 @@ export default function encodeCommand(args: ReadonlyArray<RedisArgument>): Reado
     const arg = args[i];
     if (typeof arg === 'string') {
       strings += '$' + Buffer.byteLength(arg) + CRLF + arg + CRLF;
-    } else if (arg instanceof Buffer) {
+    } else if (arg instanceof Buffer) { // Check Buffer before Uint8Array since Buffer extends Uint8Array in Node.js
       toWrite.push(
         strings + '$' + arg.length.toString() + CRLF,
         arg
       );
       strings = CRLF;
+    } else if (arg instanceof Uint8Array) {
+      const buffer = Buffer.from(arg.buffer, arg.byteOffset, arg.byteLength);
+      toWrite.push(
+        strings + '$' + buffer.length.toString() + CRLF,
+        buffer
+      );
+      strings = CRLF;
     } else {
-      throw new TypeError(`"arguments[${i}]" must be of type "string | Buffer", got ${typeof arg} instead.`);
+      throw new TypeError(`"arguments[${i}]" must be of type "string | Buffer | Uint8Array", got ${typeof arg} instead.`);
     }
   }
 

packages/client/lib/RESP/types.spec.ts

@@ -44,6 +44,16 @@ describe('RESP Type Mapping', () => {
             Buffer.isBuffer(verbatimRes)
         );
 
+        const uint8Res = await client
+            .withTypeMapping({
+                [RESP_TYPES.BLOB_STRING]: Uint8Array
+            })
+            .get('key');
+        if (uint8Res !== null && typeof uint8Res !== 'string') {
+            assert.ok(ArrayBuffer.isView(uint8Res));
+            assert.ok((uint8Res as any) instanceof Uint8Array);
+        }
+
         // Recursive Collections
         // ARRAY infers nested mapped types
         const arrayRes = await client

packages/client/lib/RESP/types.ts

@@ -68,17 +68,17 @@ export interface SimpleStringReply<
 > extends RespType<
   RESP_TYPES['SIMPLE_STRING'],
   T,
-  Buffer,
-  string | Buffer
+  Buffer | Uint8Array,
+  string | Buffer | Uint8Array
 > { }
 
 export interface BlobStringReply<
   T extends string = string
 > extends RespType<
   RESP_TYPES['BLOB_STRING'],
   T,
-  Buffer,
-  string | Buffer
+  Buffer | Uint8Array,
+  string | Buffer | Uint8Array
 > {
   toString(): string
 }
@@ -88,8 +88,8 @@ export interface VerbatimStringReply<
 > extends RespType<
   RESP_TYPES['VERBATIM_STRING'],
   T,
-  Buffer | VerbatimString,
-  string | Buffer | VerbatimString
+  Buffer | Uint8Array | VerbatimString,
+  string | Buffer | Uint8Array | VerbatimString
 > { }
 
 export interface SimpleErrorReply extends RespType<
@@ -198,6 +198,8 @@ type UnwrapConstructor<T> =
   T extends NumberConstructor ? number :
   T extends BooleanConstructor ? boolean :
   T extends BigIntConstructor ? bigint :
+  T extends BufferConstructor ? Buffer :
+  T extends Uint8ArrayConstructor ? Uint8Array :
   T;
 export type UnwrapReply<REPLY extends RespType<any, any, any, any>> = REPLY['DEFAULT' | 'TYPES'];
 
@@ -217,7 +219,7 @@ export type ReplyWithTypeMapping<
       REPLY extends Set<infer T> ? Set<ReplyWithTypeMapping<T, TYPE_MAPPING>> :
       REPLY extends Map<infer K, infer V> ? Map<MapKey<K, TYPE_MAPPING>, ReplyWithTypeMapping<V, TYPE_MAPPING>> :
       // `Date | Buffer | Error` are supersets of `Record`, so they need to be checked first
-      REPLY extends Date | Buffer | Error ? REPLY :
+      REPLY extends Date | Buffer | Uint8Array | Error ? REPLY :
       REPLY extends Record<PropertyKey, any> ? {
         [P in keyof REPLY]: ReplyWithTypeMapping<REPLY[P], TYPE_MAPPING>;
       } :
@@ -228,7 +230,7 @@ export type ReplyWithTypeMapping<
 
 export type TransformReply = (this: void, reply: any, preserve?: any, typeMapping?: TypeMapping) => any; // TODO;
 
-export type RedisArgument = string | Buffer;
+export type RedisArgument = string | Buffer | Uint8Array;
 
 export type CommandArguments = Array<RedisArgument> & { preserve?: unknown };
 

packages/client/lib/cluster/cluster-slots.ts

@@ -682,7 +682,7 @@ export default class RedisClusterSlots<
       };
     }
 
-    const slotNumber = calculateSlot(firstKey);
+    const slotNumber = calculateSlot(firstKey as string | Buffer);
     if (!isReadonly) {
       return {
         client: await this.nodeClient(this.slots[slotNumber].master),

packages/client/lib/commands/GEOSEARCH.ts

@@ -32,7 +32,7 @@ export function parseGeoSearchArguments(
 ) {
   parser.pushKey(key);
 
-  if (typeof from === 'string' || from instanceof Buffer) {
+  if (typeof from === 'string' || from instanceof Buffer || from instanceof Uint8Array) {
     parser.push('FROMMEMBER', from);
   } else {
     parser.push('FROMLONLAT', from.longitude.toString(), from.latitude.toString());

packages/client/lib/commands/HSET.ts

@@ -31,7 +31,7 @@ export default {
     parser.push('HSET');
     parser.pushKey(key);
 
-    if (typeof value === 'string' || typeof value === 'number' || value instanceof Buffer) {
+    if (typeof value === 'string' || typeof value === 'number' || value instanceof Buffer || value instanceof Uint8Array) {
       parser.push(
         convertValue(value),
         convertValue(fieldValue!)

packages/search/lib/commands/AGGREGATE.ts

@@ -337,7 +337,7 @@ export function parseGroupByReducer(parser: CommandParser, reducer: GroupByReduc
 }
 
 function pushSortByProperty(args: Array<RedisArgument>, sortBy: SortByProperty) {
-  if (typeof sortBy === 'string' || sortBy instanceof Buffer) {
+  if (typeof sortBy === 'string' || sortBy instanceof Buffer || sortBy instanceof Uint8Array) {
     args.push(sortBy);
   } else {
     args.push(sortBy.BY);

packages/search/lib/commands/SEARCH.ts

@@ -134,7 +134,7 @@ export function parseSearchOptions(parser: CommandParser, options?: FtSearchOpti
   if (options?.SORTBY) {
     parser.push('SORTBY');
 
-    if (typeof options.SORTBY === 'string' || options.SORTBY instanceof Buffer) {
+    if (typeof options.SORTBY === 'string' || options.SORTBY instanceof Buffer || options.SORTBY instanceof Uint8Array) {
       parser.push(options.SORTBY);
     } else {
       parser.push(options.SORTBY.BY);

packages/time-series/lib/commands/helpers.ts

@@ -275,7 +275,7 @@ function transformRESP2Sources(sourcesRaw: BlobStringReply) {
     return unwrappedSources.split(',');
   }
 
-  const indexOfComma = unwrappedSources.indexOf(',');
+  const indexOfComma = unwrappedSources.indexOf(','.charCodeAt(0));
   if (indexOfComma === -1) {
     return [unwrappedSources];
   }
@@ -286,7 +286,7 @@ function transformRESP2Sources(sourcesRaw: BlobStringReply) {
 
   let previousComma = indexOfComma + 1;
   while (true) {
-    const indexOf = unwrappedSources.indexOf(',', previousComma);
+    const indexOf = unwrappedSources.indexOf(','.charCodeAt(0), previousComma);
     if (indexOf === -1) {
       sourcesArray.push(
         unwrappedSources.subarray(previousComma)