remove not needed subscriptions

ReactiveX · Mar 13, 2017 · 1c5282a · 1c5282a
1 parent 270f2ad
commit 1c5282a
Show file tree

Hide file tree

Showing 3 changed files with 12 additions and 12 deletions.
diff --git a/src/Observable.ts b/src/Observable.ts
@@ -95,7 +95,7 @@ export class Observable<T> implements Subscribable<T> {
  if (operator) {
  operator.call(sink, this.source);
  } else {
- sink.addParentTeardown(this._trySubscribe(sink));
+ sink.addParentTeardownLogic(this._trySubscribe(sink));
  }
 
  if (sink.syncErrorThrowable) {

diff --git a/src/Subscriber.ts b/src/Subscriber.ts
@@ -41,7 +41,7 @@ export class Subscriber<T> extends Subscription implements Observer<T> {
  public syncErrorThrown: boolean = false;
  public syncErrorThrowable: boolean = false;
 
- private parentSubscription: Subscription = new Subscription();
+ private parentSubscription: Subscription | null = null;
 
  protected isStopped: boolean = false;
  protected destination: PartialObserver<any>; // this `any` is the escape hatch to erase extra type param (e.g. R)
@@ -71,7 +71,7 @@ export class Subscriber<T> extends Subscription implements Observer<T> {
  if (typeof destinationOrNext === 'object') {
  if (destinationOrNext instanceof Subscriber) {
  this.destination = (<Subscriber<any>> destinationOrNext);
- (<any> this.destination).add(this);
+ (<Subscriber<any>> this.destination).addParentTeardownLogic(this);
  } else {
  this.syncErrorThrowable = true;
  this.destination = new SafeSubscriber<T>(this, <PartialObserver<any>> destinationOrNext);
@@ -85,13 +85,14 @@ export class Subscriber<T> extends Subscription implements Observer<T> {
  }
  }
 
- addParentTeardown(parentTeardown: TeardownLogic) {
- this.parentSubscription.add(parentTeardown);
- this.add(this.parentSubscription);
+ addParentTeardownLogic(parentTeardownLogic: TeardownLogic) {
+ this.parentSubscription = this.add(parentTeardownLogic);
  }
 
  unsubscribeParentSubscription() {
- this.parentSubscription.unsubscribe();
+ if (this.parentSubscription !== null) {
+ this.parentSubscription.unsubscribe();
+ }
  }
 
  /**
@@ -167,7 +168,7 @@ export class Subscriber<T> extends Subscription implements Observer<T> {
  this.isStopped = false;
  this._parent = _parent;
  this._parents = _parents;
- this.parentSubscription = new Subscription();
+ this.parentSubscription = null;
  return this;
  }
 }

diff --git a/src/observable/dom/WebSocketSubject.ts b/src/observable/dom/WebSocketSubject.ts
@@ -250,9 +250,8 @@ export class WebSocketSubject<T> extends AnonymousSubject<T> {
  if (!this.socket) {
  this._connectSocket();
  }
- let subscription = new Subscription();
- subscription.add(this._output.subscribe(subscriber));
- subscription.add(() => {
+ this._output.subscribe(subscriber);
+ subscriber.add(() => {
  const { socket } = this;
  if (this._output.observers.length === 0) {
  if (socket && socket.readyState === 1) {
@@ -261,7 +260,7 @@ export class WebSocketSubject<T> extends AnonymousSubject<T> {
  this._resetState();
  }
  });
- return subscription;
+ return subscriber;
  }
 
  unsubscribe() {