How to properly handle onError on my EventBus using RX Java

Question

I am using rx java with android and my event bus class is as below

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);
}

}

I am getting a lot of error logs with rx.exceptions.OnErrorNotImplementedException:

How can i fix this ? Please suggest me some solution to this problem

Answer 1

When you subscribe to getStickyObservable() you need to implement the onError method (don't just use for example the .subscribe(action) overload).