Topic test many readers, many writers

Author: Eryk Solecki

packages/host/src/lib/serviceDiscovery/topic.ts

@@ -73,13 +73,27 @@ class Topic extends Duplex implements StreamHandler {
             this.setState(WorkState.Flowing)
     }
 
-    pipe<T extends NodeJS.WritableStream>(destination: T, options: { end?: boolean; } = { end: false }): T {
+    pipe<T extends StreamWrapper<Writable>>(destination: T, options?: { end?: boolean; }): T
+    pipe<T extends NodeJS.WritableStream>(destination: T, options?: { end?: boolean; }): T
+    pipe(destination: StreamWrapper<Writable> | NodeJS.WritableStream, options?: { end?: boolean; }): typeof destination {
+        if (destination instanceof StreamWrapper<Writable>)
+            destination = destination.stream();
+
+        if (!(destination instanceof Writable))
+            throw new Error("Streams not extending Writable are not supported");
+
         this.addConsumer(destination);
         return super.pipe(destination, options);
     };
 
-    unpipe(destination?: NodeJS.WritableStream): this {
-        if (destination) this.removeConsumer(destination)
+    unpipe(destination?: StreamWrapper<Writable> | NodeJS.WritableStream): this {
+        if (destination) {
+            if (destination instanceof StreamWrapper<Writable>)
+                destination = destination.stream();
+            if (!(destination instanceof Writable))
+                throw new Error("Streams not extending Writable are not supported");
+            this.removeConsumer(destination)
+        }
         else this.removeAllConsumers();
 
         return super.unpipe(destination);
@@ -154,18 +168,12 @@ class Topic extends Duplex implements StreamHandler {
         this.emit(TopicEvent.ProvidersChanged);
     }
 
-    protected addConsumer<T extends NodeJS.WritableStream>(destination: T) {
-        if (!(destination instanceof Writable))
-            throw new Error("Streams not extending Writable are not supported");
-
+    protected addConsumer<T extends Writable>(destination: T) {
         if (!this.addStream(destination, this.consumers)) return
         this.updateState();
         this.emit(TopicEvent.ConsumersChanged);
     }
-    protected removeConsumer<T extends NodeJS.WritableStream>(destination: T) {
-        if (!(destination instanceof Writable))
-            throw new Error("Streams not extending Writable are not supported");
-
+    protected removeConsumer<T extends Writable>(destination: T) {
         if (!this.removeStream(destination, this.consumers)) return;
         this.updateState()
         this.emit(TopicEvent.ConsumersChanged);

packages/host/test/serviceDiscovery/topic.spec.ts

@@ -65,14 +65,14 @@ describe("Consumer management", () => {
 
     test("Automaticly add consumer on pipe", () => {
         expect(testTopic.consumers.size).toBe(0);
-        testTopic.pipe(testConsumer.stream());
+        testTopic.pipe(testConsumer);
         expect(testTopic.consumers.size).toBe(1);
     })
 
     test("Add only unique streams on pipe", () => {
-        testTopic.pipe(testConsumer.stream());
-        testTopic.pipe(testConsumer.stream());
-        testTopic.pipe(testConsumer.stream());
+        testTopic.pipe(testConsumer);
+        testTopic.pipe(testConsumer);
+        testTopic.pipe(testConsumer);
         expect(testTopic.consumers.size).toBe(1);
     })
 
@@ -81,9 +81,9 @@ describe("Consumer management", () => {
         const testConsumer2 = StreamWrapper.create(new Writable({ write: () => { } }), "testWriteStream2", StreamType.Instance, {});
         const testConsumer3 = StreamWrapper.create(new Writable({ write: () => { } }), "testWriteStream3", StreamType.Instance, {});
 
-        testTopic.pipe(testConsumer1.stream());
-        testTopic.pipe(testConsumer2.stream());
-        testTopic.pipe(testConsumer3.stream());
+        testTopic.pipe(testConsumer1);
+        testTopic.pipe(testConsumer2);
+        testTopic.pipe(testConsumer3);
 
         expect(testTopic.consumers.size).toBe(3);
     })
@@ -106,17 +106,16 @@ describe("Consumer management", () => {
         const testConsumer2 = StreamWrapper.create(new Writable({ write: () => { } }), "testWriteStream2", StreamType.Instance, {});
         const testConsumer3 = StreamWrapper.create(new Writable({ write: () => { } }), "testWriteStream3", StreamType.Instance, {});
 
-        testTopic.pipe(testConsumer1.stream());
-        testTopic.pipe(testConsumer2.stream());
-        testTopic.pipe(testConsumer3.stream());
+        testTopic.pipe(testConsumer1);
+        testTopic.pipe(testConsumer2);
+        testTopic.pipe(testConsumer3);
 
         expect(testTopic.consumers.size).toBe(3);
         testTopic.unpipe();
         expect(testTopic.consumers.size).toBe(0);
     })
 })
 
-
 describe("Event flow", () => {
     const waitForEvent = (eventName: string, source: Stream) => {
         return new Promise<boolean>((resolve, reject) => {
@@ -182,7 +181,7 @@ describe("Event flow", () => {
             testProvider.stream().pipe(testTopic)
             expect(testTopic.state()).toBe(WorkState.Waiting);
             const eventPromise = waitForEvent(TopicEvent.StateChanged, testTopic);
-            testTopic.pipe(testConsumer.stream());
+            testTopic.pipe(testConsumer);
             await eventPromise
             expect(testTopic.state()).toBe(WorkState.Flowing);
 
@@ -208,7 +207,7 @@ describe("Event flow", () => {
 
         test("State flowing after resume()", async () => {
             let eventPromise = waitForEvent(TopicEvent.StateChanged, testTopic);
-            testProvider.stream().pipe(testTopic).pipe(testConsumer.stream());
+            testProvider.stream().pipe(testTopic).pipe(testConsumer);
             await eventPromise;
             expect(testTopic.state()).toBe(WorkState.Flowing);
 
@@ -238,16 +237,16 @@ describe("Event flow", () => {
         })
         test("ConsumersChanged on add", async () => {
             const eventPromise = waitForEvent(TopicEvent.ConsumersChanged, testTopic);
-            testTopic.pipe(testConsumer.stream());
+            testTopic.pipe(testConsumer);
             await eventPromise;
             expect(testTopic.consumers.size).toBe(1);
         })
         test("ConsumersChanged on remove", async () => {
             let eventPromise = waitForEvent(TopicEvent.ConsumersChanged, testTopic);
-            testTopic.pipe(testConsumer.stream());
+            testTopic.pipe(testConsumer);
             await eventPromise;
             eventPromise = waitForEvent(TopicEvent.ConsumersChanged, testTopic);
-            testTopic.unpipe(testConsumer.stream());
+            testTopic.unpipe(testConsumer);
             expect(testTopic.consumers.size).toBe(0);
         })
     })
@@ -256,6 +255,13 @@ describe("Event flow", () => {
 describe("Data flow", () => {
     const testText = "Lorem ipsum dolor sit amet, consectetur adipiscing elit.";
 
+    const createWaitingPromise = (): [Promise<void>, () => void, (_: any) => void] => {
+        let res = () => { }
+        let rej = (reason: any) => { };
+        const promise = new Promise<void>((resolve, reject) => { res = resolve, rej = reject })
+        return [promise, res, rej]
+    }
+
     test("Basic flow", async () => {
         const topicFinished = new Promise(resolve => testTopic.on("readable", () => {
             resolve(testTopic.read());
@@ -269,7 +275,7 @@ describe("Data flow", () => {
         const testProvider = StreamWrapper.create(new PassThrough({ encoding: "ascii" }), "testReadStream", StreamType.Instance, {});
         const testConsumer = StreamWrapper.create(new PassThrough({ encoding: "ascii" }), "testWriteStream", StreamType.Instance, {});
 
-        testProvider.stream().pipe(testTopic).pipe(testConsumer.stream());
+        testProvider.stream().pipe(testTopic).pipe(testConsumer);
 
         const readPromise = new Promise(resolve => testConsumer.stream().on("readable", () => {
             resolve(testConsumer.stream().read())
@@ -279,10 +285,73 @@ describe("Data flow", () => {
 
         expect(readValue).toBe(testText);
     });
-    test("Many Providers writing", () => {
-
+    test("Many Providers writing", async () => {
+        const [startGeneratingPromise, startGenerating] = createWaitingPromise();
+        async function* generator(from: number, to: number) {
+            let i = from;
+            while (i <= to) {
+                await startGeneratingPromise;
+                yield Number(i++).toString()
+            }
+        }
+
+        const createStreamProvider = (name: string, from: number, to: number): [StreamWrapper<Readable>, Promise<void>] => {
+            const gen = generator(from, to);
+            const provider = StreamWrapper.create(Readable.from(gen).setEncoding("ascii"), name, StreamType.Instance, {});
+            const [streamEndPromise, streamEnd, streamError] = createWaitingPromise();
+            provider.stream().on("close", streamEnd).on("error", streamError)
+            return [provider, streamEndPromise];
+        }
+
+        const [provider1, provider1End] = createStreamProvider("TestReadStream1", 1, 10);
+        const [provider2, provider2End] = createStreamProvider("TestReadStream2", 11, 20);
+        const [provider3, provider3End] = createStreamProvider("TestReadStream3", 21, 30);
+
+        provider1.stream().pipe(testTopic, { end: false });
+        provider2.stream().pipe(testTopic, { end: false });
+        provider3.stream().pipe(testTopic, { end: false });
+
+        const result: number[] = [];
+        testTopic.on("readable", () => { result.push(Number(testTopic.read())) })
+
+        startGenerating();
+        await Promise.all([provider1End, provider2End, provider3End]);
+        result.sort((a: number, b: number) => a - b)
+        const expectedResult = [...Array(30).keys()].map(val => val + 1);
+        const match = result.length === expectedResult.length && !expectedResult.some((value, index) => result[index] !== value)
+        expect(match).toBe(true);
     })
-    test("Many Consumers reading", () => {
+    test("Many Consumers reading", async () => {
+        const consumer1 = StreamWrapper.create(new PassThrough({ encoding: "ascii" }), "TestWriteStream1", StreamType.Instance, {});
+        const consumer2 = StreamWrapper.create(new PassThrough({ encoding: "ascii" }), "TestWriteStream1", StreamType.Instance, {});
+        const consumer3 = StreamWrapper.create(new PassThrough({ encoding: "ascii" }), "TestWriteStream1", StreamType.Instance, {});
+        
+        let result = ["", "", ""];
+        const [readed1Promise, readed1] = createWaitingPromise();
+        const [readed2Promise, readed2] = createWaitingPromise();
+        const [readed3Promise, readed3] = createWaitingPromise();
+        
+        consumer1.stream().on("readable", () => {
+            result[0] = consumer1.stream().read();
+            readed1();
+        })
+        consumer2.stream().on("readable", () => {
+            result[1] = consumer2.stream().read();
+            readed2();
+        })
+        consumer3.stream().on("readable", () => {
+            result[2] = consumer3.stream().read();
+            readed3();
+        })
 
+        testTopic.pipe(consumer1);
+        testTopic.pipe(consumer2);
+        testTopic.pipe(consumer3);
+        testTopic.write(testText);
+        
+        await Promise.all([readed1Promise, readed2Promise, readed3Promise]);
+        expect(result[0]).toBe(testText);
+        expect(result[1]).toBe(testText);
+        expect(result[2]).toBe(testText);
     })
 })