fix broken types

clear · May 7, 2024 · 52f03a1 · 52f03a1
1 parent 06686a3
commit 52f03a1
Show file tree

Hide file tree

Showing 6 changed files with 25 additions and 22 deletions.
diff --git a/.changeset/loud-jeans-switch.md b/.changeset/loud-jeans-switch.md
@@ -0,0 +1,5 @@
+---
+"windpipe": patch
+---
+
+fix broken types
diff --git a/src/index.ts b/src/index.ts
@@ -15,7 +15,4 @@ export type {
 // Re-export useful utility types
 export type { MaybePromise, Truthy, CallbackOrStream } from "./util";
 
-// Export the `StreamEnd` type
-export { StreamEnd } from "./stream";
-
 export default Stream;
diff --git a/src/stream/base.ts b/src/stream/base.ts
@@ -3,21 +3,21 @@ import { Stream } from ".";
 import { Readable, Writable } from "stream";
 import { createNodeCallback } from "../util";
 
-/**
- * Marker for the end of a stream.
- */
-export const StreamEnd = Symbol.for("STREAM_END");
-
 /**
  * Unique type to represent the stream end marker.
  */
-export type StreamEnd = typeof StreamEnd;
+export type StreamEnd = typeof StreamBase.StreamEnd;
 
 export class StreamBase {
     protected stream: Readable;
     protected stackTrace: string[] = [];
     protected traceComplete: boolean = false;
 
+    /**
+     * Marker for the end of a stream.
+     */
+    static StreamEnd = Symbol.for("STREAM_END");
+
     constructor(stream: Readable) {
         this.stream = stream;
     }
@@ -124,7 +124,7 @@ export class StreamBase {
 
                 return normalise(await promise);
             } else {
-                return StreamEnd;
+                return StreamBase.StreamEnd;
             }
         });
     }
@@ -144,7 +144,7 @@ export class StreamBase {
             const { value, done } = result instanceof Promise ? await result : result;
 
             if (done) {
-                return StreamEnd;
+                return StreamBase.StreamEnd;
             } else {
                 return normalise(value);
             }
@@ -178,7 +178,7 @@ export class StreamBase {
      */
     static fromArray<T, E>(array: MaybeAtom<T, E>[]): Stream<T, E> {
         return Stream.fromNext(async () => {
-            return array.shift() ?? StreamEnd;
+            return array.shift() ?? StreamBase.StreamEnd;
         });
     }
 
@@ -199,9 +199,11 @@ export class StreamBase {
                         const value = await next();
 
                         // Promise returned as normal
-                        if (value === StreamEnd) {
+                        if (value === StreamBase.StreamEnd) {
                             this.push(null);
                         } else {
+                            // @ts-expect-error - The previous `if` statement doesn't cause TS to
+                            // type-narrow out `symbol`
                             this.push(normalise(value));
                         }
                     } catch (e) {
@@ -226,7 +228,7 @@ export class StreamBase {
                 consumed = true;
                 return value;
             } else {
-                return StreamEnd;
+                return StreamBase.StreamEnd;
             }
         });
     }
@@ -285,7 +287,7 @@ export class StreamBase {
             },
             async final(callback) {
                 // Emit a `StreamEnd` to close the stream
-                enqueue(StreamEnd);
+                enqueue(StreamBase.StreamEnd);
 
                 callback();
             },

diff --git a/src/stream/index.ts b/src/stream/index.ts
@@ -12,7 +12,7 @@ import {
 } from "../atom";
 import { HigherOrderStream } from "./higher-order";
 
-export { StreamEnd } from "./base";
+export type { StreamEnd } from "./base";
 
 /**
  * @template T - Type of the 'values' on the stream.

diff --git a/test/benchmarks/index.bench.ts b/test/benchmarks/index.bench.ts
@@ -1,7 +1,6 @@
 import { describe, bench } from "vitest";
 import Stream from "../../src";
 import Highland from "highland";
-import { StreamEnd } from "../../src/stream/base";
 import { error, ok } from "../../src/atom";
 
 const SAMPLE_SIZE = 10;
@@ -14,7 +13,7 @@ describe("stream creation from next function", () => {
             if (i < SAMPLE_SIZE) {
                 return i++;
             } else {
-                return StreamEnd;
+                return Stream.StreamEnd;
             }
         }).toArray();
     });

diff --git a/test/creation.test.ts b/test/creation.test.ts
@@ -1,5 +1,5 @@
 import { describe, test } from "vitest";
-import $, { StreamEnd } from "../src";
+import $ from "../src";
 import { Readable } from "stream";
 
 describe.concurrent("stream creation", () => {
@@ -109,7 +109,7 @@ describe.concurrent("stream creation", () => {
                 if (i < 4) {
                     return i++;
                 } else {
-                    return StreamEnd;
+                    return $.StreamEnd;
                 }
             });
 
@@ -124,7 +124,7 @@ describe.concurrent("stream creation", () => {
                 if (atoms.length > 0) {
                     return atoms.shift();
                 } else {
-                    return StreamEnd;
+                    return $.StreamEnd;
                 }
             });
 
@@ -151,7 +151,7 @@ describe.concurrent("stream creation", () => {
                     return i;
                 }
 
-                return StreamEnd;
+                return $.StreamEnd;
             });
 
             expect(await s.toArray({ atoms: true })).toEqual([