Firebase Callback to Rx Observable

firebaseRepo.kt

private class ValueEventListenerToEmitter(private val emitter: Emitter<DataSnapshot>) : ValueEventListener {

    override fun onCancelled(error: DatabaseError) {
        emitter.onError(error.toException())
    }

    override fun onDataChange(snapshot: DataSnapshot) {
        emitter.onNext(snapshot)
    }
}

private fun (() -> DatabaseReference).observe(): Observable<DataSnapshot> = Observable.create { emitter ->
    val dbRef = this()
    val eventListener = ValueEventListenerToEmitter(emitter)
    dbRef.addValueEventListener(eventListener)
    emitter.setCancellable { dbRef.removeEventListener(eventListener) }
}