FirebaseDatabase events listener with RxKotlin

RxFirebaseDatabase.kt

import com.google.firebase.database.*
import rx.Observable
import rx.lang.kotlin.add
import rx.lang.kotlin.observable

/**
 * @author kakajika
 * @since 2016/11/08
 */
class RxFirebaseDatabase {
    enum class ChildEventType {
        ADDED,
        MOVED,
        CHANGED,
        REMOVED
    }
    class ChildEventPayload(
            val eventType: ChildEventType,
            val snapshot: DataSnapshot,
            val previousChildName: String? = null
    ) {}
}

// region Single Value Event

fun Query.getSnapshot(): Observable<DataSnapshot> {
    return observable { subscriber ->
        val listener = object : ValueEventListener {
            override fun onDataChange(snapshot: DataSnapshot) {
                subscriber.onNext(snapshot)
                subscriber.onCompleted()
            }
            override fun onCancelled(error: DatabaseError) {
                subscriber.onCompleted()
            }
        }
        addListenerForSingleValueEvent(listener)
        subscriber.add {
            removeEventListener(listener)
        }
    }
}

inline fun <reified T: Any> Query.getValue(): Observable<T> {
    return getSnapshot().map { snapshot ->
        snapshot.getValue(T::class.java)
    }
}

fun Query.getChildrenSnapshots(): Observable<Iterable<DataSnapshot>> {
    return getSnapshot().map { snapshot ->
        snapshot.children
    }
}

inline fun <reified T: Any> Query.getChildrenValues(): Observable<Iterable<T>> {
    return getSnapshot().map { snapshot ->
        snapshot.children.map { it.getValue(T::class.java) }
    }
}

// endregion

// region Observe Value Event

fun Query.observeValueSnapshot(): Observable<DataSnapshot> {
    return observable { subscriber ->
        val listener = object : ValueEventListener {
            override fun onDataChange(snapshot: DataSnapshot) {
                subscriber.onNext(snapshot)
            }
            override fun onCancelled(error: DatabaseError) {
                subscriber.onCompleted()
            }
        }
        addValueEventListener(listener)
        subscriber.add {
            removeEventListener(listener)
        }
    }
}

inline fun <reified T: Any> Query.observeValue(): Observable<T> {
    return observeValueSnapshot().map { snapshot ->
        snapshot.getValue(T::class.java)
    }
}

fun Query.observeChildrenSnapshots(): Observable<Iterable<DataSnapshot>> {
    return observeValueSnapshot().map { snapshot ->
        snapshot.children
    }
}

inline fun <reified T: Any> Query.observeChildrenValues(): Observable<Iterable<T>> {
    return observeValueSnapshot().map { snapshot ->
        snapshot.children.map { it.getValue(T::class.java) }
    }
}

// endregion

// region Observe Child Event

fun Query.observeChildEvents(): Observable<RxFirebaseDatabase.ChildEventPayload> {
    return observable { subscriber ->
        val listener = object : ChildEventListener {
            override fun onChildAdded(snapshot: DataSnapshot, previousChildName: String?) {
                subscriber.onNext(RxFirebaseDatabase.ChildEventPayload(RxFirebaseDatabase.ChildEventType.ADDED, snapshot, previousChildName))
            }
            override fun onChildMoved(snapshot: DataSnapshot, previousChildName: String?) {
                subscriber.onNext(RxFirebaseDatabase.ChildEventPayload(RxFirebaseDatabase.ChildEventType.MOVED, snapshot, previousChildName))
            }
            override fun onChildChanged(snapshot: DataSnapshot, previousChildName: String?) {
                subscriber.onNext(RxFirebaseDatabase.ChildEventPayload(RxFirebaseDatabase.ChildEventType.CHANGED, snapshot, previousChildName))
            }
            override fun onChildRemoved(snapshot: DataSnapshot) {
                subscriber.onNext(RxFirebaseDatabase.ChildEventPayload(RxFirebaseDatabase.ChildEventType.REMOVED, snapshot))
            }
            override fun onCancelled(error: DatabaseError) {
                subscriber.onCompleted()
            }
        }
        addChildEventListener(listener)
        subscriber.add {
            removeEventListener(listener)
        }
    }
}

fun Query.observeChildEvent(eventType: RxFirebaseDatabase.ChildEventType): Observable<RxFirebaseDatabase.ChildEventPayload> {
    return observeChildEvents().filter { payload ->
        payload.eventType == eventType
    }
}

// endregion

RxFirebaseDatabaseExample.kt

val users = FirebaseDatabase.getInstance().reference.child("users")

// Single value event
users.getChildrenValues<User>().subscribe() { user -> 
    println(user.name)
}

// Observe value changed event
users.observeChildrenValues<User>().subscribe() { user ->
    println(user.name)                                                 
}

// Observe child event
users.observeChildEvent(RxFirebaseDatabase.ChildEventType.ADDED).subscribe() { payload ->
    println(payload.snapshot.key)
}