baku-toolkit/lib/src/main/java/io/v/rx/syncbase/RxTable.java
/**
     * Wraps a prefix watch stream in a key-specific observable. It remains to be seen whether it
     * will be better to feature-request an exact-match watch API from Syncbase or consolidate all
     * watches into one stream. Exact-match presents a cleaner API boundary but results in more
     * underlying streams, whereas consolidating at the library level will usually be more efficient
     * unless large portions of data won't need to be watched, and also it opens up questions of
     * whether we should computationally optimize the prefix query.
     *
     * @return an observable wrapping the watch stream. This observable should only be subscribed to
     * once, as we can only iterate over the underlying stream once.
     */
Wraps a prefix watch stream in a key-specific observable. It remains to be seen whether it
will be better to feature-request an exact-match watch API from Syncbase or consolidate all
watches into one stream. Exact-match presents a cleaner API boundary but results in more
underlying streams, whereas consolidating at the library level will usually be more efficient
unless large portions of data won't need to be watched, and also it opens up questions of
whether we should computationally optimize the prefix query.

@return an observable wrapping the watch stream. This observable should only be subscribed to
once, as we can only iterate over the underlying stream once.
Wraps a prefix watch stream in a key - specific observable . It remains to be seen whether it will be better to feature - request an exact - match watch API from Syncbase or consolidate all watches into one stream . Exact - match presents a cleaner API boundary but results in more underlying streams whereas consolidating at the library level will usually be more efficient unless large portions of data won ' t need to be watched and also it opens up questions of whether we should computationally optimize the prefix query . @return an observable wrapping the watch stream . This observable should only be subscribed to once as we can only iterate over the underlying stream once .
private static <T> Observable<SingleWatchEvent<T>> observeWatchStream(
            final InputChannel<WatchChange> s, final String key, final TypeToken<T> tt,
            final T defaultValue) {
        return RxInputChannel.wrap(s)
                .autoConnect()
                .filter(c -> c.getRowName().equals(key))
                .concatMap(VFn.wrap(c -> SingleWatchEvent.fromWatchChange(c, tt, defaultValue)))
                .distinctUntilChanged();
    }
private static < T > Observable < SingleWatchEvent < T > > observeWatchStream ( final InputChannel < WatchChange > s , final String key , final TypeToken < T > tt , final T defaultValue ) { return RxInputChannel . wrap ( s ) . autoConnect ( ) . filter ( c -> c . getRowName ( ) . equals ( key ) ) . concatMap ( VFn . wrap ( c -> SingleWatchEvent . fromWatchChange ( c , tt , defaultValue ) ) ) . distinctUntilChanged ( ) ; }
Wraps a prefix watch stream in a key-specific observable.
Wraps a prefix watch stream in a key - specific observable .
// About the Vfn.wrap, on error, the wrapping replay will disconnect, // calling cancellation (see cancelOnDisconnect). The possible source of // VException here is VOM decoding.