Rxjava2 Observable的輔助操做詳解及實例（二）

時間 2020-01-01

標籤 rxjava2 rxjava observable 輔助詳解實例欄目 Java 简体版

原文原文鏈接

目錄html

接續上篇： Rxjava2 Observable的輔助操做詳解及實例（一）java

8. TimeInterval

將一個發射數據的Observable轉換爲發射那些數據發射時間間隔的Observable。react

TimeInterval 操做符攔截原始Observable發射的數據項，替換爲發射表示相鄰發射物時間間隔的對象。git

這個操做符將原始 Observable 轉換爲另外一個 Observable ，後者發射一個標誌替換前者的數據項，這個標誌表示前者的兩個連續發射物之間流逝的時間長度。新的Observable的第一個發射物表示的是在觀察者訂閱原始 Observable 到原始 Observable 發射它的第一項數據之間流逝的時間長度。不存在與原始 Observable 發射最後一項數據和發射 onCompleted 通知之間時長對應的發射物。github

示例代碼：數據結構

/**
     * 1. timeInterval(Scheduler scheduler)
     *  scheduler: 可選參數，指定調度線程
     *  接收原始數據項，發射射表示相鄰發射物時間間隔的對象
     */
    Observable.intervalRange(1, 10, 100, 100, TimeUnit.MILLISECONDS)
            .timeInterval()
         // .timeInterval(Schedulers.newThread())       // 指定工做線程
            .subscribe(new Observer<Timed<Long>>() {
                @Override
                public void onSubscribe(Disposable d) {
                    System.out.println("--> onSubscribe(1)");
                }

                @Override
                public void onNext(Timed<Long> longTimed) {
                    long time = longTimed.time();           // 連續數據間的間隔時間
                    TimeUnit unit = longTimed.unit();       // 連續數據間的時間間隔單位
                    Long value = longTimed.value();         // Observable發送的數據項
                    System.out.println("--> onNext(1): " + longTimed.toString());
                }

                @Override
                public void onError(Throwable e) {
                    System.out.println("--> onError(1): " + e);
                }

                @Override
                public void onComplete() {
                    System.out.println("--> onComplete(1)");
                }
            });

    System.in.read();
    System.out.println("-------------------------------------------------");
    /**
     *  2. timeInterval(TimeUnit unit, Scheduler scheduler)
     *  指定時間間隔單位和指定工做線程，接收原始數據項，發射射表示相鄰發射物時間間隔的對象
     */
    Observable.intervalRange(1, 10, 1000, 1200, TimeUnit.MILLISECONDS)
        //  .timeInterval(TimeUnit.SECONDS)                             // 指定時間間隔單位
            .timeInterval(TimeUnit.SECONDS, Schedulers.newThread())     // 指定時間間隔單位和指定工做線程
            .subscribe(new Observer<Timed<Long>>() {
                @Override
                public void onSubscribe(Disposable d) {
                    System.out.println("--> onSubscribe(2)");
                }

                @Override
                public void onNext(Timed<Long> longTimed) {
                    System.out.println("--> onNext(2): " + longTimed.toString());
                }

                @Override
                public void onError(Throwable e) {
                    System.out.println("--> onError(2): " + e);
                }

                @Override
                public void onComplete() {
                    System.out.println("--> onComplete(2)");
                }
            });

    System.in.read();

輸出：app

--> onSubscribe(1)
--> onNext(1): Timed[time=104, unit=MILLISECONDS, value=1]
--> onNext(1): Timed[time=100, unit=MILLISECONDS, value=2]
--> onNext(1): Timed[time=100, unit=MILLISECONDS, value=3]
--> onNext(1): Timed[time=100, unit=MILLISECONDS, value=4]
--> onNext(1): Timed[time=100, unit=MILLISECONDS, value=5]
--> onNext(1): Timed[time=100, unit=MILLISECONDS, value=6]
--> onNext(1): Timed[time=100, unit=MILLISECONDS, value=7]
--> onNext(1): Timed[time=100, unit=MILLISECONDS, value=8]
--> onNext(1): Timed[time=100, unit=MILLISECONDS, value=9]
--> onNext(1): Timed[time=100, unit=MILLISECONDS, value=10]
--> onComplete(1)
-------------------------------------------------
--> onSubscribe(2)
--> onNext(2): Timed[time=1, unit=SECONDS, value=1]
--> onNext(2): Timed[time=1, unit=SECONDS, value=2]
--> onNext(2): Timed[time=1, unit=SECONDS, value=3]
--> onNext(2): Timed[time=1, unit=SECONDS, value=4]
--> onNext(2): Timed[time=1, unit=SECONDS, value=5]
--> onNext(2): Timed[time=2, unit=SECONDS, value=6]
--> onNext(2): Timed[time=1, unit=SECONDS, value=7]
--> onNext(2): Timed[time=1, unit=SECONDS, value=8]
--> onNext(2): Timed[time=1, unit=SECONDS, value=9]
--> onNext(2): Timed[time=1, unit=SECONDS, value=10]
--> onComplete(2)

Javadoc: timeInterval()
Javadoc: timeInterval(Scheduler scheduler)
Javadoc: timeInterval(TimeUnit unit)
Javadoc: timeInterval(TimeUnit unit, Scheduler scheduler)ide

9. Timeout

對原始Observable的一個鏡像，若是過了一個指定的時長仍沒有發射數據，它會發一個錯誤通知。函數

RxJava中的實現爲 timeout 操做符，具備多個不一樣的變體。this

9.1 timeout(timeout, timeUnit)

若是原始 Observable 過了指定的一段時長沒有發射任何數據，Timeout操做符會以一個 onError 通知終止這個Observable。

示例代碼：

/**
     *  1. timeout(long timeout, TimeUnit timeUnit)
     *  接受一個時長參數，若是在指定的時間段內沒有數據項發射，將會發射一個Error通知，
     *  或者每當原始Observable發射了一項數據， timeout 就啓動一個計時器，
     *  若是計時器超過了指定指定的時長而原始Observable沒有發射另外一項數據， 
     *  就拋出 TimeoutException ，以一個錯誤通知終止Observable。
     */
    Observable.create(new ObservableOnSubscribe<Long>() {
        @Override
        public void subscribe(ObservableEmitter<Long> emitter) throws Exception {
            //  Thread.sleep(2000);     // 延遲2秒後發射數據，此時會有TimeoutException
            emitter.onNext(1L);
            Thread.sleep(2000);     // 延遲2秒後發射數據，此時會有TimeoutException
            emitter.onNext(2L);
            emitter.onComplete();
        }
    }).timeout(1, TimeUnit.SECONDS)     // 指定超時時間段爲1秒
      .subscribe(new Observer<Long>() {
            @Override
            public void onSubscribe(Disposable d) {
                System.out.println("--> onSubscribe(1)");
            }

            @Override
            public void onNext(Long aLong) {
                System.out.println("--> onNext(1): " + aLong);
            }

            @Override
            public void onError(Throwable e) {
                System.out.println("--> onError(1): " + e);
            }

            @Override
            public void onComplete() {
                System.out.println("--> onComplete(1)");
            }
      });

    System.in.read();

輸出：

--> onSubscribe(1)
--> onNext(1): 1
--> onError(1): java.util.concurrent.TimeoutException: The source did not signal an event for 1 seconds and has been terminated.

Javadoc: timeout(long timeout, TimeUnit timeUnit)

9.2 timeout(timeout, timeUnit, scheduler, other)

在指定時間段後超時時會切換到使用一個你指定的備用的 Observable，而不是發onError通知，能夠經過scheduler 來指定工做線程。

示例代碼：

/**
     *  2. timeout(long timeout, TimeUnit timeUnit,
     *  Scheduler scheduler,        // 可選參數，指定線程調度器
     *  ObservableSource other      // 可選參數，超時備用Observable
     *  )
     *
     *  在指定時間段後超時時會切換到使用一個你指定的備用的Observable，而不是發onError通知。
     */
    Observable.create(new ObservableOnSubscribe<Long>() {
        @Override
        public void subscribe(ObservableEmitter<Long> emitter) throws Exception {
            //  Thread.sleep(2000);     // 延遲2秒後發射數據，此時會有TimeoutException
            emitter.onNext(1L);
            Thread.sleep(2000);         // 延遲2秒後發射數據，此時會有TimeoutException
            emitter.onNext(2L);
            emitter.onComplete();
        }
    }).timeout(1, TimeUnit.SECONDS,             // 指定超時時間段爲1秒
            Schedulers.newThread(),             // 指定工做線程爲子線程
            Observable.just(888L))              // 超時後默認發射的Observable
            .subscribe(new Observer<Long>() {
                @Override
                public void onSubscribe(Disposable d) {
                    System.out.println("--> onSubscribe(2)");
                }

                @Override
                public void onNext(Long aLong) {
                    System.out.println("--> onNext(2): " + aLong);
                }

                @Override
                public void onError(Throwable e) {
                    System.out.println("--> onError(2): " + e);
                }

                @Override
                public void onComplete() {
                    System.out.println("--> onComplete(2)");
                }
            });
    
    System.in.read();

輸出：

--> onSubscribe(2)
--> onNext(2): 1
--> onNext(2): 888
--> onComplete(2)

Javadoc: timeout(long timeout, TimeUnit timeUnit, Scheduler scheduler, ObservableSource other)

9.3 timeout(Function itemTimeoutIndicator, ObservableSource other)

使用一個函數 itemTimeoutIndicator 針對原始 Observable 的每一項返回一個 Observable，若是當這個 Observable 終止時原始 Observable 尚未發射另外一項數據，就會認爲是超時了，若是沒有指定超時備用的 other，就拋出 TimeoutException，以一個錯誤通知終止 bservable，不然超時後發射備用的 Observable。

示例代碼：

/**
     *  3. timeout(Function<T, ObservableSource> itemTimeoutIndicator
     *  ObservableSource other      // 可選參數，當超時後發射的備用Observable
     *  )
     *  對原始Observable的每一項返回一個Observable，
     *  若是當這個Observable終止時原始Observable尚未發射另外一項數據，就會認爲是超時了，
     *  若是沒有指定超時備用的Observable，就拋出TimeoutException，以一個錯誤通知終止Observable，
     *  不然超時後發射備用的Observable。
     */
    Observable.create(new ObservableOnSubscribe<Long>() {
        @Override
        public void subscribe(ObservableEmitter<Long> emitter) throws Exception {
            emitter.onNext(1L);
            Thread.sleep(3000);     // 延遲3秒後發射數據，此時會有TimeoutException
            emitter.onNext(2L);
            emitter.onComplete();
        }
    }).timeout(new Function<Long, ObservableSource<Long>>() {
        @Override
        public ObservableSource<Long> apply(Long aLong) throws Exception {
            // 爲每個原始數據發射一個Observable來指示下一個數據發射的Timeout，這裏指定1秒超時時間
            return Observable.timer(1, TimeUnit.SECONDS);
        }
    }, Observable.just(888L))  // 超時後默認發射的Observable
            .subscribe(new Observer<Long>() {
                @Override
                public void onSubscribe(Disposable d) {
                    System.out.println("--> onSubscribe(3)");
                }

                @Override
                public void onNext(Long aLong) {
                    System.out.println("--> onNext(3): " + aLong);
                }

                @Override
                public void onError(Throwable e) {
                    System.out.println("--> onError(3): " + e);
                }

                @Override
                public void onComplete() {
                    System.out.println("--> onComplete(3)");
                }
            });

    System.in.read();

輸出：

--> onSubscribe(3)
--> onNext(3): 1
--> onNext(3): 888
--> onComplete(3)

Javadoc: timeout(Function<T, ObservableSource> itemTimeoutIndicator)
Javadoc: timeout(Function<T, ObservableSource> itemTimeoutIndicator, ObservableSource other)

10. Timestamp

給Observable發射的數據項附加一個指定的時間戳。

timestamp ，它將一個發射Timed類型數據的Observable轉換爲一個發射類型爲 Timestamped<Timed> 的數據的Observable，每一項都包含數據的原始發射時間信息和原始數據。

示例代碼：

/**
     *  1. timestamp(Scheduler scheduler)
     *  scheduler: 可選參數，指定線程調度器
     *
     *  給Observable發射的數據項附加一個時間戳信息
     */
    Observable.intervalRange(1, 5, 1, 100, TimeUnit.MILLISECONDS)
            .timestamp(Schedulers.newThread())      // 指定在子線程調度處理
            .subscribe(new Observer<Timed<Long>>() {
                @Override
                public void onSubscribe(Disposable d) {
                    System.out.println("--> onSubscribe(1)");
                }

                @Override
                public void onNext(Timed<Long> longTimed) {
                    long time = longTimed.time();           // 連續數據間的間隔時間
                    TimeUnit unit = longTimed.unit();       // 連續數據間的時間間隔單位
                    Long value = longTimed.value();         // Observable發送的數據項
                    System.out.println("--> onNext(1): " + longTimed);
                }

                @Override
                public void onError(Throwable e) {
                    System.out.println("--> onError(1): " + e);
                }

                @Override
                public void onComplete() {
                    System.out.println("--> onComplete(1)");
                }
            });

    System.in.read();
    System.out.println("-------------------------------------------");
    /**
     *  2. timestamp(TimeUnit unit, Scheduler scheduler)
     *  scheduler: 可選參數，指定線程調度器
     *
     *  給Observable發射的數據項附加一個指定單位的時間戳信息
     */
    Observable.intervalRange(1, 5, 1, 1200, TimeUnit.MILLISECONDS)
            .timestamp(TimeUnit.SECONDS, Schedulers.newThread())    // 指定時間單位爲秒，在子線程調度處理
            .subscribe(new Observer<Timed<Long>>() {
                @Override
                public void onSubscribe(Disposable d) {
                    System.out.println("--> onSubscribe(2)");
                }

                @Override
                public void onNext(Timed<Long> longTimed) {
                    System.out.println("--> onNext(2): " + longTimed);
                }

                @Override
                public void onError(Throwable e) {
                    System.out.println("--> onError(2): " + e);
                }

                @Override
                public void onComplete() {
                    System.out.println("--> onComplete(2)");
                }
            });

    System.in.read();

輸出：

--> onSubscribe(1)
--> onNext(1): Timed[time=1577455367446, unit=MILLISECONDS, value=1]
--> onNext(1): Timed[time=1577455367545, unit=MILLISECONDS, value=2]
--> onNext(1): Timed[time=1577455367645, unit=MILLISECONDS, value=3]
--> onNext(1): Timed[time=1577455367745, unit=MILLISECONDS, value=4]
--> onNext(1): Timed[time=1577455367845, unit=MILLISECONDS, value=5]
--> onComplete(1)
-------------------------------------------
--> onSubscribe(2)
--> onNext(2): Timed[time=1577455369, unit=SECONDS, value=1]
--> onNext(2): Timed[time=1577455370, unit=SECONDS, value=2]
--> onNext(2): Timed[time=1577455371, unit=SECONDS, value=3]
--> onNext(2): Timed[time=1577455373, unit=SECONDS, value=4]
--> onNext(2): Timed[time=1577455374, unit=SECONDS, value=5]
--> onComplete(2)

Javadoc: timestamp()
Javadoc: timestamp(Scheduler scheduler)
Javadoc: timestamp(TimeUnit unit)
Javadoc: timestamp(TimeUnit unit, Scheduler scheduler)

11. Using

建立一個只在Observable生命週期內存在的一次性資源。

Using 操做符讓你能夠指示Observable建立一個只在它的生命週期內存在的資源，當Observable終止時這個資源會被自動釋放。

using 操做符接受三個參數：

observableFactory：一個用戶建立一次性資源的工廠函數
resourceFactory：一個用於建立Observable的工廠函數
disposeFunction：一個用於釋放資源的函數

當一個觀察者訂閱 using 返回的Observable時， using 將會使用Observable工廠函數建立觀察者要觀察的Observable，同時使用資源工廠函數建立一個你想要建立的資源。當觀察者取消訂閱這個Observable時，或者當觀察者終止時（不管是正常終止仍是因錯誤而終止）， using 使用第三個函數釋放它建立的資源。

示例代碼：

/**
     * 用於在Observable的生命週期內存在的資源對象
     */
    class MyResource {
        private String resource;

        public MyResource(String resource) {
            this.resource = resource;
        }

        @Override
        public String toString() {
            return "MyResource{" +
                    "resource='" + resource + '\'' +
                    '}';
        }

        public void releaseResource() {
            System.out.println("----> MyResource resource is release. ");
            resource = null;
        }
    }
    
    /**
     *  1. using(Callable resourceSupplier, Function sourceSupplier, Consumer disposer, boolean eager)
     *
     *  resourceSupplier:   // 一個用戶建立一次性資源的工廠函數
     *  sourceSupplier:     // 一個用於建立Observable的工廠函數
     *  disposer:           // 一個用於釋放資源的函數
     *  eager:              // 可選參數，若是爲true的話，則第三個函數disposer的處理在Observable的結束前執行
     *
     *  當一個觀察者訂閱 using 返回的Observable時， using 將會使用Observable工廠函數建立觀察者要觀察的Observable，
     *  同時使用資源工廠函數建立一個你想要建立的資源。
     *  當觀察者取消訂閱這個Observable時，或者當觀察者終止時（不管是正常終止仍是因錯誤而終止）， 
     *  using 使用第三個函數釋放它建立的資源。
     */
    Observable.using(
            // 一個用戶建立一次性資源的工廠函數
            new Callable<MyResource>() {
                @Override
                public MyResource call() throws Exception {
                    System.out.println("----> resourceSupplier call");
                    return new MyResource("This is Observable resource!");
                }
            },
            // 一個用於建立Observable的工廠函數，這個函數返回的Observable就是最終被觀察的Observable
            new Function<MyResource, ObservableSource<Long>>() {
                @Override
                public ObservableSource<Long> apply(MyResource myResource) throws Exception {
                    System.out.println("----> sourceSupplier apply: " + myResource);
                    return Observable.rangeLong(1, 5);
                }
            },
            // 一個用於釋放資源的函數
            new Consumer<MyResource>() {
                @Override
                public void accept(MyResource myResource) throws Exception {
                    System.out.println("----> disposer accept: ");
                    myResource.releaseResource();
                }
            },
            // 可選參數，若是爲true的話，則在Observable的結束前執行釋放資源的函數
            true).subscribe(new Observer<Long>() {
                @Override
                public void onSubscribe(Disposable d) {
                    System.out.println("--> onSubscribe");
                }

                @Override
                public void onNext(Long aLong) {
                    System.out.println("--> onNext: " + aLong);
                }

                @Override
                public void onError(Throwable e) {
                    System.out.println("--> onError: " + e);
                }

                @Override
                public void onComplete() {
                    System.out.println("--> onComplete");
                }
            });

輸出：

----> resourceSupplier call(1)
----> sourceSupplier apply(1): MyResource{resource='This is Observable resource!'}
--> onSubscribe(1)
--> onNext(1): 1
--> onNext(1): 2
--> onNext(1): 3
--> onNext(1): 4
--> onNext(1): 5
----> disposer accept(1): 
----> MyResource resource is release. 
--> onComplete

Javadoc: using(Callable resourceSupplier, Function sourceSupplier, Consumer disposer)
Javadoc: using(Callable resourceSupplier, Function sourceSupplier, Consumer disposer, boolean eager)

12. To

將Observable轉換爲另外一個對象或數據結構。

將 Observable 或者Observable 發射的數據序列轉換爲另外一個對象或數據結構。它們中的一些會阻塞直到 Observable 終止，而後生成一個等價的對象或數據結構；另外一些返回一個發射那個對象或數據結構的 Observable。

因爲 rxjava 的 To 操做符中有不少 toXXX 操做符的實現和不一樣的變體重載，此處就不詳細的展開了，有興趣的能夠查看官方的API 文檔詳細參閱。

下面幾個是常見的幾種To操做符的：

toList()：讓Observable將多項數據組合成一個List，而後調用一次onNext方法傳遞整個列表。
toMap(Function keySelector,Function valueSelector)：toMap收集原始Observable發射的全部數據項到一個Map（默認是HashMap）而後發射這個Map。你能夠提供一個用於生成Map的Key的函數，還能夠提供一個函數轉換數據項到Map存儲的值（默認數據項自己就是值）。
toSortedList()：它會對產生的列表排序，默認是天然升序，若是發射的數據項沒有實現Comparable接口，會拋出一個異常，你也能夠傳遞一個函數做爲用於比較兩個數據項。
toMultimap(Function keySelector, Function valueSelector)：相似於toMap，不一樣的是，它生成的這個Map的value類型仍是一個ArrayList。

示例代碼：

/**
         *  1. toList()
         *  讓Observable將多項數據組合成一個List，而後調用一次onNext方法傳遞整個列表。
         */
        range.toList()
                .subscribe(new Consumer<List<Integer>>() {
                    @Override
                    public void accept(List<Integer> integers) throws Exception {
                        System.out.println("--> toList accept(1): " + integers);
                    }
                });

        System.out.println("------------------------------------------");
        /**
         *  2. toMap(Function<? super T, ? extends K> keySelector,Function<? super T, ? extends V> valueSelector)
         *   toMap收集原始Observable發射的全部數據項到一個Map（默認是HashMap）而後發射這個Map。
         *   你能夠提供一個用於生成Map的Key的函數，還能夠提供一個函數轉換數據項到Map存儲的值（默認數據項自己就是值）。
         */
        range.toMap(new Function<Integer, String>() {
            @Override
            public String apply(Integer integer) throws Exception {
                return "key" + integer;     // 返回一個Map的key
            }
        }, new Function<Integer, Integer>() {
            @Override
            public Integer apply(Integer integer) throws Exception {
                return integer;                     // 返回一個Map的value
            }
        }).subscribe(new SingleObserver<Map<String, Integer>>() {
            @Override
            public void onSubscribe(Disposable d) {
                System.out.println("--> onSubscribe(2)");
            }

            @Override
            public void onSuccess(Map<String, Integer> stringIntegerMap) {
                System.out.println("--> onSuccess(2): " + stringIntegerMap);
            }

            @Override
            public void onError(Throwable e) {
                System.out.println("--> onError(2): " + e);
            }
        });

        System.out.println("------------------------------------------");
        /**
         *  3. toSortedList()
         *  它會對產生的列表排序，默認是天然升序，若是發射的數據項沒有實現Comparable接口，會拋出一個異常。
         *  然而，你也能夠傳遞一個函數做爲用於比較兩個數據項
         */
        Observable.just(5, 3, 8, 6, 9, 10)
                .toSortedList()
                .subscribe(new SingleObserver<List<Integer>>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        System.out.println("--> onSubscribe(3)");
                    }

                    @Override
                    public void onSuccess(List<Integer> integers) {
                        System.out.println("--> onSuccess(3): " + integers);
                    }

                    @Override
                    public void onError(Throwable e) {
                        System.out.println("--> onError(3): " + e);
                    }
                });

        System.out.println("------------------------------------------");
        /**
         *  4. toSortedList(Comparator comparator)
         *
         *  傳遞一個函數comparator做爲用於比較兩個數據項，它會對產生的列表排序
         */
        Observable.just(5, 3, 8, 6, 9, 10)
                .toSortedList(new Comparator<Integer>() {
                    @Override
                    public int compare(Integer o1, Integer o2) {
                        System.out.println("--> compare: o1 = " + o1 + ", o2 = " + o2);
                        return o1 - o2;     // 比較器的排序邏輯
                    }
                }).subscribe(new SingleObserver<List<Integer>>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        System.out.println("--> onSubscribe(4)");
                    }

                    @Override
                    public void onSuccess(List<Integer> integers) {
                        System.out.println("--> onSuccess(4): " + integers);
                    }

                    @Override
                    public void onError(Throwable e) {
                        System.out.println("--> onError(4): " + e);
                    }
                });

        System.out.println("------------------------------------------");
        /**
         *  5. toMultimap(Function<T, K> keySelector, Function<T, V> valueSelector)
         *  相似於 toMap ，不一樣的是，它生成的這個Map的value類型仍是一個ArrayList
         */
        range.toMultimap(new Function<Integer, String>() {
            @Override
            public String apply(Integer integer) throws Exception {
                return "key" + integer;     // 返回一個Map的key
            }
        }, new Function<Integer, Integer>() {
            @Override
            public Integer apply(Integer integer) throws Exception {
                return integer;                  // 返回一個Map的value
            }
        }).subscribe(new SingleObserver<Map<String, Collection<Integer>>>() {
            @Override
            public void onSubscribe(Disposable d) {
                System.out.println("--> onSubscribe(5)");
            }

            @Override
            public void onSuccess(Map<String, Collection<Integer>> stringCollectionMap) {
                System.out.println("--> onSuccess(5): " + stringCollectionMap);
            }

            @Override
            public void onError(Throwable e) {
                System.out.println("--> onError(5): " + e);
            }
        });

輸出：

--> toList accept(1): [1, 2, 3, 4, 5]
------------------------------------------
--> onSubscribe(2)
--> onSuccess(2): {key1=1, key2=2, key5=5, key3=3, key4=4}
------------------------------------------
--> onSubscribe(3)
--> onSuccess(3): [3, 5, 6, 8, 9, 10]
------------------------------------------
--> onSubscribe(4)
--> compare: o1 = 3, o2 = 5
--> compare: o1 = 8, o2 = 3
--> compare: o1 = 8, o2 = 5
--> compare: o1 = 6, o2 = 5
--> compare: o1 = 6, o2 = 8
--> compare: o1 = 9, o2 = 6
--> compare: o1 = 9, o2 = 8
--> compare: o1 = 10, o2 = 6
--> compare: o1 = 10, o2 = 9
--> onSuccess(4): [3, 5, 6, 8, 9, 10]
------------------------------------------
--> onSubscribe(5)
--> onSuccess(5): {key1=[1], key2=[2], key5=[5], key3=[3], key4=[4]}

Javadoc: toList()
Javadoc: toMap(Function keySelector,Function valueSelector)
Javadoc: toSortedList()
Javadoc: toMultimap(Function keySelector, Function valueSelector)