Rxjs: Subscription.add should return the subscription.

[Dorus]

It would be really nice if we could chain subscriptions with .add().

Previously i wrote the following code:

function publish(source, selector) {
  return Rx.Observable.create(ob => {
    var xs = source.publish();
    return selector(xs).subscribe(ob).add(xs.connect());
  });
}

But this is actually wrong. It would be nice if this would be possible. Currently we need to write:

function publish(source, selector) {
  return Rx.Observable.create(ob => {
    var xs = source.publish();
    var sub = selector(xs).subscribe(ob)
    sub.add(xs.connect())
    return sub;
  });
}

[david-driscoll]

It looks like it's falling into this trap: https://github.com/ReactiveX/rxjs/blob/master/src/Subscription.ts#L133

We should probably return this in that scenario.

[benlesh]

We should probably return this in that scenario.

Actually it should return Subscription.EMPTY. The subscription returned is the subscription made to wrap the teardown (if necessary).

See the docs here

(Also @david-driscoll, remember to hit Y when copying a link to lines of code in github to get the exact commit SHA in the url)

[david-driscoll]

Thanks for the tip, didn't know that! :smile:

Wouldn't there be two cases? In the above scenario, if they pass in a specific subscription, shouldn't it return that subscription? It's entirely possible I'm missing something though.

if (!teardown || teardown === Subscription.EMPTY)
    return Subscription.EMPTY;

if (teardown === this)
    return this;

[BrainCrumbz]

Hello all, been pointed here by @Dorus , thanks for that. Don't know if the following request fits into current issue, let's see:

We're working with RxJs 5 in TypeScript, coming from a C# Reactive Extensions background. I'm looking on how to _extend_ (as in C# Extension Methods) the Subscription class.
I'd like to provide a new (instance) method Subscription.addTo(collector: Subscription) which adds the instance to passed collector Subscription.
Aim is to be able to write this type of code:

class SomeClass {
  constructor(/*...*/) {
    someObservable$.subscribe(value => /*...*/).addTo(this.collector);
    anotherObservable$.subscribe(value => /*...*/).addTo(this.collector);
  }

  onSomeDestroy() { // e.g. in Angular2
    this.collector.unsubscribe();
  }

  private collector: Subscription = new Subscription();
}

At the moment we have a workaround that extends Subscription.prototype and adds the method also to a custom ISubscription interface, but it's not the best thing ever.

Ideally, this would just be an additional method on Subscription:

addTo(collectorSub: Subscription): void {
  collectorSub.add(this);
}

[matthewwithanm]

Related (dupe?): #1583

[david-driscoll]

You could use module augmentation, in your local files to "add" it to the prototype.

That's how operators are composed onto the observable classes today. https://github.com/ReactiveX/rxjs/blob/ceb99904c6b56a5d2e832fdd9e07dcf58cb2ad3d/src/add/operator/audit.ts

In the file you augment the prototype you would do something like:

declare module 'rxjs' {
  interface Subscription {
    addTo(collectorSub: Subscription): void {
      collectorSub.add(this);
    }
  }
}

[BrainCrumbz]

@matthewwithanm also checked that issue. What we're after in my team, all in all, is to be able to collect all subscriptions created inside c'tor, without having to declare, assign and append local subscription variables. So the simplest way (as tried in C# already) seems to append the subscription to a common _container_ as soon as it is returned by subscribe(). It does not seem exactly the same use case as that issue, with arrays of subscriptions, and BTW I find this proposed way more elegant :smile_cat: (just kiddin')

@david-driscoll thanks for the hint. I tried in the past to do module augmentation in our local files, but did not succeed. Currently we're on 5.0.0-beta.3 and node_modules\Subscription.d.ts is defined out of any module, apparently:

export declare class Subscription {
    static EMPTY: Subscription;
    isUnsubscribed: boolean;
    constructor(_unsubscribe?: () => void);
    unsubscribe(): void;
    add(subscription: Subscription | Function | void): void;
    remove(subscription: Subscription): void;
}
export declare class UnsubscriptionError extends Error {
    errors: any[];
    constructor(errors: any[]);
}

I tried your approach, but there's something strange (that makes that fail): adding implementation to an interface?

[BrainCrumbz]

Update
We tried with the following:

// addTo.ts
import {Subscription} from 'rxjs/Subscription';

export function addTo(collectorSub: Subscription): void {
  collectorSub.add(this);
}

export interface AddToSignature {
  (collectorSub: Subscription): void;
}

// SubscriptionExtensions.ts
import {Subscription} from 'rxjs/Subscription';
import {addTo, AddToSignature} from './addTo';

Subscription.prototype.addTo = addTo;

declare module 'rxjs/Subscription' {
  interface Subscription {
    addTo: AddToSignature;
  }
}

which made VS Code happy, but then failed when building:

ERROR in XXXXXXXnode_modulesrxjsSubject.d.ts
(9,22): error TS2420: Class 'Subject' incorrectly implements interface 'Subscription'.
Property 'addTo' is missing in type 'Subject'.

[david-driscoll]

You may have to augment Subject as well. In the latest (I think) Subject no longer extends Subscription, so that goes away with more recent versions.

[BrainCrumbz]

That did the trick, thank you!
For reference and for future readers, here's the working solution:

// addTo.js
import {Subscription} from 'rxjs/Subscription';

export function addTo(collectorSub: Subscription): void {
  collectorSub.add(this);
}

export interface AddToSignature {
  (collectorSub: Subscription): void;
}

// SubscriptionExtensions.js
import { Subscription } from 'rxjs/Subscription';
import { addTo, AddToSignature } from './addTo';

// Create an augmentation for 'rxjs/Subscription' module
declare module 'rxjs/Subscription' {

  // Augment the 'Subscription' class definition with interface merging
  interface Subscription {
    addTo: AddToSignature;
  }

}

// Actually patch 'Subscription' type with added implementation
Subscription.prototype.addTo = addTo;

// 'Subject' implements 'Subscription' as well, so augment that as well

import { Subject } from 'rxjs/Subject';

// Create an augmentation for 'rxjs/Subject' module
declare module 'rxjs/Subject' {

  // Augment the 'Subject' class definition with interface merging
  interface Subject {
    addTo: AddToSignature;
  }

}

// Actually patch 'Subject' type with added implementation
Subject.prototype.addTo = addTo;

// consumer.ts
import { Subscription } from 'rxjs/Subscription';
import './SubscriptionExtensions';
//...
private collector: Subscription = new Subscription();
//...
someObservable$
  .subscribe(value => { /*...*/ })
  .addTo(this.collector);

[matthewwithanm]

@BrainCrumbz Huh? That's exactly the use case of that issue! I was just trying not to prescribe a specific solution when naming it. I'd be fine with a chainable version of add too, but the OP didn't mention arrays or temp vars at all, just a static like

Subscription.of(
  something.subscribe(...),
  somethingElse.subscribe(...),
);

This is basically the solution we're using now in Nuclide but I'd prefer to have something in Rx.

[BrainCrumbz]

Yep, same use case and as you said different way of solving it.

[matthewwithanm]

I've just been bitten by this again…

IMO a mutative method that returns something of the same type that isn't this is asking for trouble. Is it too late to back out of that somehow?

[trxcllnt]

@matthewwithanm IMO a mutative method that returns something of the same type that isn't this is asking for trouble. Is it too late to back out of that somehow?

I tend to agree. I don't think we're depending on this behavior inside Rx anywhere, right @blesh?

[jayphelps]

@matthewwithanm does #1840 look good for you?

[lock[bot]]

This thread has been automatically locked since there has not been any recent activity after it was closed. Please open a new issue for related bugs.