feat(json-pack): 🎸 add skipping ability to streaming RESP decoder

Author: streamich

src/json-pack/resp/RespStreamingDecoder.ts

@@ -66,4 +66,26 @@ export class RespStreamingDecoder {
       } 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.
+   * @returns `null` if a value was skipped, `undefined` if there is not
+   * enough data to skip.
+   */
+  public skip(): null | undefined {
+    const reader = this.reader;
+    if (reader.size() === 0) return undefined;
+    const x = reader.x;
+    try {
+      this.decoder.skipAny();
+      reader.consume();
+      return null;
+    } catch (error) {
+      if (error instanceof RangeError) {
+        reader.x = x;
+        return undefined;
+      } else throw error;
+    }
+  }
 }

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

@@ -1,23 +1,41 @@
 import {RespEncoder} from '../RespEncoder';
 import {RespDecoder} from '../RespDecoder';
+import {RespStreamingDecoder} from '../RespStreamingDecoder';
 import {documents} from '../../../__tests__/json-documents';
 import {binaryDocuments} from '../../../__tests__/binary-documents';
 
 const docs = [...documents, ...binaryDocuments];
 
 const encoder = new RespEncoder();
 const decoder = new RespDecoder();
+const streamingDecoder = new RespStreamingDecoder();
 
 describe('skipping', () => {
-  for (const t of docs) {
-    (t.only ? test.only : test)(t.name, () => {
-      encoder.writeAny(t.json);
-      encoder.writeAny({foo: 'bar'});
-      const encoded = encoder.writer.flush();
-      decoder.reader.reset(encoded);
-      decoder.skipAny();
-      const decoded = decoder.val();
-      expect(decoded).toEqual({foo: 'bar'});
-    });
-  }
+  describe('RespDecoder', () => {
+    for (const t of docs) {
+      (t.only ? test.only : test)(t.name, () => {
+        encoder.writeAny(t.json);
+        encoder.writeAny({foo: 'bar'});
+        const encoded = encoder.writer.flush();
+        decoder.reader.reset(encoded);
+        decoder.skipAny();
+        const decoded = decoder.val();
+        expect(decoded).toEqual({foo: 'bar'});
+      });
+    }
+  });
+
+  describe('RespStreamingDecoder', () => {
+    for (const t of docs) {
+      (t.only ? test.only : test)(t.name, () => {
+        encoder.writeAny(t.json);
+        encoder.writeAny({foo: 'bar'});
+        const encoded = encoder.writer.flush();
+        streamingDecoder.push(encoded);
+        streamingDecoder.skip();
+        const decoded = streamingDecoder.read();
+        expect(decoded).toEqual({foo: 'bar'});
+      });
+    }
+  });
 });