リプレイ別個の要素を持つサブジェクト

Question

RxJavaでEventBusを実装したいので、スティッキーイベントが必要です。私はBehaviorSubjectを使うことができますが、タイプ（クラス名）によって異なるすべてのイベントをキャッシュしたいときに、最後に発行されたアイテムだけをキャッシュします。 ReplaySubjectにはオーバーヘッドがありますが、放射された要素はすべて保持されています。 タイプ要素によってユニークなReplaySubjectを作成する方法はありますか？

Answer 1

私はクリーンな解決策があるとは思わない。しかし、あなたはこの仕事をすることができるかもしれません。

1. イベントタイプごとにBehaviorSubject<>を作成します。 ConcurrentMapを使用して、到着する各イベントを正しいSubjectにディスパッチします。
2. イベントの到着順にそれらのサブジェクトのリストを維持します。
3. 新しいサブスクリプションが受信されると、すべてのサブジェクトのマージであるオブザーバブルが作成されます。サブジェクトの最初のリストは、すでにイベントを受信して​​います。

これは、上記を明確にするコードです。テストされていない。

// The subscription operation will perform a merge of the two lists 
Map<EventType, BehaviorSubject<Event>> map = new ConcurrentHashMap<>(); 
List<BehaviorSubject<Event>> listOfUnseenEvents = new ArrayList<>(); 
List<BehaviorSubject<Event>> listOfSeenEvents = new ArrayList<>(); 
// ... 
listOfUnseenEvents = map.values().asList(); 

public Observable<Event> busSub() { 
    List<BehaviorSubject<Event>> allEvents = new ArrayList<>(); 
    synchronized (map) { 
    allEvents.addAll(listOfSeenEvents); 
    allEvents.addAll(listOfUnseenEvents); 
    } 
    return Observable.merge(allEvents); 
} 

// receive an event and dispatch it 
eventSource 
    .subscribe(event -> processEvent(event)); 

public void processEvent(Event event) { 
    BehaviorSubject<Event> eSubject = map.get(event.getEventType()); 
    synchronized (map) { 
     if (containsEventType(listOfSeenEvents, event.getEventType())) { 
     removeEventType(listOfSeenEvents, event.getEventType()); 
     } else { 
     removeEventType(listOfUnseenEvents, event.getEventType()); 
     } 
     listOfSeenEvents.add(eSubject); 
    } 
    eSubject.onNext(event); 
} 

このコードはmerge()が順番に与えられた観測のそれぞれに加入するという事実を利用していますのでご注意ください。 RxJavaのドキュメントにそのような保証はありません。

すべてのイベントタイプがあらかじめわかっていない場合、これは機能しません。