baku-toolkit/lib/src/main/java/io/v/rx/syncbase/RxTable.java
// TODO(rosswang): Possibly unsubscribe previous watch on mRxDb onNext.
Possibly unsubscribe previous watch on mRxDb onNext.
Possibly unsubscribe previous watch on mRxDb onNext .
private <T> Observable<T> watch(final Action2<Database, Subscriber<? super T>> subscribeWatch) {
        return Observable.<T>create(s -> mRxDb.getObservable()
                .subscribe(db -> subscribeWatch.call(db, s), s::onError));
    }
private < T > Observable < T > watch ( final Action2 < Database , Subscriber < ? super T > > subscribeWatch ) { return Observable . < T > create ( s -> mRxDb . getObservable ( ) . subscribe ( db -> subscribeWatch . call ( db , s ) , s :: onError ) ) ; }
TODO(rosswang): Possibly unsubscribe previous watch on mRxDb onNext.
TODO ( rosswang ) : Possibly unsubscribe previous watch on mRxDb onNext .
//onComplete is connected by subscribeWatch/observeWatchStream.subscribe