delaySubscription(Observable) breaks upstream unsubscription

[hannesstruss]

This test case is failing (tested with 1.1.2):

  @Test
  public void testWithSubjects() {
    PublishSubject<Integer> delayUntil = PublishSubject.create();
    PublishSubject<Integer> interrupt = PublishSubject.create();
    final AtomicBoolean subscribed = new AtomicBoolean(false);

    Observable.just(1)
        .doOnSubscribe(() -> subscribed.set(true))
        .delaySubscription(delayUntil)
        .takeUntil(interrupt)
        .subscribe();

    interrupt.onNext(9000);
    delayUntil.onNext(1);

    assertFalse(subscribed.get());
  }

I stumbled upon this using Completable.andThen (which delegates to delaySubscription).

[akarnokd]

TakeUntil subscribes to its source before the subscription to upstream could happen. This is partly due to how lift works and partly to allow interrupting a synchronous source as it would otherwise possibly rush through without letting the operator subscribe to the until source.

[hannesstruss]

In other words, takeUntil only guarantees to suppress emission of items given a condition, but does not guarantee what happens upstream?

(I was confused, as the analogous example with time based delaySubscription works:)

  @Test
  public void testWithSubjects() {
    PublishSubject<Integer> interrupt = PublishSubject.create();
    final AtomicBoolean subscribed = new AtomicBoolean(false);
    TestScheduler testScheduler = new TestScheduler();

    Observable.just(1)
        .doOnSubscribe(() -> subscribed.set(true))
        .delaySubscription(1, TimeUnit.SECONDS, testScheduler)
        .takeUntil(interrupt)
        .subscribe();

    interrupt.onNext(9000);
    testScheduler.advanceTimeBy(1, TimeUnit.SECONDS);

    assertFalse(subscribed.get());
  }

[akarnokd]

Interesting. The second case shouldn't pass either. I'll investigate.

[akarnokd]

Okay, I was wrong. Both tests should pass because they verify that the subscription doesn't happen - as expected. There is a bug in the non-timed delaySubscription that doesn't propagate the unsubscription properly. I'll post a PR to fix that.

[akarnokd]

Fix posted: #3845

[hannesstruss]

This works in 1.1.3, thanks so much! ✨