From 1c257dbb389d021f72b07690add4ad4074f817f6 Mon Sep 17 00:00:00 2001
From: peaBerberian <pea.berberian@gmail.com>
Date: Sun, 19 Aug 2018 14:05:50 +0200
Subject: [PATCH] fix(skipUntil): stop listening to a synchronous notifier
 after its first nexted value

---
 spec/operators/skipUntil-spec.ts    | 23 ++++++++++++++++++++++-
 src/internal/operators/skipUntil.ts |  5 ++++-
 2 files changed, 26 insertions(+), 2 deletions(-)

diff --git a/spec/operators/skipUntil-spec.ts b/spec/operators/skipUntil-spec.ts
index 5248973db7..12430e8879 100644
--- a/spec/operators/skipUntil-spec.ts
+++ b/spec/operators/skipUntil-spec.ts
@@ -1,6 +1,6 @@
 import { expect } from 'chai';
 import { hot, cold, expectObservable, expectSubscriptions } from '../helpers/marble-testing';
-import { Observable, of, Subject } from 'rxjs';
+import { concat, defer, Observable, of, Subject } from 'rxjs';
 import { skipUntil, mergeMap } from 'rxjs/operators';
 
 declare function asDiagram(arg: string): Function;
@@ -246,4 +246,25 @@ describe('skipUntil', () => {
     expectObservable(result).toBe(expected);
     expectSubscriptions(notifier.subscriptions).toBe(nSubs);
   });
+
+  it('should stop listening to a synchronous notifier after its first nexted value', () => {
+    // const source =   hot('-^-o---o---o---o---o---o---|');
+    const sideEffects: number[] = [];
+    const synchronousNotifer = concat(
+      defer(() => {
+        sideEffects.push(1);
+        return of(1);
+      }),
+      defer(() => {
+        sideEffects.push(2);
+        return of(2);
+      }),
+      defer(() => {
+        sideEffects.push(3);
+        return of(3);
+      })
+    );
+    of(null).pipe(skipUntil(synchronousNotifer)).subscribe(() => { /* noop */ });
+    expect(sideEffects).to.deep.equal([1]);
+  });
 });
diff --git a/src/internal/operators/skipUntil.ts b/src/internal/operators/skipUntil.ts
index f2fc2d186f..f565ae4d6c 100644
--- a/src/internal/operators/skipUntil.ts
+++ b/src/internal/operators/skipUntil.ts
@@ -44,7 +44,10 @@ class SkipUntilSubscriber<T, R> extends OuterSubscriber<T, R> {
 
   constructor(destination: Subscriber<R>, notifier: ObservableInput<any>) {
     super(destination);
-    this.add(this.innerSubscription = subscribeToResult(this, notifier));
+    const innerSubscriber = new InnerSubscriber(this, undefined, undefined);
+    this.add(innerSubscriber);
+    this.innerSubscription = innerSubscriber;
+    subscribeToResult(this, notifier, undefined, undefined, innerSubscriber);
   }
 
   protected _next(value: T) {