feat(takeUntil): change behavior of takeUntil to not subscribe to source if notifier is synchronous

[trxcllnt]

Description:
If the notifier supplied to takeUntil synchronously emits a value, the source should not be subscribed to.

Related issue (if exists):
#1360

cc: @mattpodwysocki

Side note: subscribeToResult should definitely not return null cc: @Blesh @jayphelps

[coveralls]

Coverage increased (+0.001%) to 97.688% when pulling 0e669b8 on trxcllnt:fix-takeuntil into 89b506d on ReactiveX:master.

[benlesh]

So looking at this, it changes the behavior of takeUntil from any previously existing version of RxJS. The effect is that basically you're not going to subscribe to the source observable at all if the takeUntil notifier synchronously emits.

While that's definitely a performance tweak, there's a tradeoff being made here. A previous version of RxJS would have subscribed to the source observable. Subscription is a common point at which side effects occur. For example, in the scenario that the source was an ajax POST. The data would have been posted. Maybe the teardown wants to "unpost" it by rolling back, maybe it doesn't. The point is if there's a takeUntil ala this PR on it, that's hitting some sync notifier (a BehaviorSubject maybe?) it's not going to perform the post at all.

It's definitely not a "fix" per se... it's more of a feature change.

@mattpodwysocki @staltz I'd be interested in your thoughts here.

[trxcllnt]

Hm, I may have messed up this PR. I was going off memory of the Rx4 behavior, which we currently don't have either. Though it seems like this may be what RxJava does?

[staltz]

I'm generally favorable to this change, I think it makes more sense. But I don't have a hard opinion. It would be a breaking change, and this needs to be clearly communicated. For the record, most.js subscribes to the notifier before it subscribes to the source, and so does xstream.

I think it's mostly a question of how to manage the breaking change and communication. Looking at the use cases, most of the cases where both source and notifier are synchronous should intuitively yield an empty Observable. And we don't have something like source.doOnSubscribe(sideEffect). If we did, that would be a legit use case for not passing this PR.

[benlesh]

For the record, most.js subscribes to the notifier before it subscribes to the source, and so does xstream

I'm totally in favor of the notifier being subscribed first. But I hesitate to not subscribe to the source if the notifier synchronously emits. I don't think most people would expect that behavior. The act of subscription is a common place for side effects and this would completely change the behavior with takeUntil to not trigger side effects that might be important to some systems.

[staltz]

Can you come up with a compelling example? I tried, and couldn't come up with something non-silly.

[trxcllnt]

@staltz doOnSubscribe is lift :)

@Blesh initially I was concerned, but the more I think about it, it seems more correct that the source Observable is never subscribed. For example, what if the source makes an AJAX request, and the takeUntil notifier is sync. Should we initiate the AJAX request, only to synchronously cancel it once the stack unwinds back to the TakeUntilSubscriber?

You might ask why would someone ever pass takeUntil a notifier that synchronously emits? In my use-case, I was multicasting a scroll event through a ReplaySubject and using concat + takeUntil to say "do some stuff, then concat on an AJAX request, but don't do (or cancel) the AJAX request if the ReplaySubject emits a scroll event."

[benlesh]

Can you come up with a compelling example? I tried, and couldn't come up with something non-silly.

@staltz I did in my original comment.

It's as simple as a post() ajax request composed with a takeUntil.

someStreamOfUpdate$.flatMap(stuff => $.ajax.post(stuff))).takeUntil(component.teardown$)

In the above, if both someStreamOfUpdate$ and component.teardown$ emit synchronously, we'd really be breaking the current behavior. Because currently it would still post at least one update, and the new way it would not. Now maybe this is something people might want, but we can't be sure. Generally, as a rule, everyone assumes operators will subscribe to their sources when you subscribe to the result. For consistency, I think we should stick with that. Do we have a precedent for this non-subscribing behavior?

[trxcllnt]

@Blesh yes, take(0) doesn't subscribe to the source. this is the same use case.

[benlesh]

@trxcllnt since you can't know if a given observable is sync or async, take(0) and takeUntil(foo) are two different things.

[benlesh]

I'm not against the change... but I would find it a little disturbing if no one thought of the implications of this breaking change.

[mattpodwysocki]

@trxcllnt I'm against this change for the very reasons that @Blesh stated that there may be side effects that would have been expected higher in the chain due to expecting both the source and notifier to be subscribed to.

[trxcllnt]

@Blesh takeUntil with a (possibly) sync Observable is a way to defer the take(x) behavior until subscription, where x is a value you don't know at creation. It's like saying, "I'm going to take some values, but I don't know how many until this other Observable emits." So the whole point is that, yes, you can't know whether an Observable will be sync or async, but if it is sync, it should act like take(0).

To illustrate, I'll use everyone's favorite functional programming example, the fibonacci sequence! Here's a simple fibonacci sequence Observable:

function fibs(period = 0) {
  return Observable
    .interval(period)
    .scan([i, n] => [n, i + n], [1, 0])
    .pluck(0)
}

If you have a start button and user input field for the number of values to compute ahead of time, you can print 0-Infinity values from the fibonacci sequence over some period of time:

starts.switchMap(() =>
  fibs().take(textInput.value)
).subscribe(::console.log)

But what if you have a start button and some reactive state, let's say a BehaviorSubject of checkbox values, that you can't know until subscribe? Something like:

starts.switchMap(() =>
  fibs().takeUntil(stops.filter(Boolean))
).subscribe(::console.log)

In the first example take(0) doesn't subscribe to the fibs() sequence, because taking zero values from the sequence has identical behavior as never subscribing in the first place. This is a safe assumption to make because Observable subscription is referentially transparent.

takeUntil is the temporal (time-respective) dual of take, so it would seem takeUntil with a sync Observable is the temporal dual of take(0). The fact that the operators aren't congruent in this way is a bug, albeit one that seems to have gone unnoticed for a long time.

p.s. it also seems the reason this wasn't an issue in Rx < 5 was the default trampoline scheduling. If the notifier had been subscribed to first, the trampoline scheduler would have ensured that it didn't emit until after the stack had unwound, ensuring the takeUntil subscriber didn't yet know not to subscribe to the source.

[trxcllnt]

and I'm shocked @mattpodwysocki is here arguing for more side-effects! ;-> but more curious if you have counter examples where it's imperative that takeUntil does subscribe to the source when given a sync notifier Observable.

[trxcllnt]

@Blesh and thinking through your example of post(), I can actually argue the counter-point that doing the post isn't desired. If the post happens, and we immediately cancel it due to the sync notifier, the browser will cancel/drop the request.

If the browser were smart enough to optimize against sending AJAX requests that are aborted in the same tick of the event loop, then we'd be fine. If not (and I think this is the case with HTTP post), you've just wasted the server's time telling it to do something that you then immediately want it to not do. This could have catastrophic implications for systems that don't have a way to roll-back from the transaction, or just good ol' performance problems for systems that do. Either way it doesn't seem like something we should do.

[benlesh]

it is sync, it should act like take(0).

Actually, I think take(0) should subscribe to the source observable too, honestly. If for no other reason than consistency. I can see what you're saying, but in my mind it's an unnecessary breaking change.

As far as a counter point... what happens to people who expect: Observable.of(1, 2, 3).takeUntil(Observable.of(1)) to emit 1 synchronously (with trampoline scheduling)? That's the behavior of every other version of RxJS.

I can tell you're passionate about this change. I just don't think it's right that we change it. Probably shouldn't make a breaking change to support what seems to be an edge case that will break someone else's edge case.

[trxcllnt]

@Blesh Actually, I think take(0) should subscribe to the source observable too, honestly. If for no other reason than consistency. I can see what you're saying, but in my mind it's an unnecessary breaking change.

It could, but it'd defeat the purpose of shortcut fusion.

Consistency is great, and we're both arguing for more of it.

In the absence of a supplied scheduler, we've explicitly chosen to employ recursive scheduling semantics. This has always been acknowledged as a backwards-incompatible breaking change, but necessary for performance, and still correct as long as all the operators implement it consistently.

We agree that right now, depending on how you look at it, either take or takeUntil needs to change to be consistent. I believe takeUntil needs to change in order to be correct with respect to immediate dispatch, thus consistent with the rest of the library.

Think of it in terms of subscription order: given source.map().takeUntil().subscribe(), subscription order is "takeUntil -> map -> source". It makes total sense the takeUntil logic (including the notifier subscription) runs before the source subscription. Indeed, takeUntil has been working this way in Rx5 since it was written, so we're not breaking that. We're just doing less work if that's what the customer wants.

As far as a counter point... what happens to people who expect: Observable.of(1, 2, 3).takeUntil(Observable.of(1)) to emit 1 synchronously (with trampoline scheduling)?

It's possible to do this with observeOn. It isn't possible to get the behavior in this PR any other way. Not subscribing to the source is also clearly better for performance, which is a primary goal of this project.

That's the behavior of every other version of RxJS.

If this was more important than performance, we wouldn't have switched to recursive scheduling by default. Slight deviations in operator internals (such as subscription order) from past versions of Rx have already been made, and are to be expected.

[coveralls]

Coverage increased (+0.001%) to 97.675% when pulling a1b18ec on trxcllnt:fix-takeuntil into 5f93f81 on ReactiveX:master.

[benlesh]

@trxcllnt haha... You've almost sold me. I think that I'd be okay with including this in the next version of RxJS, but I'm not okay implementing this as a "fix", because it's really a breaking change to a feature of the library. If anything, altering take(0) would be a "fix" at this point, because we didn't really want to break the semantics that had from Rx4. But I think that, maybe, possibly, I'm okay with implementing this is v6... @mattpodwysocki what do you think?

[mattpodwysocki]

@Blesh, no objection to moving it to the next major.

[trxcllnt]

@Blesh I updated this PR to target the next branch

[benlesh]

We need to get these changes to target what's in master now. I almost completely forgot about this PR. LOL

[trxcllnt]

@benlesh is this not making it into v6?

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