PublishSubject with subscribeOn ==> loss of data during onNext calls

[OleksandrKucherenko]

Library version: 1.3.0

Run Test N times (I try at least 10), flaky behavior. Very often subscribers do not receive the value:

    private final Executor mExecutor = Executors.newFixedThreadPool(20);

    @Test
    public void testMultiThreadingAssignment() throws Exception {
        final Subject<Object, Object> subject = PublishSubject.create();

        final CountDownLatch countdown = new CountDownLatch(2 /* subscribers */ * 2 /* values */);
        final List<Subscription> subscriptions = new ArrayList<>();

        final Observable<Object> caller = subject.asObservable()
                .subscribeOn(Schedulers.from(mExecutor))
                .doOnNext(new Action1<Object>() {
                    @Override
                    public void call(final Object lookup) {
                        logTo(_log).info("-- @%s", lookup.hashCode());
                    }
                })
                .observeOn(Schedulers.from(mExecutor))
                ;

        final Action1<Object> onNext = new Action1<Object>() {
            @Override
            public void call(final Object lookup) {
                countdown.countDown();
                logTo(_log).info("%s @%s", lookup, lookup.hashCode());
            }
        };

        // 2 subscriptions
        subscriptions.add(caller.subscribe(onNext));
        subscriptions.add(caller.subscribe(onNext));

        // 2 values
        subject.onNext(new Object());
        subject.onNext(new Object());

        assertThat(countdown.await(1, TimeUnit.SECONDS))
                .describedAs("counter %s", countdown.getCount())
                .isTrue();

        // verify that subscriptions are still in active state
        assertThat(subscriptions.get(0).isUnsubscribed()).isFalse();
        assertThat(subscriptions.get(1).isUnsubscribed()).isFalse();

        // cleanup, free threads
        for (final Subscription s : subscriptions) {
            s.unsubscribe();
        }
    }

I play with test and found that the reason of failure is only one line:

                .subscribeOn(Schedulers.from(mExecutor))

replacement of subscribeOn by observeOn make test working without any problem and return expected result.

[OleksandrKucherenko]

looks like the same issue: #2805

[OleksandrKucherenko]

[akarnokd]

And the answer is the same. There is a race between PublishSubject getting subscribed to asynchronously and sending the subject values. subscribeOn delays the subscription a tiny bit so probabilistically you are concurrently onNext-ing a subject that has no subscribers.

[OleksandrKucherenko]

Is it possible to use something similar to PublishSubject but with fixed race condition behavior ?!

[akarnokd]

Don't use subscribeOn with PublishSubject use ReplaySubject or UnicastSubject to hold onto the data until there is a subscriber.

[OleksandrKucherenko]

Thanks!