RxJavaは、連鎖的な非同期処理だけでなく、イベントバスとしてメッセージやイベントの伝達機能を実現するためにも活用できます。ここでは、RxJavaを用いたRxBusの実装方法と利用例について解説します。

RXBUSの実装

RxJavaでイベントバスを実装するためのクラスです。

import java.util.HashMap;
import java.util.Map;
import rx.Observable;
import rx.Subscription;
import rx.android.schedulers.AndroidSchedulers;
import rx.functions.Action0;
import rx.functions.Action1;
import rx.schedulers.Schedulers;
import rx.subjects.PublishSubject;
import rx.subjects.SerializedSubject;
import rx.subjects.Subject;
import rx.subscriptions.CompositeSubscription;

public class RxBus {

    private static volatile RxBus instance;
    // SubjectはObservableとObserverの両方のインターフェースを実装しており、
    // オブザーバーと被オブザーバーの両方の役割を果たすことができます。
    private final Subject<Object, Object> subject;
    // 複数のSubscriptionオブジェクトを管理し、一括でキャンセルするために使用します。
    private final Map<String, CompositeSubscription> subscriptions;

    private RxBus() {
        // PublishSubjectはホットなObservableであり、
        // 新しいオブザーバーが購読すると、それ以降のイベントのみを通知します。
        // SerializedSubjectはSubjectをスレッドセーフにします。
        subject = new SerializedSubject<>(PublishSubject.create());
        subscriptions = new HashMap<>();
    }

    public static RxBus getInstance() {
        if (instance == null) {
            synchronized (RxBus.class) {
                if (instance == null) {
                    instance = new RxBus();
                }
            }
        }
        return instance;
    }

    /**
     * イベントをポストします。
     * @param event ポストするイベントオブジェクト。
     */
    public void post(Object event) {
        subject.onNext(event);
    }

    /**
     * 指定された型のObservableインスタンスを返します。
     * @param eventType 処理するイベントの型。
     * @param <T> イベントの型。
     * @return 指定された型のイベントを流すObservable。
     */
    public <T> Observable<T> toObservable(Class<T> eventType) {
        // ofType演算子は、指定された型と一致するイベントのみをフィルタリングします。
        return subject.ofType(eventType);
    }

    /**
     * 現在購読者がいるかどうかを確認します。
     * @return 購読者がいる場合はtrue、そうでない場合はfalse。
     */
    public boolean hasObservers() {
        return subject.hasObservers();
    }

    /**
     * デフォルトの購読メソッド（エラーハンドリングなし）。
     * @param eventType 購読するイベントの型。
     * @param onNext イベント受信時のコールバック。
     * @param <T> イベントの型。
     * @return 作成されたSubscription。
     */
    public <T> Subscription subscribe(Class<T> eventType, Action1<T> onNext) {
        return toObservable(eventType)
                .subscribeOn(Schedulers.io()) // I/Oスレッドで実行
                .observeOn(AndroidSchedulers.mainThread()) // メインスレッドで結果を処理
                .subscribe(onNext);
    }

    /**
     * デフォルトの購読メソッド（エラーハンドリングあり）。
     * @param eventType 購読するイベントの型。
     * @param onNext イベント受信時のコールバック。
     * @param onError エラー発生時のコールバック。
     * @param <T> イベントの型。
     * @return 作成されたSubscription。
     */
    public <T> Subscription subscribe(Class<T> eventType, Action1<T> onNext, Action1<Throwable> onError) {
        return toObservable(eventType)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(onNext, onError);
    }

    /**
     * デフォルトの購読メソッド（完了ハンドリングあり）。
     * @param eventType 購読するイベントの型。
     * @param onNext イベント受信時のコールバック。
     * @param onError エラー発生時のコールバック。
     * @param onComplete イベントシーケンス完了時のコールバック。
     * @param <T> イベントの型。
     * @return 作成されたSubscription。
     */
    public <T> Subscription subscribe(Class<T> eventType, Action1<T> onNext, Action1<Throwable> onError, Action0 onComplete) {
        return toObservable(eventType)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(onNext, onError, onComplete);
    }

    /**
     * 購読されたSubscriptionを保存し、一括でのキャンセルを可能にします。
     * @param owner Subscriptionを所有するオブジェクト。
     * @param subscription 保存するSubscription。
     */
    public void addSubscription(Object owner, Subscription subscription) {
        String key = owner.getClass().getName(); // オブジェクトのクラス名でキーを生成
        CompositeSubscription compositeSubscription = subscriptions.computeIfAbsent(key, k -> new CompositeSubscription());
        compositeSubscription.add(subscription);
    }

    /**
     * 指定されたオブジェクトに関連するすべての購読をキャンセルします。
     * @param owner キャンセルする購読を持つオブジェクト。
     */
    public void unsubscribe(Object owner) {
        String key = owner.getClass().getName();
        CompositeSubscription compositeSubscription = subscriptions.get(key);
        if (compositeSubscription != null && !compositeSubscription.isUnsubscribed()) {
            compositeSubscription.unsubscribe();
        }
        subscriptions.remove(key);
    }
}

使い方

イベントの送信:

RxBus.getInstance().post("これは送信されるメッセージです");

イベントの処理:

//RxBus.getInstance().subscribe(String.class, s -> {
//    Toast.makeText(MainActivity.this, s, Toast.LENGTH_SHORT).show();
//});

// または、より詳細な購読方法
Subscription subscription = RxBus.getInstance().toObservable(String.class)
        .filter(s -> !s.isEmpty()) // 例: 空の文字列を除外
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(s -> {
            Log.i(TAG, "Received: " + s);
        }, throwable -> {
            Log.e(TAG, "Error receiving event", throwable);
        });
RxBus.getInstance().addSubscription(this, subscription); // このアクティビティに関連付けて購読を管理

購読のキャンセル: アクティビティの破棄時などに、メモリリークを防ぐために購読をキャンセルします。

@Override
protected void onDestroy() {
    RxBus.getInstance().unsubscribe(this); // このアクティビティに関連する購読をすべてキャンセル
    super.onDestroy();
}

機能の拡張

PublishSubject は、購読開始以降のイベントのみを通知します。イベント発生後に購読を開始した場合、そのイベントは失われてしまいます。これを避けるためには、イベントの永続化機能（EventBusのStickyイベントのようなもの）が必要です。

PublishSubject の代わりに BehaviorSubject を使用すると、直近の1つのイベントをキャッシュし、後から購読したオブザーバーにもそのイベントを通知できます。

// RxBusコンストラクタ内を修正
// subject = new SerializedSubject<>(BehaviorSubject.create());

さらに多くのイベントをキャッシュしたい場合は、ReplaySubject を使用します。

// RxBusコンストラクタ内を修正
// subject = new SerializedSubject<>(ReplaySubject.create());