如何使用 RX Java 在我的 EventBus 上正确处理 onError
How to properly handle onError on my EventBus using RX Java
我将 rx java 与 android 一起使用,我的事件总线 class 如下所示
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 的错误日志:
我该如何解决这个问题?请给我一些解决这个问题的建议
当您订阅 getStickyObservable()
时,您需要实现 onError
方法(不要只使用例如 .subscribe(action)
重载)。
我将 rx java 与 android 一起使用,我的事件总线 class 如下所示
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 的错误日志:
我该如何解决这个问题?请给我一些解决这个问题的建议
当您订阅 getStickyObservable()
时,您需要实现 onError
方法(不要只使用例如 .subscribe(action)
重载)。