RxJava Ошибка Почтовый оператор Работа с итератора

Question

фон

У меня есть процесс, который использует RxJava, чтобы получить данные из разных мест на основе списка. Каждый предмет получает другой метод (все возвращаемые Observables). Из-за того, что N элементов для получения логического оператора для использования - zip с итератором.

Проблема

Приведенный ниже код работает, как ожидалось, но это, кажется, «неправильно», что мне нужно try-catch блок, чтобы поймать исключение, брошенный getBigFoo() - которая возвращает FooNotFoundException. Не покрывают ли другие операторы, связанные с ошибкой, такие как onErrorResumeNext() и onErrorReturn()?

private Observable<Bar> processFoos(List<Foo> foos) { 

     List<Observable<? extends IBar>> observablesToZip = new ArrayList<>(); 

     for(Foo foo : foos) { 

      switch (foo.getType()) { 

       case BIG_FOO : 

        try { 
         observablesToZip.add(getBigFoo(foo.getId())); 
        } catch (Exception exception) { 
         //do nothing - but this seems wrong 
        } 
      } 
     } 

     return Observable.zip(observablesToZip, results -> mergeFoosIntoBar(results)); 
    } 

Попытки

Попытка ниже, кажется, не поймать исключение генерируется. Я не понимаю, почему, поскольку в последовательности нет технических элементов вверх или вниз, так что Observable.empty() должен работать?

private Observable<Bar> processFoos(List<Foo> foos) { 

     List<Observable<? extends IBar>> observablesToZip = new ArrayList<>(); 

     for(Foo foo : foos) { 

      switch (foo.getType()) { 

       case BIG_FOO : 
        observablesToZip.add(getBigFoo(foo.getId().onErrorResumeNext(Observable.empty())); 
      } 
     } 

     return Observable.zip(observablesToZip, results -> mergeFoosIntoBar(results)); 
    } 

Answer 1

@dwursteisen был на правильный ответ, но не совсем там.

Мой вопрос заключается в том, что я бросал новый FooNotFoundException:

throw new FooNotFoundException() 

Но что мне нужно было сделать:

return Observable.error(new FooNotFoundException()); 

Тогда в моей Zip функции:

observablesToZip.add(getBigFoo(foo.getId())).onExceptionResumeNext(Observable.just(null); 

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

Answer 2

Возможно, вы захотите использовать defer. getBigFoo не должен генерировать исключение, а вместо этого возвращать Observable по ошибке. Так defer может помочь вам исправить:

Observable<IBar> obs = Observable.defer(() -> { 
      try { 
       return getBigFoo(foo.getId()); 
       } catch (Exception ex) { 
        return Observable.error(ex); 
       } 
}); 
observablestoZip.add(obs); 

Answer 3

Не могли бы вы сделать getBigFoo (foo.getId()) бросить RuntimeException вместо Exception ?. Все исключения на конвейере должны быть захвачены, но не время выполнения.

Посмотрите на этот глупый пример

  /** 
* Here is a silly example how runtimeExceptions are not needed 
*/ 
@Test 
public void observableRuntimeException() { 
    Integer[] numbers = {0, 1, 2, 3, 4, 5}; 

    Observable.from(numbers) 
       .doOnNext(number -> throwRuntimeException()) 
       .doOnError(t -> System.out.println("Expecting illegal argument exception:" + t.getMessage())) 
       .subscribe(); 

} 

private void throwRuntimeException() { 
    throw new RuntimeException(); 
} 

вы можете увидеть больше примеров здесь https://github.com/politrons/reactive