Rxjs: takeUntil exhibits weird behavior when notifier is a previously completed connectable observable

[beverlycodes]

Bug Report

Current Behavior

Some connectable observables when used as takeUntil notifiers seem to cause takeUntil's source observable to enter a sort of hung state where nothing will be emitted and in most cases no completion will happen either. Appears to only happen when the source observable is created after the notifier has completed.

Reproduction

const test1$ = of('test').pipe(publish())
const test2$ = of('test').pipe(take(1), publish())
const test3$ = from(['test']).pipe(publish())

test1$.connect()
test2$.connect()
test3$.connect()

// control case
setTimeout(function () {
  of('test').pipe().subscribe(
    () => console.log("control onNext: This gets printed once"),
    null,
    () => console.log("control onComplete: This gets printed")
  )
}, 1000)

// test1$
setTimeout(function () {
  test1$.subscribe(() => console.log('test1$ emit'), null, () => console.log('test1$ complete'))

  of('test').pipe(takeUntil(test1$)).subscribe(
    () => console.log("test1 onNext: test1$ did not emit, so this should get printed, but it does not"),
    null,
    () => console.log("test1 onComplete: This does get printed, but should have been preceded by the single emission of the source")
  )
}, 1000)

// test2$
setTimeout(function () {
  test2$.subscribe(() => console.log('test2$ emit'), null, () => console.log('test2$ complete'))

  of('test').pipe(takeUntil(test2$)).subscribe(
    () => console.log("test2 onNext: test2$ did not emit, so this should get printed, but does not"),
    null,
    () => console.log("test2 onComplete: This should get printed, but does not")
  )
}, 1000)

// test3$
setTimeout(function () {
  test3$.subscribe(() => console.log('test3$ emit'), null, () => console.log('test3$ complete'))

  of('test').pipe(takeUntil(test3$)).subscribe(
    () => console.log("test3 onNext: test2$ did not emit, so this should get printed, but does not"),
    null,
    () => console.log("test3 onComplete: This should get printed, but does not")
  )
}, 1000)

Expected behavior

The beta docs indicate that if takeUntil's notifier never emits and then completes, takeUntil is supposed to allow all values. Supplying takeUntil with NEVER confirms this reading. Based on that understanding:

• In the test1 example above, takeUntil should have allowed an emission, but at least allowed the expected completion.
• In the test2 and test3 examples above, takeUntil should have allowed emissions, and somehow didn't even allow completions.
• In all three tests, the notifier for takeUntil definitely completed before the test observables were created.

My assumption is that all three test cases should have behaved identical to the control case, in which a single value was emitted, followed by a completion.

This would not be the first time I've been bitten by some misunderstood nuance of hot observables, but as far as I can tell the hot observable is behaving exactly as I expect it to, but takeUntil is not.

Environment

• Runtime: Node v8.5.0
• RxJS version: 6.2.1

[kwonoj]

Before getting into details, is this happens same with asynchronous sources other than synchronous ones? operators like takeuntil usually do not play well with synchronous soruces (cause it's being observed synchronously, literally) so curious about those.

[beverlycodes]

@kwonoj I tried with timer(0, 100).pipe(take(1), publish()) and got the same results as the test2 and test3 cases.

EDIT: In all cases, there's a publish at the end of the notifier pipe followed by a connect, so what is being passed to takeUntil is always a connectable, at least from how I'm seeing things. And all of my tests behave correctly if I remove the connect, or if I make sure that the notifier doesn't emit until after the 1000ms timeouts have run.

[bene-starzengruber]

I'm experiencing the same issue in one of our bigger applications where this pattern is kind of common.
The source observable never emits or completes, no matter if it's sync or async

let subject$ = new Subject();
subject$.complete();

interval(1000).pipe(takeUntil(subject$))
.subscribe(console.log, console.error, () => console.log('completed'));

of(1, 2, 3).pipe(takeUntil(subject$))
.subscribe(console.log, console.error, () => console.log('completed'));

I would expect that the source observable completes directly in this case

[cartant]

FYI, I started looking into this a short while ago. The reported behaviour seems to be related to two issues one of which is what @bene-starzengruber has pointed out: if the notifier has completed, the source is not even subscribed to. I'll add a comment with everything that I've found once I've spent some more time with it and have completely figured out what's going on.

[bene-starzengruber]

@cartant

if the notifier has completed, the source is not even subscribed to

so does that mean, that there is no risk of creating a memory leak this way?

[cartant]

@bene-starzengruber From what I've seen, I doubt there'll be a problem with it leaking, as it doesn't subscribe. I'll be able to say more once I fully understand what's going on. TBH, the part that I cannot explain yet is the difference in behaviour between a notifier that's scalar and one that's not. I'll get to the bottom of it. Eventually.

[BioPhoton]

@cartant as some time passed since the issue was created and you were looking into it I thought I could ask for feedback on how to proceed with this.

[benlesh]

This is no longer an issue in 6.5 and higher.