publishBehavior and publishReplay semantics when completed

[staltz]

Continuing the discussion that started here: https://github.com/ReactiveX/RxJS/pull/235/files

RxJS Legacy (v4) passes these tests:

var results1 = [];
var results2 = [];
var subscriptions = 0;

var source = Rx.Observable.create(function (observer) {
  subscriptions++;
  observer.onNext(1);
  observer.onNext(2);
  observer.onNext(3);
  observer.onNext(4);
  observer.onCompleted();
});

var hot = source.shareValue(0);

hot.subscribe(x => results1.push(x));

expect(results1).toBe([0,1,2,3,4]);
expect(results2).toBe([]);

hot.subscribe(x => results2.push(x));

expect(results1).toBe([0,1,2,3,4]);
expect(results2).toBe([]);
expect(subscriptions).toBe(2);

and

var results1 = [];
var results2 = [];
var subscriptions = 0;

var source = Rx.Observable.create(function (observer) {
  subscriptions++;
  observer.onNext(1);
  observer.onNext(2);
  observer.onNext(3);
  observer.onNext(4);
  observer.onCompleted();
});

var hot = source.shareReplay(2);

hot.subscribe(x => results1.push(x));

expect(results1).toBe([1,2,3,4]);
expect(results2).toBe([]);

hot.subscribe(x => results2.push(x));

expect(results1).toBe([1,2,3,4]);
expect(results2).toBe([3,4]);
expect(subscriptions).toBe(2);

Yet RxJS Next behaves basically the opposite way:

Failures:
1) Observable.prototype.publishBehavior() should not emit next events to observer after completed
  Message:
    Expected [ 0 ] to equal [  ].
  Stack:
    Error: Expected [ 0 ] to equal [  ].
      at Object.<anonymous> (./RxJSNext/spec/operators/publishBehavior-spec.js:65:21)
2) Observable.prototype.publishReplay() should emit replayed events to observer after completed
  Message:
    Expected [  ] to equal [ 3, 4 ].
  Stack:
    Error: Expected [  ] to equal [ 3, 4 ].
      at Object.<anonymous> (./RxJSNext/spec/operators/publishReplay-spec.js:98:21)
145 specs, 2 failures
Finished in 12.951 seconds

Because in RxJS legacy there is

    ConnectableObservable.prototype._subscribe = function (o) {
      return this._subject.subscribe(o);
    };

While in RxJS Next:

class ConnectableObservable<T> extends Observable<T> {
  subject: Subject<T>;

  // ...

  _subscribe(subscriber) {
    return this._getSubject().subscribe(subscriber);
  }

  _getSubject() {
    const subject = this.subject;
    if (subject && !subject.isUnsubscribed) {
      return subject;
    }
    // Subject will be recreated
    return (this.subject = this.subjectFactory());
  }

  // ...
}

---

In my opinion, the new behavior in RxJS Next is illogical at first sight, and a potential source for a lot of developer confusion. Besides that, it differs from RxJS legacy features, making migration a potential headache. I think the semantics argument (with regard to subscriptions after complete(), publishReplay should behave just like a ReplaySubject and publishBehavior should behave just like a BehaviorSubject) is straightforward and also aligned with traditional Rx, such as in Rx.NET: check Lee Campbells book on this part.

[benlesh]

In RxJava this behaves as so:

   ConnectableObservable<Integer> conn = just(1).doOnSubscribe(() -> System.out.println("subscribed")).publish();

        conn.subscribe(System.out::println);
        conn.subscribe(System.out::println);

        conn.connect();

        conn.subscribe(System.out::println);
        conn.subscribe(System.out::println);

        conn.connect();    

outputs:

subscribed
1
1
subscribed
1
1

however in RxJS 4 the same thing:

let s = Observable.create(observer => {
  console.log('subscribed');
  observer.onNext(1);
  observer.onCompleted();
})
  .publish();

s.subscribe(::console.log);
s.subscribe(::console.log);

s.connect();

s.subscribe(::console.log);
s.subscribe(::console.log);

s.connect();

outputs only:

subscribed
1
1

RxJS 5(Next) was designed to match the semantics of RxJava, and also to remove the potential footgun from people passing the same subject into two multicast calls.

[staltz]

Isn't multicast less commonly used than the more popular share variants?
And nothing actually stops people from passing the same subject to multicast. The point being, if you're using multicast, you should know what you're doing. If you're using shareFoo or publishFoo, you have a vague yet safe idea of what you're doing.

[benlesh]

The the problem RxJS 5 is trying to solve here is that ConnectableObservables should be "reconnectable", and refCount should return an observable that is cold until subscribed to, then hot until all subscriptions have ended, then cold again.

[staltz]

Why is the lack of reconnectability a problem?

Also, we're talking about publishFoo so far, not yet about refCount. I can understand the refCount argument about from cold to hot and back to cold, but what we have here is simply one connect on a publishReplay observable which behaves oddly.

[staltz]

people passing the same subject into two multicast calls.

I recall Erik Meijer recommending to "make the correct patterns easy and make the wrong patterns possible", so it's easy to detect bad patterns in code reviews. Passing directly a subject to multiple multicasts is easy to catch in code reviews.

[benlesh]

Why is the lack of reconnectability a problem?

If you want to model anything you should be able to reconnect to, say a WebSocket, it's a problem.

Also, we're talking about publishFoo so far

The reason I brought up refCount is it's the behavior of RxJS 2/3/4 regarding ConnectableObservables that makes this impossible. FWIW, I don't think any other flavor of Rx behaves like RxJS in this manner.

I'm not sure what you want is to make ConnectableObservables broken after the the underlying subject has unsubscribed. I think what you want is a specific behavior around publishBehavior, so let's talk about that.

[staltz]

I'm not sure what you want is to make ConnectableObservables broken after the the underlying subject has unsubscribed.

I guess you meant "subject has completed". That's what I mean at least. I think the ConnectableObservable could still be connected. Maybe I should explain what should happen in marble diagrams:

code:       -----C--------------------C?
behaviorsub:     --1-2----3-|
subscriber1: -^----1-2----3-|
subscriber2:                  (^|)
subscriber3: ---------(^2)3-|

Capital C is the connect() moment, ^ is a subscribe moment, and anything between parentheses happens synchronously. Notice that I'm not talking about what happens in subsequent connects, but simply what happens for multiple subscribers related to one connect(). If you assume every BehaviorSubject unsubscribes its subscribers when it completes, maybe we should question that assumption. In RxJS 4 it doesn't. Subscribers must take care of themselves.

If you want to model anything you should be able to reconnect to, say a WebSocket, it's a problem.

Can't this approach below model that?

var reconnectionMoments = // some infinite Observable

reconnectionMoments
  .switchMap(moment => Rx.DOM.fromWebSocket('ws://echo.websockets.org', null))
  .subscribe(/* ... */)

[staltz]

I had a typo/mistake in the marble diagram. Second subscriber should see (^|). Updated now

[benlesh]

Sure that's the behavior of the BehaviorSubject, but ConnectableObservables need to be reconnectable, which means it'll need to recreate the underlying BehaviorSubject.

Can't this approach below model that?

That actually doesn't help, because fromWebSocket returns a Subject, and you'd likely need access to the observer side of the thing.

It's better to be able to say: Rx.DOM.fromWebsocket('ws://echo.websockets.org', null).share() and see a behavior that allowed you to subscribe multiple times without recreating the websocket, then unsubscribe from everyone of those and have the websocket automatically close (because of disposal)... and then be able to reconnect to that websocket simply by subscribing again.

[benlesh]

Question: Why is it important to you that ConnectableObservables never allow reconnection? What use case does that serve?

[benlesh]

I'm not sure how anything passed in those tests above, though.

expect(results1).toBe([0,1,2,3,4]);

would never assert true, because results1 and [0,1,2,3,4] are different instances. You'd need to use toEqual rather than toBe. Unless something has changed in Jasmine.

[benlesh]

... it seems like your issue is specifically with ReplaySubject. Which is the only subject that will actually emit values after it's been disposed/completed/errored.

It seems like the ReplaySubject semantics in RxJS 4/3/2 are actually broken. A Subject that has errored should still emit values. A subject that has completed shouldn't still emit values:

In RxJS 4:

var sub = new Rx.ReplaySubject(2);
sub.subscribe(::console.log, ::console.error);

Rx.Observable.range(0, 3).concat(Rx.Observable.throw('bad')).subscribe(sub);

sub.subscribe(::console.log, ::console.error);

outputs:

0
1
2
error 'bad'
0
1
2
error 'bad'

That doesn't seem right, a Subject that's in error can still emit values?

[staltz]

Sure that's the behavior of the BehaviorSubject, but ConnectableObservables need to be reconnectable, which means it'll need to recreate the underlying BehaviorSubject.

Reconnectability is not my gripe. It's about what happens in the space between the subject's complete and the next connect. In particular, we are discussing what should happen to subscriber2 in this case below:

code:       -----C--------------------C?
behaviorsub:     --1-2----3-|
subscriber1: -^----1-2----3-|
subscriber2:                  ^??
subscriber3: ---------(^2)3-|

I'm saying subscriber2 should see (^|) because it should refer to its latest Observable. For two reasons: what an observer sees after onComplete is important, and an Observer cannot know what comes next. Observers of BehaviorSubject expect to see nothing (that's the BehaviorSubject contract/guarantee), and Observers of ReplaySubject expect to see replayed values. These are guarentees relied on. Second reason about "Observer cannot know what comes next": if subscriber2 is supposed to prepare for the next connection, subscriber2 cannot know if there will eventually be an incoming connect() or not. In the meanwhile, it is certain that subscriber2 could get information from the previous execution of the subject. To which side should we pull subscriber2 to? To refer to the previous or to prepare for the next? Might be a matter of preferred usage style. I have never used or seen the practical need to use connect() multiple times.

That actually doesn't help, because fromWebSocket returns a Subject, and you'd likely need access to the observer side of the thing.

Well then this was an unfortunate example to talk about, because a ConnectableObservable is not an Observer.

and see a behavior that allowed you to subscribe multiple times without recreating the websocket, then unsubscribe from everyone of those and have the websocket automatically close (because of disposal)... and then be able to reconnect to that websocket simply by subscribing again.

This behavior can be built with the switchMap and a publish() (with a one-time connect()), by modifying reconnectionMoments making it emit either true or false, switchMapping to the websocket Observable if true, switchMapping to Observable.never if false. Etc.

Question: Why is it important to you that ConnectableObservables never allow reconnection? What use case does that serve?

Because (refer to the marble diagram above) it serves the use case of giving me a guarantee of what subscribers see after the subject has completed will follow the same conventions as subscribing to an actual subject. These guarantees might be particularly important in Replay case, for instance. I have a cold Observable, I want multiple subscribers to see the same events simultaneously, and after the cold Observable completes, I want it to replay its values to late subscribers. Because that's what ReplaySubjects do. And in case it's a publishBehavior, I do not want it to emit any value after the cold Observable has completed, because that's the BehaviorSubject contract. I do not want to get the initialValue for the next upcoming connect(). And above all, I never had a need for multiple connect(), which is one of the uglier parts of Rx, since it's purely imperative.

expect(results1).toBe([0,1,2,3,4]);

That was wrong indeed. Those tests weren't actually jasmine tests. I just wrote pseudocode. But I'm sure that with console.log results1 is indeed [0,1,2,3,4].

That doesn't seem right, a Subject that's in error can still emit values?

It is correct for two reasons: from the perspective of an Observer, the Observable contract is preserved. Second reason because it's a ReplaySubject and that's what they do: they replay the values and the error.

Your example above of a reconnectable Observable would also "violate" this contract because

let s = Observable.create(observer => {
  console.log('subscribed');
  observer.onNext(1);
  observer.onCompleted();
})
  .publish();

s.subscribe(::console.log);
s.subscribe(::console.log);

s.connect();

s.subscribe(::console.log);
s.subscribe(::console.log);

s.connect();

should emit

subscribed
1
1
subscribed
1
1

And after the first instance, the observable completed, so "why can an Observable which completed still emit values?" Isn't that a Replay feature? ;)

[benlesh]

Well then this was an unfortunate example to talk about, because a ConnectableObservable is not an Observer.

Ah... a side effect of the implementation, I was thinking that this used @trxcllnt's lift override on Subject, which would indeed return a BidirectionalSubject, but it wouldn't for publish()... hmmm...

[benlesh]

In speaking with some RxJava folks, it might be that we just need to implement shareValue and shareReplay without Subjects, and rather just create "1 Producer, Many Consumer" observables for them to return directly. It would probably perform better (fewer object allocations) and it might help us "trim the fat" with our publish operators, as it were.

[staltz]

and rather just create "1 Producer, Many Consumer" observables

That would be a very bold move, would be a big shift in Rx's foundation with regard to cold vs hot.

[benlesh]

would be a big shift

Nah... I'm not talking "all observables" ... just the ones returned from shareReplay or shareValue for example. The Observable returned from refCount is technically a "one producer, many consumers" observable. It's just there's a Subject under the hood. I think we can implement this functionality and skip that completely. Less object allocation, fewer subscriptions.

[headinthebox]

Wow, this is a long thread ;-) But I think that I am with @staltz on this

Reconnectability is not my gripe. It's about what happens in the space between the subject's complete
and the next connect.

Until the next connect any "publish" variant should behave like the underlying subject.

[benlesh]

@headinthebox, I don't disagree with that at all. The thing that @staltz didn't like is if the connectable observable completes, at next connect it recycles the underlying subject. This is how it works in RxJava as well, essentially. RxJS 4/3/2 would instead try to use the same underlying subject, meaning the connectable observable was "single use only". @staltz had come to depend on this behavior with regards to replay subjects, I think. In all discussions with @benjchristensen, @jhusian and many others, it was decided this behavior was "broken".

[Frikki]

This is unexpected new behavior. Could anybody clarify how I can prevent that replay values are dropped once complete? I need to be able to replay what has happened in past. Usually, this would have been a simple publishReplay(), now what?

[staltz]

@Blesh I started the process of migrating Cycle.js to RxJS Next with the help of the community, and we hit our first big problem with this new behavior of connectable Observables. For my purposes, ConnectableObservable is "broken" here. Please consider that a lot of existing codebases out there might depend on how ConnectableObservable works.

Our case goes like this:

• In tests, we use source = Observable.of('foo') to mock an observable. of() gives us (foo|) (next and complete).
• Library code takes source and then does result = source.map(s => stream).publishReplay(1), notice how the map returns a stream (Observable).
• Observable.of() has an optimized .map() which is called immediately, not at subscribe time.
• of().map() has completed because of() completed
• publishReplay(1) kicks in and because of the broken semantics after completion, it cannot replay the value.
• When I call result.connect() nothing happens.

To me, this is a really really big issue, and honestly, I'd consider not using RxJS Next at all in the Cycle.js community. Can you please consider alternatives?

For instance:

The thing that @staltz didn't like is if the collectable observable completes, at next connect it recycles the underlying subject.

Actually it's not recycling at the next connect(), it's recycling on complete. This happens in order in time: (1) first connect(), (2) source completes, (3) second connect(). Recycling is happening on moment (2), not on moment (3). I'd be perfectly ok with recycling on (3). But with it on (2), it goes against BehaviorSubject and ReplaySubject semantics after completed (therefore unintuitive, we're opening possibilities to confuse developers), and also blocks me from accomplishing what I need to accomplish.

The other thing I'd propose is to look for different ways of achieving "reconnection". In my experience I have never had to do multiple calls to connect(). Doing complicated connect() multiple times is IMHO an anti-pattern.

Why can't this approach work for reconnecting? I suggested it above but you didn't comment on it.

and see a behavior that allowed you to subscribe multiple times without recreating the websocket, then unsubscribe from everyone of those and have the websocket automatically close (because of disposal)... and then be able to reconnect to that websocket simply by subscribing again.

This behavior can be built with the switchMap and a publish() (with a one-time connect()), by modifying reconnectionMoments making it emit either true or false, switchMapping to the websocket Observable if true, switchMapping to Observable.never if false. Etc.

Please.

[mattpodwysocki]

@Blesh I'm also with @staltz and @headinthebox on this one. Changing the semantics now would be really confusing and not bring much benefit.

[staltz]

For example this test fails (it timeouts):

  it('publishReplay(1) should replay values just after connect() is called', function (done) {
    var obs = Observable.of('foo').publishReplay(1);
    obs.connect();
    obs.subscribe(
      function (x) { expect(x).toBe('foo'); },
      done.fail,
      done
    );
  });

[Frikki]

Is there any particular reason why this issue hasn’t been labeled discussion? Is it not up for discussion? Have the final words been said and the decision is final?

It seems that those, e.g., @benjchristensen, @jhusain et al, who initially made the decision of this new alignment with RxJava are not even participating here.

[benlesh]

Is there any particular reason why this issue hasn’t been labeled discussion?

Sorry... I didn't label it. :\ heh

It seems that those, e.g., @benjchristensen, @jhusain et al, who initially made the decision of this new alignment with RxJava are not even participating here.

FWIW: I'm also one of those people that made the decision, from discussion with several others really early on in the development of this library who include @headinthebox, @abersnaze, @stealthcoder and @mattpodwysocki... There were changes made around multicast to help prevent certain antipatterns (like reusing the same Subject in two multicast calls) as well as facilitate hot observables being resubscribable. The old behavior was quirky and not friendly to new developers in particular.

However, the side-effect of this change is that shareReplay and shareBehavior now act differently when the refCount returns to zero.

[Frikki]

@Blesh Thanks for the info.

[jhusain]

Seems like this issue shouldn't cause a fork. We can leave the existing multicast operator backward-compatible. In my experience the most common indirect use of multi cast is in the following scenarios:

• share side effects to avoid concurrent identical network requests
• same as the previous scenario except once the data has been retrieved it is forever cached in the Observable

As long as there are two operators that make these scenarios easy, I don't really care what the multicast operator does.

I propose we change share() and avoid using the multicast operator to allow it to be retried. Given it is so commonly used for network requests, this seems like the right decision.

The shareReplay operator doesn't need retry because it's results are cached. It would seem to satisfy the second scenario.

[trxcllnt]

To be clear: the new behavior isn't correct because @Blesh says so, it's correct because it's referentially transparent. Connecting a ConnectableObservable should behave the same way every time, regardless of whether you've connected it before or not.

We also have the option to change multicast to a polymorphic function signature: pass it a function, and it'll re-create the inner Subject on every connection; pass it a Subject instance, and it'll always use that instead. @Blesh I know you want to minimize polymorphic operators, but it may be advantageous to make an exception here.

[benjchristensen]

We also have the option to change multicastto a polymorphic function signature: pass it a function, and it'll re-create the inner Subject on every connection; pass it a Subject instance, and it'll always use that instead.

This is a good middle-ground and the type of thing we've done in RxJava.

[mattpodwysocki]

@trxcllnt no, that's not the way it was designed nor intended to be designed, unless you happened to be in the original design meetings. It's YOUR opinion as to whether it was right or wrong.

[trxcllnt]

@mattpodwysocki that's exactly my point: it's not "correct" because someone decided so. It's correct because it's functionally pure, and the behavior is consistent with the rest of the library.

The Observable referential transparency is beautiful, allowing them to be retried, etc. In the original design, ConnectableObservables aren't referentially transparent; it behaves differently whether you connect it the first time vs. the second.

My argument is the same here as it is in other discussions re: hot-by-default, primarily: if we start from the pure implementation, we can graduate to the impure as-needed. If we start impure, we can never go back to pure.

I'm arguing for this specifically because I've been bitten by repeat/retry on ConnectableObservable failing.

[benlesh]

Ok, so it's going to be a community fork.

OMG drama. Stop. This issue is open for discussion and compromise.

Okay, so here's a proposed compromise:

1. change publish, publishBehavior and publishReplay to match RxJS 4 semantics.
2. multicast can be polymorphic and allow for all behaviors.
3. Have share maintain the RxJS 5 (Next) semantics.
4. Remove shareBehavior and shareReplay to prevent user confusion.

[trxcllnt]

@benlesh what if share took the subjectFactory and/or Subject arg just like multicast, effectively meaning "multicast with this, then refCount."

[benlesh]

what if share took the subjectFactory and/or Subject arg just like multicast, effectively meaning "multicast with this, then refCount."

I'd prefer that share() was a simple call. I view it as the operator for "the masses" to use to get what they would expect out of a multicast observable (retry-able, repeatable, etc)

[TylorS]

I'm no expert of the inner workings of Rx, but in my attempts to update a library from Rx 4.0.6 to Rx 5.0.0-alpha.6 I was unable to recreate the behavior that I came to expect regarding replay() and connect(). Again, I know I'm no expert. I know for sure that, as simply a consumer of the great work you are all creating (seriously thank you!), I would have never come up with "solution" proposed above on my own.

[benlesh]

@TylorS ... my proposal above would bring back that behavior.

Out of curiousity, what exact behavior are you relying on, and why are you relying on it?

[staltz]

Okay, so here's a proposed compromise:

change publish, publishBehavior and publishReplay to match RxJS 4 semantics.
multicast can be polymorphic and allow for all behaviors.
Have share maintain the RxJS 5 (Next) semantics.
Remove shareBehavior and shareReplay to prevent user confusion.

Correct, let's do that and sorry about the community fork comment.

[benlesh]

Correct, let's do that and sorry about the community fork comment.

It's all good... and to show it's all good, I made this emoji kiss poop:

😘💩

The agreed upon changes are merged and happy. We still need more tests around these operators though.

[mattpodwysocki]

Much better answer, thank you!

[headinthebox]

Like!

[donaldpipowitch]

Thanks!

[Frikki]

👍

[HighOnDrive]

Great! Now we can write more articles, like this one: https://medium.com/@fkrautwald/plug-and-play-all-your-observable-streams-with-cycle-js-e543fc287872

[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.