From 52f03a112d29ef01e5027348f7fe474fd816e5ca Mon Sep 17 00:00:00 2001
From: Tom Anderson <tom@ando.gq>
Date: Tue, 7 May 2024 16:32:36 +1000
Subject: [PATCH] fix broken types

---
 .changeset/loud-jeans-switch.md |  5 +++++
 src/index.ts                    |  3 ---
 src/stream/base.ts              | 26 ++++++++++++++------------
 src/stream/index.ts             |  2 +-
 test/benchmarks/index.bench.ts  |  3 +--
 test/creation.test.ts           |  8 ++++----
 6 files changed, 25 insertions(+), 22 deletions(-)
 create mode 100644 .changeset/loud-jeans-switch.md

diff --git a/.changeset/loud-jeans-switch.md b/.changeset/loud-jeans-switch.md
new file mode 100644
index 0000000..9d29c47
--- /dev/null
+++ 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
index 23ae95a..d1bf96b 100644
--- 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
index 21198a0..eb6d7a9 100644
--- 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
index 0c317aa..af53647 100644
--- 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
index 00a116d..82a1d68 100644
--- 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
index dbb07a3..d55f646 100644
--- 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([