Firebase Kotlin RxJava

RxFirebase.kt

import com.firebase.client.*
import com.firebase.client.core.view.Event
import rx.Observable
import rx.subscriptions.Subscriptions


data class RxFirebasePayload(val dataSnapshot: DataSnapshot, val previousChildKey: String? = null)

class RxFirebase{

    private data class RxFirebasePayloadWithEventType(val payload: RxFirebasePayload, val eventType: Event.EventType)

    public val query : Query
    public val ref : Firebase
        get() = this.query.ref


    constructor(url: String){
        this.query = Firebase(url)
    }

    constructor(query: Query){
        this.query = query
    }


    public fun rx_observe(eventType: Event.EventType) : Observable<RxFirebasePayload> {
        val listenerObs : Observable<RxFirebasePayloadWithEventType> = Observable.create { subscriber ->
            val listener = object : ChildEventListener {
                override fun onChildRemoved(snapshot: DataSnapshot) {
                    val payload = RxFirebasePayload(snapshot)
                    subscriber.onNext(RxFirebasePayloadWithEventType(payload, Event.EventType.CHILD_REMOVED))

                }
                override fun onCancelled(firebaseError: FirebaseError) {
                    val error = FirebaseException(firebaseError.message);
                    subscriber.onError(error);
                }

                override fun onChildChanged(snapshot: DataSnapshot, previousChildKey: String?) {
                    val payload = RxFirebasePayload(snapshot, previousChildKey)
                    subscriber.onNext(RxFirebasePayloadWithEventType(payload, Event.EventType.CHILD_CHANGED))
                }

                override fun onChildMoved(snapshot: DataSnapshot, previousChildKey: String?) {
                    val payload = RxFirebasePayload(snapshot, previousChildKey)
                    subscriber.onNext(RxFirebasePayloadWithEventType(payload, Event.EventType.CHILD_MOVED))
                }

                override fun onChildAdded(snapshot: DataSnapshot, previousChildKey: String?) {
                    val payload = RxFirebasePayload(snapshot, previousChildKey)
                    subscriber.onNext(RxFirebasePayloadWithEventType(payload, Event.EventType.CHILD_ADDED))
                }
            }
            this.query.addChildEventListener(listener);
            subscriber.add(Subscriptions.create {
                this.query.removeEventListener(listener);
            });
        }
        val finalObservable = listenerObs.filter{ pay -> pay.eventType == eventType}
            .map { pay -> pay.payload}
        return finalObservable
    }

    fun child(path: String) : RxFirebase {
        var newQuery : Query = this.query.ref.child(path);
        return RxFirebase(query = newQuery)
    }


    fun rx_setValue(map: Map<String, Object>, priority: Object? = null): Observable<Firebase> {
        return Observable.create { subscriber ->
            val listener = object: Firebase.CompletionListener {
                override fun onComplete(firebaseError: FirebaseError?, updatedRef: Firebase?) {
                    if(firebaseError != null){
                        subscriber.onError(FirebaseException(firebaseError.message))
                    }else{
                        subscriber.onNext(updatedRef);
                        subscriber.onCompleted();
                    }
                }
            }
            if(priority == null){
                this.ref.setValue(map, listener);
            }else{
                this.ref.setValue(map, priority, listener)
            }
        }
    }

    fun rx_updateChildren(map : Map<String, Object>) : Observable<Firebase> {
        return Observable.create { subscriber ->
            val listener = object: Firebase.CompletionListener {
                override fun onComplete(firebaseError: FirebaseError?, updatedRef: Firebase?) {
                    if(firebaseError != null){
                        subscriber.onError(FirebaseException(firebaseError.message))
                    }else{
                        subscriber.onNext(updatedRef);
                        subscriber.onCompleted();
                    }
                }
            }
            this.ref.updateChildren(map, listener);
        }
    }
}