こんにちは。Origami Advent Calender 22日目の記事になります。

ことのはじまり

最近 RxJava 使い始めたんですが、
毎度おなじみ非同期で投げたイベントの登録元が破棄されていたときに、
多いに困る系の話に悩んでいました。
巷では RxLifecycle というのが割と使われているみたいです。

WeakSubscriber と ReusableCompositeSubscription

ただ、hogehogeActivityを継承する系は避けたかったので調べたところ、
SubscriberのラッパーとCompositeSubscriptionのラッパーをつくって
頑張っている人がいたので参考にさせていただきました。

SubscriberのラッパーではWeakSubscrberというものがあり、
やり方はこちらのほうが全体的にすっきりしそうでしたが、
インタフェースに影響してしまうことがあるのでちょっと断念。

そこで差し込むように使える
もう一方のCompositeSubscriptionのラッパー
ReusableCompositeSubscriptionにしました。
-> 参考コード

CompositeSubscriptionというのは、RxJavaに入っているSubscriptionを
まとめてunsubscribeできるものです。

ただ、一度unsubscribeすると再利用できないので、
ReusableCompositeSubscriptionでは
一度画面遷移などしてまた戻ったときに作り直すという作業を
使う側が気にせずできるようにしたものというわけです。

早すぎるイベント登録に対処

このCompositeSubscription自体は、
onStartでSubscriptionをadd
onStopでunsubscribeするのを推奨しています。

また、Viewのライフサイクルの早い段階でSubscriptionをaddすると
落ちてしまうケースがあったので、
onStartの前までにaddしたSubscriptionをプールする作りにしてみました。

ReusableCompositeSubscription.java


public class ReusableCompositeSubscription {

    private CompositeSubscription mCompositeSubscription;

    private boolean subscrisable = false;
    private boolean firstSubscribe = true;
    private List<Subscription> mUnregisteredSubscriptions = new ArrayList<Subscription>();
    private List<Subscription> mPoolSubscriptions = new ArrayList<Subscription>();

    /**
     * Add subscription to CompositeSubscription if prepared.
     * @param subscription
     */
    public void add(@NonNull Subscription subscription) {
        if (mCompositeSubscription == null) {
            mCompositeSubscription = new CompositeSubscription();
        }

        if (subscription == null) {
            return;
        }

        mPoolSubscriptions.add(subscription);
        if (subscrisable) {
            mCompositeSubscription.add(subscription);
        } else {
            mUnregisteredSubscriptions.add(subscription);
        }
    }

    /**
     * Should be called 'onStart','onDraw'...
     * Enable to become prepared CompositeSubscription frag and Add pools to CompositeSubscription
     */
    public void subscribe() {
        subscrisable = true;
        if (mCompositeSubscription == null) {
            mCompositeSubscription = new CompositeSubscription();
        }

        if (firstSubscribe) {
            firstSubscribe = false;
            for (Subscription s : mUnregisteredSubscriptions) {
                if (s != null) {
                    mCompositeSubscription.add(s);
                }
            }
            mUnregisteredSubscriptions = new ArrayList<Subscription>();
        } else {
            // seconde or later subscribe
            for (Subscription s : mPoolSubscriptions) {
                if (s != null) {
                    mCompositeSubscription.add(s);
                }
            }
        }
    }

    /**
     * Should be called 'onStop'...
     */
    public void unsubscribe() {
        subscrisable = false;
        if (mCompositeSubscription != null) {
            mCompositeSubscription.unsubscribe();
            mCompositeSubscription = null;
        }
    }
}

使い方

TestActivity.java


public class TestActivity extends SingleFragmentActivity {

    private ReusableCompositeSubscription mCompositeSubscription = 
                                   new ReusableCompositeSubscription();

    @Override
    protected Fragment buildFragment() {
        return null;
    }

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);

        // Scheduler は Schedulers.io() に投げて、
        // Android の main thread で受け取るようにhogehoge
        Subscription s = APIClient.get(new Subscriber<Result>() {
            @Override
            public void onCompleted() {
                // Completed
            }

            @Override
            public void onError(Throwable e) {
                // Error
            }

            @Override
            public void onNext(Result result) {
                // Success
            }
        });
        // この時点で original の CompositeSubscription は生成されておらず
　　　　　　　　　　　　　　　　// add した Subscription　はプールされています
        mCompositeSubscription.add(s);
    }

    @Override
    protected void onStart() {
        super.onStart();
        // ここでoriginalの CompositeSubscription を生成、
        // プールした　Subscription　を実際にはaddされている
        mCompositeSubscription.subscribe();
    }

    @Override
    protected void onStop() {
        super.onStop();
        mCompositeSubscription.unsubscribe();
    }
}