Как правильно обрабатывать OnError на моем EventBus с помощью RX Java

Question

Я использую рецептурный Java с андроид и мои событиями класс шины, как показано ниже

public class EventBus { 
private final Subject<Event, Event> subject = new SerializedSubject<>(PublishSubject.<Event>create()); 
private Observable<Map<Type, Event>> stickyObservable; 

public EventBus() { 
    createStickyObservable(); 
} 

private void createStickyObservable() { 
    final List<Observable<Event>> observables = new ArrayList<>(); 

    final Observable<Map<Type, Event>> so = subject 
      .filter(event -> event.sticky) 
      .groupBy(event -> event.type) 
      .switchMap(groupedObservable -> { 
       BehaviorSubject<Event> bs = BehaviorSubject.create(); 
       groupedObservable.subscribe(bs); 
       observables.add(bs); 
       return Observable.combineLatest(observables, args -> { 
        Map<Type, Event> map = new HashMap<>(); 
        for (Object arg : args) { 
         Event event = (Event) arg; 
         map.put(event.type, event); 
        } 

        return map; 
       }); 
      }); 

    final BehaviorSubject<Map<Type, Event>> bs = BehaviorSubject.create(); 
    so.subscribe(bs); 
    stickyObservable = bs; 
} 

public Observable<Event> filter(final String pathExpression) { 
    final Pattern pattern = Pattern.compile(pathExpression); 

    return subject.filter(event -> { 
     if (event.path == null) { 
      return pathExpression == null; 
     } 
     return pattern.matcher(event.path).matches(); 
    }); 
} 

public Observable<Map<Type, Event>> getStickyObservable() { 
    return stickyObservable; 
} 


public void event(Event event) { 
    subject.onNext(event); 
} 

}

Я получаю много журналов ошибок с rx.exceptions.OnErrorNotImplementedException:

Как я могу это исправить? Пожалуйста, предложите мне некоторое решение этой проблемы

Answer 1

Когда вы подписываетесь на getStickyObservable(), вам необходимо реализовать метод onError (не просто используйте, например, перегрузку .subscribe(action)).