refactor(skipUntil): call super(destination) constructor

Change SkipUntilSubscriber to give the destination subscriber to the
constructor in Subscriber. Adapt SkipUntilSubscriber to handle
unsubscription-on-complete correctly, by overriding unsubscribe().

Addresses ReactiveX#577.

src/Subscriber.ts

@@ -6,8 +6,8 @@ import Observer from './Observer';
 import Subscription from './Subscription';
 
 export default class Subscriber<T> extends Subscription<T> implements Observer<T> {
- private _subscription: Subscription<T>;
- private _isUnsubscribed: boolean = false;
+ protected _subscription: Subscription<T>;
+ protected _isUnsubscribed: boolean = false;
 
  get isUnsubscribed(): boolean {
  const subscription = this._subscription;
@@ -73,9 +73,9 @@ export default class Subscriber<T> extends Subscription<T> implements Observer<T
  }
 
  unsubscribe(): void {
- if(this._isUnsubscribed) {
+ if (this._isUnsubscribed) {
  return;
- } else if(this._subscription) {
+ } else if (this._subscription) {
  this._isUnsubscribed = true;
  } else {
  super.unsubscribe();

src/operators/skipUntil.ts

@@ -18,9 +18,9 @@ class SkipUntilOperator<T, R> implements Operator<T, R> {
 class SkipUntilSubscriber<T> extends Subscriber<T> {
  private notificationSubscriber: NotificationSubscriber<any> = null;
 
- constructor(public destination: Subscriber<T>,
+ constructor(destination: Subscriber<T>,
  private notifier: Observable<any>) {
- super(null);
+ super(destination);
  this.notificationSubscriber = new NotificationSubscriber(this);
  this.add(this.notifier.subscribe(this.notificationSubscriber));
  }
@@ -41,6 +41,17 @@ class SkipUntilSubscriber<T> extends Subscriber<T> {
  }
  this.notificationSubscriber.unsubscribe();
  }
+
+ unsubscribe() {
+ if (this._isUnsubscribed) {
+ return;
+ } else if (this._subscription) {
+ this._subscription.unsubscribe();
+ this._isUnsubscribed = true;
+ } else {
+ super.unsubscribe();
+ }
+ }
 }
 
 class NotificationSubscriber<T> extends Subscriber<T> {