RxObservable based Broadcast receiver

gistfile1.txt

fun AppCompatActivity.subscribeToBroadcastsOnLifecycle(action: String, fn: (Intent) -> Unit) {
    observeBroadcasts(action).subscribeOnLifecycle(lifecycle, fn)
}

fun <T> Observable<T>.subscribeOnLifecycle(lifecycle: Lifecycle, fn: (T) -> Unit) {
    val lifecycleObserver: LifecycleObserver = object : LifecycleObserver {
        private var subscription: Disposable? = null

        @OnLifecycleEvent(Lifecycle.Event.ON_START)
        fun onStart() {
            subscription = subscribe(fn)
        }

        @OnLifecycleEvent(Lifecycle.Event.ON_STOP)
        fun onStop() {
            subscription?.dispose()
        }
    }
    lifecycle.addObserver(lifecycleObserver)
}

fun Context.observeBroadcasts(action: String): Observable<Intent> {
    return observeBroadcasts(IntentFilter(action))
}

fun Context.observeBroadcasts(intentFilter: IntentFilter): Observable<Intent> {
    val observable = Observable.create<Intent> { observer ->
        val receiver = object : BroadcastReceiver() {
            override fun onReceive(context: Context, intent: Intent) {
                observer.onNext(intent)
            }
        }

        observer.setDisposable(Disposables.fromRunnable {
            unregisterReceiver(receiver)
        })
        registerReceiver(receiver, intentFilter)
    }

    return observable
        .subscribeOn(AndroidSchedulers.mainThread())
        .observeOn(AndroidSchedulers.mainThread())
}