Concat

Question

我有一个List of ConnectableObservable，当完成前一项时，我希望从列表中运行一项。我尝试过在列表中应用concat()方法，但是这个方法显然不适用于ConnectableObservables。我该怎么做？

这就是我尝试过的：

ConnectableObservable<Long> observable1 =
    Observable.timer(1500, TimeUnit.MILLISECONDS).publish();

ConnectableObservable<Long> observable2 =
    Observable.timer(1550, TimeUnit.MILLISECONDS).publish();


List<ConnectableObservable<Long>> list = new ArrayList<>();
list.add(observable1);
list.add(observable2);

Observable.concat(list).doOnNext(aLong -> {
    Log.i("result", aLong.toString());
}).subscribe();

observable1.connect();
observable2.connect();

在这里，observable2在observable1完成后运行50毫秒，而不是预期的1550毫秒。

Answer 1

您没有在connect()上调用ConnectableObservable方法，因此它没有开始发布任何内容。

放

observable1.connect();
observable2.connect();

在你的代码末尾。或者，您可以将.autoConnect(1)添加到observable1和observable2中。

它不能同时工作，如下junit所确认的那样：

    @Test
    void connectableTest() {
        TestScheduler testScheduler = new TestScheduler();
        ConnectableObservable<Integer> observable1 =
                Observable
                        .just(1, 2, 3)
                        .zipWith(
                                Observable.interval(0, 1, TimeUnit.SECONDS, testScheduler), (integer, time) -> integer)
                        .publish();
        ConnectableObservable<Integer> observable2 =
                Observable
                        .just(4, 5, 6)
                        .zipWith(
                                Observable.interval(3, 1, TimeUnit.SECONDS, testScheduler), (integer, time) -> integer)
                        .publish();

        List<ConnectableObservable<Integer>> list = new ArrayList<>();
        list.add(observable1);
        list.add(observable2);

        TestObserver<Integer> output = Observable.concat(list).test();

        observable1.connect();
        observable2.connect();

        testScheduler.advanceTimeBy(20, TimeUnit.SECONDS);
        output.assertValues(1,2,3,4,5,6).assertComplete();
    }