Simple RxJava2 wrapper for Android BroadcastReceiver

README.md

RxReceivers

Simple RxJava2 wrapper for Android BroadcastReceiver

How to use it:

RxReceivers.from("YOUR ACTION HERE", context)
    .subscribe(intent -> {    
        //do what you want here, but don't forget to unsubscribe
});

BroadcastDisposable.java

/**
 * A custom implementation of {@link Disposable} for use with {@link RxReceivers} that
 * takes care of un-registering the underlying Broadcast Receiver when it is disposed.
 */
class BroadcastDisposable implements Disposable {

    private BroadcastReceiver receiver;
    private Context ctx;

    private boolean isDisposed = false;

    /**
     * Initializes a new instance of the {@link BroadcastDisposable} class.
     * @param receiver
     * @param ctx
     */
    BroadcastDisposable(@NonNull BroadcastReceiver receiver, @NonNull Context ctx) {
        this.receiver = receiver;
        this.ctx = ctx;
    }

    /**
     * Disposes resources used by the observable (ie.e: the underlying Broadcast Receiver).
     */
    @Override
    public void dispose() {
        if (!isDisposed) {
            ctx.unregisterReceiver(receiver);
            isDisposed = true;
        }
    }

    /**
     * Returns a value indicating whether the underlying resources have been disposed.
     * @return {@code true} if the underlying resources have been disposed, otherwise {@code false}.
     */
    @Override
    public boolean isDisposed() {
        return isDisposed;
    }
}

RxReceivers.java

/**
 * A simple RxJava 2 wrapper for Android {@link BroadcastReceiver}s.
 */
public class RxReceivers {

    /**
     * Registers a {@link BroadcastReceiver} for the specified action,
     * and wraps it in an observable stream.
     *
     * @param action  The action to match.
     * @param context The context used to register the receiver.
     * @return An {@link Observable<Intent>} that will emit received messages.
     */
    public static Observable<Intent> from(@NonNull final String action, @NonNull final Context ctx) {
        IntentFilter filter = new IntentFilter(action);
        return from(filter, ctx);
    }

    /**
     * Registers a {@link android.content.BroadcastReceiver} for the specified action,
     * and wraps it in an observable stream.
     *
     * @param intentFilter The action to match.
     * @param context      The context used to register the receiver.
     * @return An {@link Observable<Intent>} that will emit received messages.
     * 
     */
    public static Observable<Intent> from(@NonNull final IntentFilter intentFilter, @NonNull final Context ctx) {
        return Observable.create(new ObservableOnSubscribe<Intent>() {
            Context appContext = ctx.getApplicationContext();
            @Override
            public void subscribe(@NonNull final ObservableEmitter<Intent> emitter) throws Exception {
                BroadcastReceiver receiver = new BroadcastReceiver() {
                    @Override
                    public void onReceive(Context context, Intent intent) {
                        emitter.onNext(intent);
                    }
                };
                emitter.setDisposable(new BroadcastDisposable(receiver, appContext));
                appContext.registerReceiver(receiver, intentFilter);
            }
        });
    }
}