Объединение

В RxClojure здесь имеется шесть операторов, которые стоит рассмотреть:

RxCpp реализует этот оператор как merge.

RxGroovy реализует этот оператор как merge, mergeWith, и mergeDelayError.

Версия оператора merge — mergeWith, так что, например, в примере кода выше вместо Observable.merge(odds,evens) можно написать odds.mergeWith(evens).

• Javadoc: mergeWith(Observable)

Если какое-либо из отдельных Observable завершится уведомлением onError, Observable, созданное merge, немедленно завершится уведомлением onError. Если вы предпочитаете объединение, которое продолжает испускать результаты оставшихся, не содержащих ошибок Observable, прежде чем сообщить об ошибке, используйте mergeDelayError вместо этого.

RxJava реализует этот оператор как merge, mergeWith, и mergeDelayError.

Пример кода

Observable<Integer> odds = Observable.just(1, 3, 5).subscribeOn(someScheduler);
Observable<Integer> evens = Observable.just(2, 4, 6);

Observable.merge(odds, evens)
          .subscribe(new Subscriber<Integer>() {
        @Override
        public void onNext(Integer item) {
            System.out.println("Next: " + item);
        }

        @Override
        public void onError(Throwable error) {
            System.err.println("Error: " + error.getMessage());
        }

        @Override
        public void onCompleted() {
            System.out.println("Sequence complete.");
        }
    });

Next: 1
Next: 3
Next: 5
Next: 2
Next: 4
Next: 6
Sequence complete.

Вместо передачи нескольких Observables (до девяти) в merge, вы также можете передать List<> (или другой Iterable) Observables, массив Observables, или даже Observable, испускающий Observables, и merge объединит их выходные данные в выходные данные одного Observable:

Версия оператора merge это mergeWith, поэтому, например, вместо записи Observable.merge(odds,evens) вы также можете написать odds.mergeWith(evens).

Если любой из отдельных Observables, переданных в merge, завершается onError уведомлением, Observable, производимое merge, немедленно завершится onError уведомлением. Если вам требуется объединение, которое продолжает испускать результаты оставшихся без ошибок Observables перед сообщением об ошибке, используйте mergeDelayError вместо этого.

RxKotlin реализует этот оператор как merge, mergeWith, и mergeDelayError.

Вместо передачи нескольких Observables (до девяти) в merge, вы также можете передать List<> (или другой Iterable) Observables, массив Observables, или даже Observable, которое испускает Observables, и merge будет объединять их выходные данные в выходные данные одного Observable:

Вариант оператора merge — mergeWith, поэтому, например, вместо записи Observable.merge(odds,evens) вы также можете написать odds.mergeWith(evens).

Если любое из отдельных Observables, переданных в merge, завершится уведомлением об onError, Observable, созданный merge, немедленно завершится уведомлением об onError. Если вы предпочтёте объединение, которое продолжает испускать результаты оставшихся, не содержащих ошибок Observables, прежде чем сообщать об ошибке, используйте mergeDelayError вместо этого.

Rx.NET реализует этот оператор как Merge.

RxPY реализует этот оператор как merge и merge_all/merge_observable.

Rx.rb реализует этот оператор как merge, merge_concurrent, и merge_all.

merge_concurrent работает с Observable, который испускает Observables, объединяя испускания каждого из этих Observables в собственные испускания. Вы можете необязательно передать ему целочисленный параметр, указывающий, сколько из этих испущенных Observables merge_concurrent должно пытаться подписаться одновременно. Достигнув этого максимального числа подписок, оно воздержится от подписки на другие Observables, испускаемые исходным Observable, до тех пор, пока одно из уже подписанных Observables не выдаст onCompleted уведомление. По умолчанию это 1, что делает его эквивалентным merge_all.

RxScala реализует этот оператор как flatten, flattenDelayError, merge, и mergeDelayError.

RxSwift реализует этот оператор как merge.