Some items were missed when using PublishSubject.buffer().subscribeOn()

[Drteammangel]

I wrote code as follow:

import io.reactivex.schedulers.Schedulers;
import io.reactivex.subjects.PublishSubject;
import io.reactivex.subjects.Subject;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.stream.IntStream;

public class RxTest2 {

    private final Subject<Integer> subject = PublishSubject.<Integer>create().toSerialized();

    public RxTest2() {
        subject.buffer(10, TimeUnit.SECONDS, Schedulers.computation(), 10, ArrayList::new, true)
                .subscribeOn(Schedulers.computation())
                .subscribe(this::consume);
    }

    public void consume(List<Integer> list) {
        if (list.isEmpty()) return;
        Collections.sort(list);
        System.out.println(list);
    }

    public void onNext(Integer o) {
        this.subject.onNext(o);
    }

    public static void main(String[] args) {
        RxTest2 rxTest = new RxTest2();
        ExecutorService executorService = Executors.newSingleThreadExecutor();
        IntStream.range(0, 100).forEach(i -> executorService.execute(() -> rxTest.onNext(i)));
    }
}

Sometime, it prints nothing, or:

[65, 66, 67, 68, 69]
[70, 71, 72, 73, 74, 75, 76, 77, 78, 79]
[80, 81, 82, 83, 84, 85, 86, 87, 88, 89]
[90, 91, 92, 93, 94, 95, 96, 97, 98, 99]

But when I place subscribeOn(...) before buffer(...), or use observeOn(...), I get correct print.

place subscribeOn(...) before buffer(...)

subject.subscribeOn(Schedulers.computation())
        .buffer(10, TimeUnit.SECONDS, Schedulers.computation(), 10, ArrayList::new, true)
        .subscribe(this::consume);

use observeOn(...) before buffer(...)

subject.observeOn(Schedulers.computation())
        .buffer(10, TimeUnit.SECONDS, Schedulers.computation(), 10, ArrayList::new, true)
        .subscribe(this::consume);

use observeOn(...) after buffer(...)

subject.buffer(10, TimeUnit.SECONDS, Schedulers.computation(), 10, ArrayList::new, true)
        .observeOn(Schedulers.computation())
        .subscribe(this::consume);

the correct print:

[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
[10, 11, 12, 13, 14, 15, 16, 17, 18, 19]
[20, 21, 22, 23, 24, 25, 26, 27, 28, 29]
[30, 31, 32, 33, 34, 35, 36, 37, 38, 39]
[40, 41, 42, 43, 44, 45, 46, 47, 48, 49]
[50, 51, 52, 53, 54, 55, 56, 57, 58, 59]
[60, 61, 62, 63, 64, 65, 66, 67, 68, 69]
[70, 71, 72, 73, 74, 75, 76, 77, 78, 79]
[80, 81, 82, 83, 84, 85, 86, 87, 88, 89]
[90, 91, 92, 93, 94, 95, 96, 97, 98, 99]

I reproduce it with rxjava 2.2.x and 3.x.

I'm not sure if it is same with the one on so or maybe I use it in a wrong way.

[akarnokd]

PublishSubject does not wait for a subscriber and signals can be lost, especially if you use subscribeOn. Use UnicastSubject or ReplaySubject instead.

[Drteammangel]

PublishSubject does not wait for a subscriber and signals can be lost, especially if you use subscribeOn. Use UnicastSubject or ReplaySubject instead.

But subscriber has been initialized before emitting items.
And why the other three ways don't lost items?

[akarnokd]

As I mentioned, using subscribeOn can cause items to be lost because the PublishSubject simply won't see a subscriber in time you start emitting items. There is no reason in general to use subscribeOn with PublishSubject.

[Drteammangel]

Thanks for you reply, now I know the features of PublishSubject and I'm going to use UnicastSubject instead.
What's the difference between PublishSubject.buffer().subscribeOn() and PublishSubject.subscribeOn().buffer()?

[akarnokd]

Not much from the end user's perspective.

[Drteammangel]

OK.
Btw, I tested on buffer(int count) and other overload buffer(...), It seems that buffer with ObservableBufferTimed and subscribeOn will delay the subscriber being seem, so that together with PublishSubject causes the problem. Maybe it's similar with issue#534