fix: make sure the last value is flushed before terminating

README.md

@@ -23,6 +23,35 @@ pipe(
 
 ## Changelog
 
+### 4.0.0 (21/06/2021)
+
+BREAKING CHANGE:
+
+- Last value is flushed even if the stream is receiving a terminate signal before the value has been debounced (fixes #12)
+
+Before
+
+```js
+pipe(
+  of(42),
+  debounce(1000),
+  subscribe(console.log)
+)
+// Terminate without logging anything
+```
+
+After
+
+```js
+pipe(
+  of(42),
+  debounce(1000),
+  subscribe(console.log)
+)
+// log: 42
+// Then terminate
+```
+
 ### 3.0.0 (05/03/2021)
 
 BREAKING CHANGE:

src/debounce.ts

@@ -10,6 +10,7 @@ export function debounce<I>(wait: number): (source: Source<I>) => Source<I> {
   return (source) => (start: number, sink: any) => {
     if (start !== 0) return;
     let timeout: NodeJS.Timeout | undefined;
+    let shouldTerminate = false;
     source(0, (t: number, d: any) => {
       if (t === 0) {
         // handle talkback from sink
@@ -26,14 +27,24 @@ export function debounce<I>(wait: number): (source: Source<I>) => Source<I> {
       } else if (t === 1 || (t === 2 && d === undefined)) {
         // t === 1 means the source is emitting a value
         // t === 2 and d === undefined means the source emits a completion
-        if (!timeout && t === 2) {
-          return sink(t, d);
+        if (t === 2) {
+          if (!timeout) {
+            // There is not pending value, we can terminate the stream
+            return sink(t, d);
+          } else {
+            // We keep track that the stream should terminate after the next value is emitted
+            shouldTerminate = true;
+            return;
+          }
         }
         if (timeout) {
           clearTimeout(timeout);
         }
         timeout = setTimeout(() => {
           sink(t, d);
+          if (shouldTerminate) {
+            sink(2);
+          }
           timeout = undefined;
         }, wait);
       } else {

test.ts

@@ -99,3 +99,25 @@ test("it should not emit after unsubscribe", (t) => {
     t.equal(sink.getReceivedData().length, 0);
   }, 5);
 });
+
+test("it should flush last value when receiving a terminaison signal", (t) => {
+  t.plan(1);
+
+  const source = mock(true);
+  const sink = mock();
+
+  debounce(2)(source)(0, sink);
+
+  source.emit(1, true);
+  source.emit(2);
+
+  setTimeout(() => {
+    t.deepEqual(
+      sink.getMessages().filter(([t]) => t !== 0),
+      [
+        [1, true],
+        [2, undefined],
+      ]
+    );
+  }, 3);
+});