Prevent duplicate pull signals

0no-co · Jul 20, 2023 · 4cdc6a0 · 4cdc6a0
1 parent 2dd532b
commit 4cdc6a0
Show file tree

Hide file tree

Showing 2 changed files with 15 additions and 10 deletions.
diff --git a/src/__tests__/sinks.test.ts b/src/__tests__/sinks.test.ts
@@ -246,18 +246,17 @@ describe('toAsyncIterable', () => {
     const asyncIterator = sinks.toAsyncIterable(source)[Symbol.asyncIterator]();
     const next$ = asyncIterator.next();
 
-    expect(pulls).toBe(2);
     sink!(push(0));
     expect(await next$).toEqual({ value: 0, done: false });
-    expect(pulls).toBe(2);
+    expect(pulls).toBe(1);
 
     sink!(push(1));
     expect(await asyncIterator.next()).toEqual({ value: 1, done: false });
-    expect(pulls).toBe(3);
+    expect(pulls).toBe(2);
 
     sink!(SignalKind.End);
     expect(await asyncIterator.next()).toEqual({ done: true });
-    expect(pulls).toBe(3);
+    expect(pulls).toBe(2);
   });
 
   it('buffers actively pushed values', async () => {
@@ -280,7 +279,7 @@ describe('toAsyncIterable', () => {
     sink!(push(1));
     sink!(SignalKind.End);
 
-    expect(pulls).toBe(2);
+    expect(pulls).toBe(1);
     expect(await next$).toEqual({ value: 0, done: false });
     expect(await asyncIterator.next()).toEqual({ value: 1, done: false });
     expect(await asyncIterator.next()).toEqual({ done: true });
@@ -301,7 +300,7 @@ describe('toAsyncIterable', () => {
 
     const asyncIterator = sinks.toAsyncIterable(source)[Symbol.asyncIterator]();
     asyncIterator.next();
-    expect(pulls).toBe(2);
+    expect(pulls).toBe(1);
 
     let resolved = false;
 

diff --git a/src/sinks.ts b/src/sinks.ts
@@ -129,6 +129,7 @@ export const toAsyncIterable = <T>(source: Source<T>): SourceIterable<T> => {
 
   let ended = false;
   let started = false;
+  let pulled = false;
   let talkback = talkbackPlaceholder;
   let next: ((value: IteratorResult<T>) => void) | void;
 
@@ -143,18 +144,23 @@ export const toAsyncIterable = <T>(source: Source<T>): SourceIterable<T> => {
             if (next) next = next(doneResult);
             ended = true;
           } else if (signal.tag === SignalKind.Start) {
+            pulled = true;
             (talkback = signal[0])(TalkbackKind.Pull);
-          } else if (next) {
-            next = next({ value: signal[0], done: false });
           } else {
-            buffer.push(signal[0]);
+            pulled = false;
+            if (next) {
+              next = next({ value: signal[0], done: false });
+            } else {
+              buffer.push(signal[0]);
+            }
           }
         });
       }
 
       if (ended && !buffer.length) {
         return doneResult;
-      } else if (!ended && buffer.length <= 1) {
+      } else if (!ended && !pulled && buffer.length <= 1) {
+        pulled = true;
         talkback(TalkbackKind.Pull);
       }