Android SocketIO Rx extension

SocketIO+Rx.kt

import io.reactivex.Observable
import io.reactivex.Single
import io.reactivex.disposables.Disposables
import io.socket.client.IO
import io.socket.client.Socket
import io.socket.emitter.Emitter

class RxIO(uri: String) {
    private val socket = IO.socket(uri)

    @androidx.annotation.CheckResult
    fun connect(): Single<RxSocket> = doConnect().doOnDispose {
        socket.disconnect()
    }

    @androidx.annotation.CheckResult
    private fun doConnect(): Single<RxSocket> = Single.create { emitter ->
        socket.on(Socket.EVENT_CONNECT) {
            emitter.onSuccess(RxSocket(socket))
        }
        socket.connect()
    }
}

class RxSocket(private val socket: Socket) {
    @androidx.annotation.CheckResult
    fun on(event: String): Observable<Array<Any>> = Observable.create { emitter ->
        val listener = Emitter.Listener {
            emitter.onNext(it)
        }
        socket.on(event, listener)
        emitter.setDisposable(Disposables.fromRunnable {
            socket.off(event, listener)
        })
    }
}