RxBroadcastReceiver.java

package rx.android.observables;

import rx.Observable;
import rx.Observable.OnSubscribe;
import rx.Subscriber;
import rx.android.subscriptions.AndroidSubscriptions;
import rx.functions.Action0;
import android.content.BroadcastReceiver;
import android.content.Context;
import android.content.Intent;
import android.content.IntentFilter;

public final class RxBroadcastReceiver {

    public static class IntentWithContext {
        private Context context;
        private Intent intent;

        public IntentWithContext(Context context, Intent intent) {
            this.context = context;
            this.intent = intent;
        }

        public Context getContext() {
            return context;
        }

        public Intent getIntent() {
            return intent;
        }
    }

    public static Observable<IntentWithContext> fromBroadcast(final Context context, final IntentFilter filter) {
        return Observable.create(new OnSubscribe<IntentWithContext>() {

            @Override
            public void call(final Subscriber<? super IntentWithContext> subscriber) {
                ViewObservable.assertUiThread();
                final BroadcastReceiver receiver = new BroadcastReceiver() {

                    @Override
                    public void onReceive(Context context, Intent intent) {
                        subscriber.onNext(new IntentWithContext(context, intent));
                    }

                };
                context.registerReceiver(receiver, filter);
                subscriber.add(AndroidSubscriptions.unsubscribeInUiThread(new Action0() {

                    @Override
                    public void call() {
                        context.unregisterReceiver(receiver);
                    }
                }));
            }
        });
    }
}