RxJava：特定类型的 BehaviorSubject

Question

我正在尝试实现某种 RxBus，它允许发布特定类型的事件并根据对象类监听它们。我正在使用 BehaviorSubject 来支持粘性事件（甚至可以在订阅之前发布）。并且应该为每种类型保留粘性事件。代码如下：

private final Subject<BaseEvent, BaseEvent> bus = new SerializedSubject<>(BehaviorSubject.create());

public <E extends BaseEvent> void post(E event) {
    bus.onNext(event);
}

public <E extends BaseEvent> Observable<E> observe(Class<E> eventClass) {
    return bus.asObservable().ofType(eventClass);
}

它适用于一种类型的事件。但是当有更多不同的事件，并且最后发布的事件与我订阅的事件类型不同时，它会被 ofType() 过滤，因为 BehaviorSubject 只保留最后一个事件，不依赖于类型。

我在想两个解决方案：

1. 为每种类型创建主题图，但子类型存在问题。
2. 使用 ReplaySubject 并使用 ofType() 和 distinct() 过滤事件。但我找不到区分订阅前后发出的事件的方法。

您认为，有没有办法让这些解决方案发挥作用，或者我可能遗漏了一些东西并且有更好的方法来实现它？

Answer 1

首先，您不需要主题而是Relay。

我向您建议第三种解决方案：编写您自己的中继（或主题）实现，该实现将记住所有不同的事件。您无需从头开始重写 Relay 或 Subject，您可以依赖现有的实现（此处为 PublishRelay）：

import com.jakewharton.rxrelay2.PublishRelay;
import com.jakewharton.rxrelay2.Relay;

import java.util.HashMap;
import java.util.Map;

import io.reactivex.Observer;

public class RxBus extends Relay<Object> {

    private PublishRelay<Object> concreteRelay = PublishRelay.create();
    private Map<Class, Object> stickyEvents = new HashMap<>();

     public <T> T getSticky(Class<T> type) {
         return (T)stickyEvents.get(type);
     }

    @Override
    public void accept(Object value) {
        stickyEvents.put(value.getClass(), value);
        concreteRelay.accept(value);
    }

    @Override
    public boolean hasObservers() {
        return concreteRelay.hasObservers();
    }

    @Override
    protected void subscribeActual(Observer<? super Object> observer) {
        concreteRelay.subscribeActual(observer);
    }
}