RxJava：特定类型的 BehaviorSubject

Question

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

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

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

我想到了两个解决方案：

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

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

Answer 1

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

我建议您使用第三种解决方案：编写您自己的 Relay（或 Subject）实现来记住所有不同的事件。您不需要从头开始重写 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);
    }
}

RxJava：特定类型的 BehaviorSubject

RxJava: BehaviorSubject of a particular type

rx-java

rx-android