feat(json-pack): 🎸 decode command in streaming decoder

Author: streamich

src/json-pack/resp/RespStreamingDecoder.ts

@@ -67,6 +67,29 @@ export class RespStreamingDecoder {
     }
   }
 
+  /**
+   * Decode only one RESP command from the stream, if the value is not a
+   * command, an error will be thrown.
+   *
+   * @returns Redis command and its arguments or `undefined` if there is
+   * not enough data to decode.
+   */
+  public readCmd(): [cmd: string, ...args: Uint8Array[]] | undefined {
+    const reader = this.reader;
+    if (reader.size() === 0) return undefined;
+    const x = reader.x;
+    try {
+      const args = this.decoder.readCmd();
+      reader.consume();
+      return args;
+    } catch (error) {
+      if (error instanceof RangeError) {
+        reader.x = x;
+        return undefined;
+      } else throw error;
+    }
+  }
+
   /**
    * Skips one value from the stream. If `undefined` is returned, then
    * there is not enough data to skip or the stream is finished.

src/json-pack/resp/__tests__/RespStreamingDecoder.spec.ts

@@ -2,6 +2,7 @@ import {RespStreamingDecoder} from '../RespStreamingDecoder';
 import {RespEncoder} from '../RespEncoder';
 import {concatList} from '../../../util/buffers/concat';
 import {documents} from '../../../__tests__/json-documents';
+import {utf8} from '../../../util/buffers/strings';
 
 const encoder = new RespEncoder();
 
@@ -66,3 +67,11 @@ test('can stream 49 bytes at a time', () => {
   }
   expect(decoded).toEqual(docs);
 });
+
+test('can decode a command', () => {
+  const encoded = encoder.encodeCmd(['SET', 'foo', 'bar']);
+  const decoder = new RespStreamingDecoder();
+  decoder.push(encoded);
+  const decoded = decoder.readCmd();
+  expect(decoded).toEqual(['SET', utf8`foo`, utf8`bar`]);
+});