Rx: Как получить последний элемент, даже если onError был вызван?

Question

Я использую RxJava, и мне нужно сделать 2 вещи:

• Получить последний элемент, испускаемый из Observable
• Определить, если onError был вызван, по сравнению с onCompleted

Я ve посмотрел на использование last и lastOrDefault (что на самом деле является поведением, которое мне нужно), но я не смог обойти onError, скрывая последний элемент. Я бы ОК с наблюдаемым существом используется дважды, один раз, чтобы получить значение last и один раз, чтобы получить статус завершения, но до сих пор я только был в состоянии сделать это, создав свой собственный Observer:

public class CacheLastObserver<T> implements Observer<T> { 

    private final AtomicReference<T> lastMessageReceived = new AtomicReference<>(); 
    private final AtomicReference<Throwable> error = new AtomicReference<>(); 

    @Override 
    public void onCompleted() { 
     // Do nothing 
    } 

    @Override 
    public void onError(Throwable e) { 
     error.set(e); 
    } 

    @Override 
    public void onNext(T message) { 
     lastMessageReceived.set(message); 
    } 

    public Optional<T> getLastMessageReceived() { 
     return Optional.ofNullable(lastMessageReceived.get()); 
    } 

    public Optional<Throwable> getError() { 
     return Optional.ofNullable(error.get()); 
    } 
} 

У меня нет проблем с созданием моего собственного Observer, но похоже, что Rx должен быть лучше в состоянии удовлетворить этот случай использования «получить последний элемент, выпущенный до завершения». Любые идеи о том, как это сделать?

Answer 1

Попробуйте это:

source.materialize().buffer(2).last() 

В случае ошибки последней эмиссии будет список из двух элементов являются последним значением, излучаемое завернутым как Notification и уведомление об ошибках. Без ошибки второй элемент будет уведомлением о завершении.

Обратите также внимание, что если значение не выбрано, то результатом будет список одного элемента, являющегося уведомлением о терминале.

Answer 2

Вы попробуйте onErrorResumeNext здесь вы можете увидеть остальные или ошибки операторов обработки https://github.com/ReactiveX/RxJava/wiki/Error-Handling-Operators

Answer 3

я использовал этот подход, чтобы решить вашу проблему.

public class ExampleUnitTest { 
    @Test 
    public void testSample() throws Exception { 
     Observable.just(1, 2, 3, 4, 5) 
       .map(number -> { 
        if (number == 4) 
         throw new NullPointerException(); 
        else 
         return number; 
       }) 
       .onErrorResumeNext(t -> Observable.empty()) 
       .lastOrDefault(15) 
       .subscribe(lastEmittedNumber -> System.out.println("onNext: " + lastEmittedNumber)); 
    } 
} 

Он будет излучать onNext: 3

Надеются, что это помогает.

Answer 4

Я решил с:

source.materialize().withPrevious().last() 

где withPrevious есть (Котлин):

fun <T> Observable<T>.withPrevious(): Observable<Pair<T?, T>> = 
    this.scan(Pair<T?, T?>(null, null)) { previous, current -> Pair(previous.second, current) } 
     .skip(1) 
     .map { it as Pair<T?, T> }