rx-java
операторы

замечания

Операторы, введение

Оператор может использоваться для управления потоком объектов от Observable to Subscriber .

Observable<Integer> integerObservable = Observable.just(1, 2, 3); // creating a simple Integer observable
Subscriber<String> mSubscriber = new Subscriber<String>() {
    @Override
    public void onCompleted() {
        System.out.println("onCompleted called!");
    }

    @Override
    public void onError(Throwable throwable) {
        System.out.println("onError called");
    }

    @Override
    public void onNext(String string) {
        System.out.println("onNext called with: " + string);
    }
}; // a simple String subscriber

integerObservable
    .map(new Func1<Integer, String>() {
        @Override
        public String call(Integer integer) {
            switch (integer) {
                case 1:
                    return "one";
                case 2:
                    return "two";
                case 3:
                    return "three";
                default:
                    return "zero";
            }
        }
}).subscribe(mSubscriber);

Результатом будет:

onNext called with: one
onNext called with: two
onNext called with: three
onCompleted called!

Оператор map изменил наблюдаемое Integer наблюдаемый String , тем самым манипулируя потоком объектов.

Цепочка операторов

Несколько операторов могут быть chained вместе для более мощных преобразований и манипуляций.

integerObservable // emits 1, 2, 3
            .map(i -> i + 10) // adds 10 to each item; emits 11, 12, 13
            .filter(i -> i > 11) // emits items that satisfy condition; 12, 13
            .last() // emits last item in observable; 13
            // unlimited operators can be added ...
            .subscribe(System.out::println); // prints 13

Любое число операторов может быть добавлено между Observable и Subscriber .

Оператор FlatMap

Оператор flatMap помогает преобразовать одно событие в другое Observable (или преобразовать событие в ноль, один или несколько событий).

Это идеальный оператор, если вы хотите вызвать другой метод, который возвращает Observable

 public Observable<String> perform(int i) {
      // ...
 }

 Observable.just(1, 2, 3)
           .flatMap(i -> perform(i))
           .subscribe(result -> System.out.println("result ->" + result);

flatMap будет сериализовать perform подписки, но события, выдаваемые perform не могут быть заказаны. Таким образом , вы можете получить события , испускаемые последним выполнение вызова , прежде чем события от первого perform вызова (вы должны использовать concatMap вместо этого).

Если вы создаете еще один Observable в своем подписчике, вместо этого вы должны использовать flatMap . Основная идея: никогда не покидайте Наблюдаемый

Например :

 Observable.just(1, 2, 3)
           .subscribe(i -> perform(i));

можно легко заменить на:

 Observable.just(1, 2, 3)
           .flatMap(i -> perform(i))
           .subscribe();

Документация Reactivex.io: http://reactivex.io/documentation/operators/flatmap.html

фильтр-оператор

Вы можете использовать оператор filter для фильтрации элементов из потока значений на основе результата предикатного метода.

Другими словами, элементы, передаваемые от наблюдателя к подписчику, будут отброшены на основе filter , который вы передадите, если функция возвращает false для определенного значения, это значение будет отфильтровано.

Пример:

List<Integer> integers = Arrays.asList(0, 1, 2, 3, 4, 5, 6, 7, 8, 9);

Observable.from(integers)
    .filter(number -> {
        return (number  % 2 == 0); 
        // odd numbers will return false, that will cause them to be filtered 
    })
    .map(i -> {
        return Math.pow(i, 2); // take each number and multiply by power of 2
    }) 
    .subscribe(onNext -> {
         System.out.println(onNext); // print out the remaining numbers
    });

Этот код распечатает

0.0
4.0
16.0
36.0
64.0

Оператор карты

Вы можете использовать оператор map для сопоставления значений потока с разными значениями в зависимости от результата для каждого значения от функции, переданной на map . Исходный поток представляет собой новую копию и не будет изменять предоставленный поток значений, поток результатов будет иметь одинаковую длину входного потока, но может быть разных типов.

Функция, переданная в .map() , должна вернуть значение.

Пример:

List<Integer> numbers = Arrays.asList(1, 2, 3);
Observable.from(numbers)
    .map(number -> {
        return number.toString(); // convert each integer into a string and return it
    }) 
    .subscribe(onNext -> {
         System.out.println(onNext); // print out the strings
    });

Этот код распечатает

"1"
"2"
"3"

В этом примере Observable принял List<Integer> список будет преобразован в List<String> в конвейере, а .subscribe будет .subscribe String 's

Оператор doOnNext

Оператор doOnNext каждый раз, когда источник Observable испускает элемент. Он может использоваться для целей отладки, применяя некоторые действия к испускаемому элементу, протоколированию и т. Д. ...

Observable.range(1, 3)
    .doOnNext(value -> System.out.println("before transform: " + value))
    .map(value -> value * 2)
    .doOnNext(value -> System.out.println("after transform: " + value))
    .subscribe();

В приведенном ниже примере doOnNext никогда не вызывается, потому что источник Observable испускает ничего, потому что Observable.empty() вызывает onCompleted после подписки.

Observable.empty()
    .doOnNext(item -> System.out.println("item: " + item))
    .subscribe();

повторить оператор

оператор repeat позволяет повторить целую последовательность из источника Observable .

Observable.just(1, 2, 3)
    .repeat()
    .subscribe(
        next -> System.out.println("next: " + next),
        error -> System.out.println("error: " + error),
        () -> System.out.println("complete")
    );

Вывод примера выше

next: 1
next: 2
next: 3
next: 1
next: 2
next: 3

Эта последовательность повторяется бесконечно много раз и никогда не завершается.

Для повторения последовательности конечное число раз просто передайте целое число в качестве аргумента для repeat оператора.

Observable.just(1, 2, 3)
    // Repeat three times and complete
    .repeat(3)
    .subscribe(
        next -> System.out.println("next: " + next),
        error -> System.out.println("error: " + error),
        () -> System.out.println("complete")
    );

Этот пример печатает

next: 1
next: 2
next: 3
next: 1
next: 2
next: 3
next: 1
next: 2
next: 3
complete

Это очень важно понимать , что repeat оператора resubscribes к источнику Observable когда источник Observable последовательность завершается. Давайте перепишем пример выше, используя Observable.create .

Observable.<Integer>create(subscriber -> {

    //Same as Observable.just(1, 2, 3) but with output message
    System.out.println("Subscribed");
    subscriber.onNext(1);
    subscriber.onNext(2);
    subscriber.onNext(3);
    subscriber.onCompleted();
})
        .repeat(3)
        .subscribe(
                next -> System.out.println("next: " + next),
                error -> System.out.println("error: " + error),
                () -> System.out.println("complete")
        );

Этот пример печатает

Subscribed
next: 1
next: 2
next: 3
Subscribed
next: 1
next: 2
next: 3
Subscribed
next: 1
next: 2
next: 3
complete

При использовании оператора сцепления, важно знать , что repeat оператор повторяет всю последовательность , а не предшествующий оператор.

Observable.<Integer>create(subscriber -> {
    System.out.println("Subscribed");
    subscriber.onNext(1);
    subscriber.onNext(2);
    subscriber.onNext(3);
    subscriber.onCompleted();
})
        .map(value -> value * 2) //First chain operator
        .map(value -> "modified " + value) //Second chain operator
        .repeat(3)
        .subscribe(
                next -> System.out.println("next: " + next),
                error -> System.out.println("error: " + error),
                () -> System.out.println("complete")
        );

Этот пример печатает

Subscribed
next: modified 2
next: modified 4
next: modified 6
Subscribed
next: modified 2
next: modified 4
next: modified 6
Subscribed
next: modified 2
next: modified 4
next: modified 6
complete

Этот пример показывает , что repeat оператор повторяет всю последовательность в Повторная подписка Observable , а не повторяющуюся последнюю map оператора , и это не имеет значения , в каком месте в последовательности repeat оператора используется.

Эта последовательность

Observable.<Integer>create(subscriber -> {
        //...
    })
    .map(value -> value * 2) //First chain operator
    .map(value -> "modified " + value) //Second chain operator
    .repeat(3)
    .subscribe(
        /*....*/
    );

равна этой последовательности

Observable.<Integer>create(subscriber -> {
        //...
    })
    .repeat(3)
    .map(value -> value * 2) //First chain operator
    .map(value -> "modified " + value) //Second chain operator
    .subscribe(
        /*....*/
    );