feat: Allow writer to be ended and error on writing to closed writer

src/connectors.ts

@@ -223,6 +223,7 @@ export class ChannelFactory {
 export interface Writer<T> {
     push(item: T): Promise<void>;
     end(): Promise<void>;
+    on(event: "end", listener: Handler<void>): this;
 }
 
 export interface Stream<T> {
@@ -242,6 +243,8 @@ export class SimpleStream<T> implements Stream<T> {
     public readonly disconnect: () => Promise<void>;
     public lastElement?: T | undefined;
 
+    private ended = false;
+
     public constructor(onDisconnect?: () => Promise<void>) {
         this.disconnect = onDisconnect || (async () => {});
     }
@@ -252,13 +255,17 @@ export class SimpleStream<T> implements Stream<T> {
     }
 
     public async push(data: T): Promise<void> {
+        if (this.ended) {
+            throw new Error("Trying to push to a stream that has ended!");
+        }
         this.lastElement = data;
         await Promise.all(this.dataHandlers.map((handler) => handler(data)));
     }
 
     public async end(): Promise<void> {
         await this.disconnect();
         await Promise.all(this.endHandlers.map((handler) => handler()));
+        this.ended = true;
     }
 
     public on(event: "data", listener: Handler<T>): this;

src/connectors/file.ts

@@ -121,21 +121,21 @@ export const startFileStreamWriter: WriterConstructor<FileWriterConfig> = (
         : `${process.cwd()}/${config.path}`;
     const encoding: BufferEncoding = <BufferEncoding>config.encoding || "utf-8";
 
+    const writer = new SimpleStream<string>();
+
     const init = async () => {
         if (!config.onReplace) {
             await writeFile(path, "", { encoding });
         }
     };
 
-    const push = async (item: string): Promise<void> => {
+    writer.push = async (item: string): Promise<void> => {
         if (config.onReplace) {
             await writeFile(path, item, { encoding });
         } else {
             await appendFile(path, item, { encoding });
         }
     };
 
-    const end = async (): Promise<void> => {};
-
-    return { writer: { push, end }, init };
+    return { writer, init };
 };

src/connectors/http.ts

@@ -101,7 +101,9 @@ export const startHttpStreamWriter: WriterConstructor<HttpWriterConfig> = (
 ) => {
     const requestConfig = <https.RequestOptions>new URL(config.endpoint);
 
-    const push = async (item: string | Buffer): Promise<void> => {
+    const writer = new SimpleStream<string | Buffer>();
+
+    writer.push = async (item: string | Buffer): Promise<void> => {
         await new Promise((resolve) => {
             const options = {
                 hostname: requestConfig.hostname,
@@ -125,7 +127,5 @@ export const startHttpStreamWriter: WriterConstructor<HttpWriterConfig> = (
         });
     };
 
-    const end = async (): Promise<void> => {};
-
-    return { writer: { push, end }, init: async () => {} };
+    return { writer, init: async () => {} };
 };

src/connectors/kafka.ts

@@ -138,15 +138,16 @@ export const startKafkaStreamWriter: WriterConstructor<KafkaWriterConfig> = (
     const kafka = new Kafka(<KafkaConfig>brokerConfig);
 
     const producer = kafka.producer(config.producer);
+
+    const writer = new SimpleStream<string>(async () => {
+        await producer.disconnect();
+    });
+
     const init = () => producer.connect();
 
-    const push = async (item: string): Promise<void> => {
+    writer.push = async (item: string): Promise<void> => {
         await producer.send({ topic, messages: [{ value: item }] });
     };
 
-    const end = async (): Promise<void> => {
-        await producer.disconnect();
-    };
-
-    return { writer: { push, end }, init };
+    return { writer, init };
 };

src/connectors/ws.ts

@@ -94,14 +94,14 @@ export const startWsStreamWriter: WriterConstructor<WsWriterConfig> = (
         ws.on("open", () => console.log("open"));
     };
 
+    const writer = new SimpleStream<string>(async () => {
+        ws.close();
+    });
+
     /* eslint-disable-next-line @typescript-eslint/no-explicit-any */
-    const push = async (item: any): Promise<void> => {
+    writer.push = async (item: any): Promise<void> => {
         ws.send(item);
     };
 
-    const end = async (): Promise<void> => {
-        ws.close();
-    };
-
-    return { writer: { push, end }, init };
+    return { writer, init };
 };