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

@@ -9,10 +9,10 @@ import Subscription from './Subscription';
 export default class Subscriber<T> extends Subscription<T> implements Observer<T> {
  protected destination: Observer<any>;
 
- private _subscription: Subscription<T>;
+ protected _subscription: Subscription<T>;
+
+ protected _isUnsubscribed: boolean = false;
 
- private _isUnsubscribed: boolean = false;
-
  static create<T>(next ?: (x?:any) => void,
  error ?: (e?:any) => void,
  complete?: () => void): Subscriber<T> {
@@ -22,23 +22,23 @@ export default class Subscriber<T> extends Subscription<T> implements Observer<T
  subscriber._complete = (typeof complete === "function") && complete || noop;
  return subscriber;
  }
- 
+
  _next(value: T) {
  this.destination.next(value);
  }
- 
+
  _error(err: any) {
  this.destination.error(err);
  }
- 
+
  _complete() {
  this.destination.complete();
  }
 
  constructor(destination?: Observer<any>) {
  super();
  this.destination = destination;
- 
+
  if (!destination) {
  return;
  }
@@ -90,9 +90,9 @@ export default class Subscriber<T> extends Subscription<T> implements Observer<T
  }
 
  unsubscribe() {
- 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

@@ -19,9 +19,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));
  }
@@ -42,6 +42,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> {