RxJava2.0 Broadcast Receiver for getting intent broadcasts as Observable<Intent>

example.java

IntentFilter intentFilter = new IntentFilter();
intentFilter.addAction("com.example.mat.ACTION_START");
intentFilter.addAction("com.example.mat.ACTION_STOP");
                       
Disposable disposable = RxBroadcastReceiver.create(this, intentFilter)
  .share() // optional if you want do not want multiply broadcast receivers created
  .map(SOMECLASS::someStaticMethod) // example mapping method for parsing data from Intent
  .subscribe(data -> {......});
  

disposable.dispose(); // that or use CompositeDisposable

SOMECLASS: 
public static Data someStaticMethod(Intent intent) {
  // fetch data from intent EXTRAs
}

RxBroadcastReceiver.java

/**
 * RxJava based broadcast reciever that registers its local BroadcastReceiver until end of subscription.
 * Listens for update and passes Intent to the Stream (Subscriber).
 * 
 * Copyright 2016 Mateusz Perlak - http://www.apache.org/licenses/LICENSE-2.0
 * Created on 11/18/16.
 */
public class RxBroadcastReceiver implements ObservableOnSubscribe<Intent> {
    protected final WeakReference<Context> contextWeakReference;

    private IntentFilter intentFilter;

    /**
     * Creates Observable with intent filter for Broadcast receiver.
     *
     * @param context
     * @param intentFilter
     * @return
     */
    public static Observable<Intent> create(Context context, IntentFilter intentFilter) {
        return Observable.defer(() -> Observable.create(new RxBroadcastReceiver(context, intentFilter))
                .subscribeOn(Schedulers.io())
        );
    }

    /**
     * @param context
     * @param intentFilter
     */
    private RxBroadcastReceiver(Context context, IntentFilter intentFilter) {
        contextWeakReference = new WeakReference<Context>(context.getApplicationContext());
        this.intentFilter = intentFilter;
    }

    @Override
    public void subscribe(ObservableEmitter<Intent> emitter) throws Exception {
        BroadcastReceiver broadcastReceiver = new BroadcastReceiver() {
            @Override
            public void onReceive(Context context, Intent intent) {
                emitter.onNext(intent);
            }
        };
        emitter.setDisposable(Disposables.fromRunnable(() -> { // thank you Jake W.
            if (contextWeakReference != null && contextWeakReference.get() != null) {
                contextWeakReference.get().unregisterReceiver(broadcastReceiver);
            }
        }));

        if (contextWeakReference != null && contextWeakReference.get() != null) {
            contextWeakReference.get().registerReceiver(broadcastReceiver, intentFilter);
        }
    }
}

If you set broadcastReceiver to null in dispose then how is this observable reusable? broadcastReceiver is only created in the constructor.

Oh ok got it, Observable.defer will make sure a new BroadcastReceiver is created every time?

What if we just set emitter to null and set isDisposed to emitter == null and reuse the BroadcastReceiver? Which would be a better way?

Hello, I'm curious, isn't better to use just this one:

    @NonNull
    public static Observable<Intent> create(@NonNull final Context context, @NonNull final IntentFilter intentFilter) {
        final Context appContext = context.getApplicationContext();
        return Observable.create(emitter -> {
            final BroadcastReceiver receiver = new BroadcastReceiver() {
                @Override
                public void onReceive(final Context context, final Intent intent) {
                    emitter.onNext(intent);
                }
            };
            appContext.registerReceiver(receiver, intentFilter);
            emitter.setCancellable(() -> appContext.unregisterReceiver(receiver));
        });
    }

Some notes here: we don't need WeakReference for Application Context, because dynamic receiver exists only in the scope of current Application instance (it is a Singleton for Android app process).