rxjs - Subjects created with Subject.create can't unsubscribe

Question

I have a subject that is responsible for subscriptions to certain observable:

var timer$ = Rx.Observable.timer(1000, 2000);

When the subject is linked to the subject like that

var timerSubject = new Rx.Subject;
timer$.subscribe(timerSubject);

var subscription1 = timerSubject.subscribe(n => console.log(n));
var subscription2 = timerSubject.subscribe(n => console.log(n));

setTimeout(() => timerSubject.unsubscribe(), 4000);

everything is fine, timerSubject.unsubscribe() can be called once and the subscriptions shouldn't be unsubscribed one by one.

When the subject is created with Subject.create like that (a plunk)

var timerSubject = Rx.Subject.create(null, timer$);

var subscription1 = timerSubject.subscribe(n => console.log(n));
var subscription2 = timerSubject.subscribe(n => console.log(n));

setTimeout(() => timerSubject.unsubscribe(), 4000);

timerSubject.unsubscribe() does nothing, while I would expect to behave it the same as in the first snippet.

If Subject.create creates a subject that can't even unsubscribe, what's the purpose of Subject.create then?

Why does this happen? Is this a bug?

How can the subject should be created to reach the desired behaviour?

It is reproducible with RxJS 5 RC1.

Answer 1

I checked the source code for Subject.create() and it's not the same as calling new Subject().

• Subject.create() returns an instance of AnonymousSubject.

• new Subject() returns an instance of Subject.

So it seems the problem why unsubscribe() on AnonymousSubject doesn't work is because it in fact never subscribes. It just keeps a reference to the source Observable and when subscribing an Observer it connects directly source with the Observer and doesn't keep track of created subscribtions.

In your case when you call timerSubject.subscribe() it subscribes directly to timer$ and AnonymousSubject acts only as mediator.

I don't know whether this is by design or it's a bug. However, the first option is more likely I think.