magic-6.kt

class WebSocketPushUpdates(private val context: Context) : PushUpdates {

    private val jobSubject = PublishSubject.create<Job>().toSerialized()

    override val jobs: Observable<Job> = jobSubject
            .doOnSubscribe { bindService() }
            .doOnDispose { unbindService() }

    private val connection = ServiceConnection()

    private var bindingsCount = 0

    @Synchronized
    private fun bindService() {
        bindingsCount++
        if (bindingsCount == 1) {
            context.bindService(
                    Intent(context, WebSocketService::class.java),
                    connection,
                    Context.BIND_AUTO_CREATE)
        }
    }

    @Synchronized
    private fun unbindService() {
        bindingsCount--
        if (bindingsCount == 0) context.unbindService(connection)
    }

    inner class ServiceConnection : android.content.ServiceConnection {

        var jobUpdatesDisposable: Disposable? = null

        override fun onServiceConnected(
            name: ComponentName, 
            service: IBinder
        ) {
            val binder = service as WebSocketService.Binder
            jobUpdatesDisposable = binder.jobUpdates
                    .subscribe { jobSubject.onNext(it) }
        }

        override fun onServiceDisconnected(name: ComponentName?) {
            jobUpdatesDisposable?.dispose()
        }
    }

}

class WebSocketService : Service() {

    private val jobSubject = PublishSubject.create<Job>().toSerialized()

    override fun onCreate() {
        // Connects to websocket and starts listening events.

        // When job updates are received, they are published
        // to jobSubject and also used to update JobCache.
    }

    override fun onDestroy() {
        // Disconnects from websocket.
    }

    override fun onBind(intent: Intent): IBinder? = Binder()

    inner class Binder : android.os.Binder() {
        val jobUpdates: Observable<Job> = jobSubject
    }

}