From 948c6cf87be646929893b541acf5e4fcdb14a929 Mon Sep 17 00:00:00 2001
From: streamich <streamich@gmail.com>
Date: Sat, 16 Dec 2023 22:53:43 +0100
Subject: [PATCH] =?UTF-8?q?feat(json-pack):=20=F0=9F=8E=B8=20add=20skippin?=
 =?UTF-8?q?g=20ability=20to=20streaming=20RESP=20decoder?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 src/json-pack/resp/RespStreamingDecoder.ts    | 22 ++++++++++
 src/json-pack/resp/__tests__/skipping.spec.ts | 40 ++++++++++++++-----
 2 files changed, 51 insertions(+), 11 deletions(-)

diff --git a/src/json-pack/resp/RespStreamingDecoder.ts b/src/json-pack/resp/RespStreamingDecoder.ts
index f454f49b56..4dac79aab4 100644
--- a/src/json-pack/resp/RespStreamingDecoder.ts
+++ b/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;
+    }
+  }
 }
diff --git a/src/json-pack/resp/__tests__/skipping.spec.ts b/src/json-pack/resp/__tests__/skipping.spec.ts
index b6eec6f723..da040c0b8a 100644
--- a/src/json-pack/resp/__tests__/skipping.spec.ts
+++ b/src/json-pack/resp/__tests__/skipping.spec.ts
@@ -1,5 +1,6 @@
 import {RespEncoder} from '../RespEncoder';
 import {RespDecoder} from '../RespDecoder';
+import {RespStreamingDecoder} from '../RespStreamingDecoder';
 import {documents} from '../../../__tests__/json-documents';
 import {binaryDocuments} from '../../../__tests__/binary-documents';
 
@@ -7,17 +8,34 @@ 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'});
+      });
+    }
+  });
 });