Generic Observable for turning BroadcastManagers into observable streams.

ObservableBroadcastReceiver.kt

package com.aniketkadam.wifimanager.androidhelpers

import android.content.BroadcastReceiver
import android.content.Context
import android.content.Intent
import android.content.IntentFilter
import io.reactivex.Observable
import io.reactivex.ObservableEmitter
import io.reactivex.ObservableOnSubscribe
import io.reactivex.disposables.Disposable
import java.io.Closeable
import java.util.concurrent.atomic.AtomicBoolean

class ObservableBroadcastReceiver<T> private constructor(
    val context: Context,
    val intentFilter: IntentFilter,
    val intentMapper: (Intent) -> T
) : ObservableOnSubscribe<T>,
    Disposable {

    companion object {
        fun <T> create(context: Context, intentFilter: IntentFilter, intentMapper: (Intent) -> T) =
            Observable.create(ObservableBroadcastReceiver(context, intentFilter, intentMapper))
    }

    private val isDisposed = AtomicBoolean(false)

    override fun isDisposed(): Boolean = isDisposed.get()

    override fun dispose() {
        broadcastBroadcastReceiverTemplate.close()
        isDisposed.set(true)
    }

    private lateinit var broadcastBroadcastReceiverTemplate: BroadcastReceiverTemplate

    override fun subscribe(emitter: ObservableEmitter<T>) {
        broadcastBroadcastReceiverTemplate =
                BroadcastReceiverTemplate(context, { i ->
                    isDisposed.get().also { disposed ->
                        if (!disposed) {
                            emitter.onNext(intentMapper(i))
                        }
                    }
                }, intentFilter)

        broadcastBroadcastReceiverTemplate.start()
    }

}

private class BroadcastReceiverTemplate(
    context: Context,
    val onEvent: (Intent) -> Unit,
    val intentFilter: IntentFilter
) : BroadcastReceiver(), Closeable {

    private var context: Context? =
        context // Have to allow for nulling out later, despite requiring a non-null one initially

    override fun onReceive(context: Context, intent: Intent) {
        onEvent(intent)
    }

    internal fun start() {
        context?.registerReceiver(this, intentFilter)
    }

    override fun close() {
        context?.unregisterReceiver(this)
        context = null
    }

}

WifiManagerService.kt

package com.aniketkadam.wifimanager.ui

import android.app.Service
import android.content.Intent
import android.content.IntentFilter
import android.net.wifi.WifiManager
import android.os.IBinder
import android.util.Log
import com.aniketkadam.wifimanager.androidhelpers.ObservableBroadcastReceiver
import io.reactivex.disposables.Disposable

class WifiManagerService : Service() {

    lateinit var disposable: Disposable

    override fun onCreate() {
        disposable =
                ObservableBroadcastReceiver.create(
                    this,
                    IntentFilter(WifiManager.RSSI_CHANGED_ACTION)
                ) { intent -> intent.getIntExtra(WifiManager.EXTRA_NEW_RSSI, -9000) }
                    .subscribe {
                        Log.d("WifiManagerService", "Wifi Strength is now ${it} rssi")
                    }
    }

    override fun onDestroy() {
        super.onDestroy()
        disposable.dispose()
    }

    override fun onStartCommand(intent: Intent?, flags: Int, startId: Int): Int = START_STICKY
    override fun onBind(intent: Intent): IBinder? = null
}