Observable.subscribe checks 'instanceof' Subscription instead of shape

[ntilwalli]

Currently Subject extends Observable and only implements Subscription so it doesn't see the Subscription function on it's inheritance chain. This is an issue because currently the subscribe method on Observable checks whether the subscriber is an instanceof Subscription. If the subscriber is a Subject it returns false and then wraps subscriber in a Subscriber and proceeds.

      if (observerOrNext instanceof Subscriber) {
        subscriber = (<Subscriber<T>> observerOrNext);
      } else {
        subscriber = new Subscriber(<Observer<T>> observerOrNext);
      }

This seems faulty because this behavior will hide any attempt to override any of the Subscriber methods in the Subject.

export class CustomSubject<T> extends Subject<T> {
  constructor(private _numObservables: number = 0) {
    super();
  }

  // Never gets called
  add(subscription?) {
    console.log("Add called")
  }
}

I was trying to create a custom Subject that reference counts the number of Observables it is subscribed to so it can defer calling complete until the last observable completes but this became an issue. I don't know if this is a bug, but it seems like non-ideal behavior. Would it be possible to add Subscriber to the prototypal chain of Subject, or, in Observable.subscribe change the instanceof call to instead check the shape of the passed in subscriber?

[staltz]

I would rather first question your overall approach. Subclassing Subject is not meant to be a common necessity. In fact, I have never encountered such case where I had to subclass Subject.

I was trying to create a custom Subject that reference counts the number of Observables it is subscribed to so it can defer calling complete until the last observable completes but this became an issue.

Seems fishy. Why can't you merge all the Observables to be observed by the Subject, and then subscribe to that merged result using the Subject as observer? Merge gives you this functionality of defer calling complete until the last observable completes:

    var e1 =    hot('--a-----b-----c----|');
    var e2 =    hot('-----d-----e-----f---|');
    var result = e1.merge(e2, rxTestScheduler);
    var expected =  '--a--d--b--e--c--f---|';

In other words, what are you trying to achieve? We might be able to find a different solution to your problem.

[ntilwalli]

Sure. This was more of a thought exercise for me to understand Subjects, but I want to simulate a multiplexed Websocket. That is, I want to create a class that exposes a shared observable. The items in the output sequence will all conform to a type T which has a tag property that allows the subscribers to filter based on tag and generate multiple observables which emulate multiple simultaneous WebSocket connections. I'd like this class to emulate the send behavior of a WebSocket by accepting a message that requests for a certain tag of data. Internally, the class should create a new source observable for that tag and merge it with the already existing observables that are sourcing the output observable. So the observables that are being merged are being created dynamically at random times (via setTimeout), so they can't all be merged statically at startup. The output observable, from the perspective of a subscriber, should never change. Subscribers should not have to resubscribe to a new observable when a new tag is requested/brought online.

Someone pointed me to your https://github.com/cyclejs/rx-injectable-observable, which may be exactly what I want and should use, but I wanted to build something rudimentary with Subjects on my own for simple learning purposes (I'm quite new to RxJS). :) The fact that I could't get my CustomSubject's add override to get called by the subscribe method seemed like an issue.

[benlesh]

Honestly, as a first attempt, the best way to create a custom subject right now is to use Subject.create and pass to it any Observer-shaped object, and any Observable. It will do the work of "gluing them together" into a Subject.

Otherwise feel free to try to extend Subject, but know that playing with things like add is playing with fire, as that hooks into the unsubscription chain.

On the instanceof versus "shape of" front. We use instanceof because in most cases is performs better than checking the shape of an object. This is particularly true in some of the resource constrained, non-JITed runtimes we have to deal with at Netflix. As such, I'm unlikely to be okay with a movement toward property checking. We're probably already doing that in a few places were we shouldn't be (isScalar comes to mind).

[trxcllnt]

@staltz @Blesh this is a problem because subscribing to an Observable with a Subject (as the Observer) is a very common use-case. We definitely need to change Observable's public subscribe implementation to support this.

[staltz]

subscribing to an Observable with a Subject (as the Observer) is a very common use-case.

Subscribing to an Observable with a Subject should be fine, seems covered. This issue is about subclasses of Subject.

[ntilwalli]

Thanks all for getting back. I recognize that overriding subjects is not the best approach in most cases and I'm not using it anymore myself so this issue doesn't actually affect me, but couldn't the Observable.prototype.subscribe implementation just be changed to check if the subscriber is also an instanceof Subject? Like this:

      if (observerOrNext instanceof Subscriber || observerOrNext instanceof Subject) { 
        subscriber = (<Subscriber<T>> observerOrNext);
      } else {
        subscriber = new Subscriber(<Observer<T>> observerOrNext);
      }

Anywhere in the code that tests for Subscriber-ness could just do this check, which seems like it would be both efficient and correct.

[staltz]

@ntilwalli indeed, that makes sense. I wonder if there is any reason to wrap a Subject in a Subscriber. I'm curious to see what performance boost would your suggestion bring.

[benlesh]

I think that I've fixed this with #724, but inadvertently (I wasn't trying to fix it, I was fixing something else), and in a way that's probably less desirable than @ntilwalli's proposal.

[ntilwalli]

This issue is actually not fixed. I've created a PR integrating my suggestion. #863

[lock]

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.