fix: Use take in delayUntil.

cartant · Feb 23, 2020 · 10e217c · 10e217c
1 parent 2b88ae9
commit 10e217c
Show file tree

Hide file tree

Showing 2 changed files with 45 additions and 3 deletions.
diff --git a/source/operators/delayUntil-spec.ts b/source/operators/delayUntil-spec.ts
@@ -23,7 +23,49 @@ describe("delayUntil", () => {
   );
 
   it(
-    "should support empty sources",
+    "should don't emit after the signal",
+    marbles(m => {
+      const source = m.cold("   -a-b------|");
+      const signal = m.cold("   -----x-----");
+      const subs = "            ^---------!";
+      const expected = m.cold(" -----(ab)-|");
+
+      const destination = source.pipe(delayUntil(signal));
+      m.expect(destination).toBeObservable(expected);
+      m.expect(source).toHaveSubscriptions(subs);
+    })
+  );
+
+  it(
+    "should don't emit before the signal",
+    marbles(m => {
+      const source = m.cold("   -------a-b|");
+      const signal = m.cold("   -----x-----");
+      const subs = "            ^---------!";
+      const expected = m.cold(" -------a-b|");
+
+      const destination = source.pipe(delayUntil(signal));
+      m.expect(destination).toBeObservable(expected);
+      m.expect(source).toHaveSubscriptions(subs);
+    })
+  );
+
+  it(
+    "should support empty sources that complete before the signal",
+    marbles(m => {
+      const source = m.cold("   ---|   ");
+      const signal = m.cold("   -----x-");
+      const subs = "            ^--!   ";
+      const expected = m.cold(" ---|   ");
+
+      const destination = source.pipe(delayUntil(signal));
+      m.expect(destination).toBeObservable(expected);
+      m.expect(source).toHaveSubscriptions(subs);
+    })
+  );
+
+  it(
+    "should support empty sources that complete after the signal",
     marbles(m => {
       const source = m.cold("   --------------|");
       const signal = m.cold("   -----x---------");

diff --git a/source/operators/delayUntil.ts b/source/operators/delayUntil.ts
@@ -4,15 +4,15 @@
  */
 
 import { concat, Observable, OperatorFunction } from "rxjs";
-import { buffer, first, mergeAll, publish } from "rxjs/operators";
+import { buffer, mergeAll, publish, take } from "rxjs/operators";
 
 export function delayUntil<T>(
   notifier: Observable<any>
 ): OperatorFunction<T, T> {
   return source =>
     source.pipe(
       publish(published =>
-        concat(published.pipe(buffer(notifier), first(), mergeAll()), published)
+        concat(published.pipe(buffer(notifier), take(1), mergeAll()), published)
       )
     );
 }