File Reader: fix rewind bug

SeaQL · Sep 28, 2023 · 4dfd601 · 4dfd601
1 parent 94cf111
commit 4dfd601
Show file tree

Hide file tree

Showing 3 changed files with 10 additions and 4 deletions.
diff --git a/sea-streamer-file/sea-streamer-file-reader/src/message.test.ts b/sea-streamer-file/sea-streamer-file-reader/src/message.test.ts
@@ -30,6 +30,11 @@ async function testRewind(mode: StreamMode) {
 
     await expectNext(0, 100);
 
+    nth = await source.rewind(new SeqPos.At(0n)); if (nth instanceof FileErr) { throwNewError(nth); }
+    expect(nth).toStrictEqual(0);
+
+    await expectNext(0, 100);
+
     const eos = await source.next(); if (eos instanceof FileErr) { throwNewError(eos); }
     expect(isEndOfStream(eos)).toBe(true);
 

diff --git a/sea-streamer-file/sea-streamer-file-reader/src/message.ts b/sea-streamer-file/sea-streamer-file-reader/src/message.ts
@@ -66,14 +66,14 @@ export class MessageSource implements ByteSource {
      * Returns the current location in terms of N-th beacon.
      */
     async rewind(target: SeqPosEnum): Promise<number | FileErr> {
-        let pos;
+        let pos: SeqPosEnum;
         if (target instanceof SeqPos.Beginning) {
             pos = new SeqPos.At(Header.size());
         } else if (target instanceof SeqPos.End) {
             pos = target;
         } else if (target instanceof SeqPos.At) {
             if (target.at === 0n) {
-                pos = Header.size();
+                pos = new SeqPos.At(Header.size());
             } else {
                 let at = target.at * this.beaconInterval();
                 if (at < this.knownSize()) {

diff --git a/sea-streamer-file/sea-streamer-file-reader/src/subprocess.ts b/sea-streamer-file/sea-streamer-file-reader/src/subprocess.ts
@@ -153,7 +153,7 @@ async function run() {
         }
         for (let i = 0; i < batchSize; i++) {
             const message = await source.next();
-            if (message instanceof FileErr) { process.send!({ error: message.toString() }); global.error = true; return; }
+            if (message instanceof FileErr) { process.send!({ error: `${message}` }); global.error = true; return; }
             buffer.push(message);
             if (isEndOfStream(message)) {
                 ended = true;
@@ -200,7 +200,8 @@ async function seek(nth: number) {
     }
     global.state = State.Seeking as State;
     const source = global.source!;
-    await source.rewind(new SeqPos.At(BigInt(nth)));
+    const rewinded = await source.rewind(new SeqPos.At(BigInt(nth)));
+    if (rewinded instanceof FileErr) { process.send!({ error: `${rewinded}` }); global.error = true; return; }
     process_log("rewinded");
     const payload = new Buffer();
     payload.append(SystemBuffer.from(PULSE_MESSAGE));