repeating RxJava pipeline with multicasting

The problem in the second example is not in the multicasting. It's in the collect operator and repeatWhen.

Consider repeatWhen like a 'hook' you are connecting to the onComplete method. This 'hook' will intercept the onComplete method and "override" it so it won't be called except for the first time that this observable has completed before starting the repeat process.

Without a onComplete method called the collect operator doesn't know how many items it should collect. So you would have to handle the logic of how to collect the items and where to store them outside of the stream but it would be a workaround.

This is an example for it:

    List<String> test = new ArrayList<>();
    final String[] currentString = {""};

    final int[] i = {0};
    ConnectableObservable<Integer> integerObservable =
            Observable.defer(() -> Observable.fromArray(i[0]++, i[0]++, i[0]++))
                    .doOnComplete(() -> {
                        test.add(currentString[0]);
                        currentString[0] = "";
                    })
                    .repeatWhen(o -> {
                        return o.concatMap(v ->
                                Observable.timer(3, TimeUnit.SECONDS)
                                .doOnComplete(() -> {
                                    test.add(currentString[0]);
                                    currentString[0] = "";
                                }));
                    })
                    .publish();

    integerObservable
            .observeOn(Schedulers.newThread())
            .subscribe(System.out::println);

    integerObservable
            .observeOn(Schedulers.newThread())
            .map((sa) -> {
                currentString[0] = currentString[0] + sa;
                System.out.println(currentString[0]);
                return sa;
            })
            .subscribe();

In this example we use the onComplete method of the observable we hold the timer to reset our state. This example doesn't take in consideration the option when the consumers are slower to consume dhe data than the delay for the repeat(it will result in a overflow of result from one chain of data to the other). I would suggest using other ways to handle the repeat part for example:

final int[] i = {0};

    ConnectableObservable<Integer> integerObservable =
            Observable.defer(() -> Observable.fromArray(i[0]++, i[0]++, i[0]++)).publish();

    Observable b = integerObservable.observeOn(Schedulers.newThread());

    Observable a = integerObservable
            .observeOn(Schedulers.newThread())
            .collect(StringBuilder::new, (sb, x) -> sb.append(x).append(","))
            .map(StringBuilder::toString)
            .toObservable();

    Observable
            .interval(1, TimeUnit.SECONDS)
            .doOnSubscribe((d) -> {
                a.subscribe(System.out::println);
                b.subscribe(System.out::println);

                integerObservable.connect();
            })
            .doOnNext((d) -> {
                a.subscribe(System.out::println);
                b.subscribe(System.out::println);

                integerObservable.connect();
            })
            .doOnComplete(() -> {})
            .subscribe();

The first connect it's for the first time of the execution and onNext is called every 1 second and it restarts the whole pipeline again from the beginning.

Hope this helps.

Still not clear if this is what you are asking for but maybe this will help you

  Observable<Integer> integerObservable1;
@Override
public void run(String... args) throws Exception {

    Integer[] integers = {1, 2, 3};
    Observable<Integer> integerObservable = Observable.fromArray(integers);

    integerObservable1 = Observable.zip(integerObservable.observeOn(Schedulers.newThread()).delay(100, TimeUnit.MILLISECONDS), integerObservable.observeOn(Schedulers.newThread()).delay(200, TimeUnit.MILLISECONDS), (integer, integer2) -> {
        System.out.println(integer + " " + integer2);
        return integer;
    })
            .doOnComplete(() -> {
                integerObservable1.subscribe();
            });
    integerObservable1.subscribe();

}