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