baku-toolkit/lib/src/main/java/io/v/rx/syncbase/RxTable.java
/**
     * Wraps a watch stream in an observable.
     *
     * @return an observable wrapping the watch stream, grouped by batches. These observables should
     * only be subscribed to once, as we can only iterate over the underlying stream once.
     */
Wraps a watch stream in an observable.
@return an observable wrapping the watch stream, grouped by batches. These observables should
only be subscribed to once, as we can only iterate over the underlying stream once.
Wraps a watch stream in an observable . @return an observable wrapping the watch stream grouped by batches . These observables should only be subscribed to once as we can only iterate over the underlying stream once .
private static <T> Observable<RangeWatchBatch<T>> observeWatchStream(
            final InputChannel<WatchChange> s, @Nullable final Func1<String, Boolean> prefixFilter,
            final TypeToken<T> tt) {
        final Observable<WatchChange> raw = RxInputChannel.wrap(s).autoConnect();
        return Observable.create(subscriber -> {
                    final RangeWatchBatchWindower<T> windower =
                            new RangeWatchBatchWindower<>(subscriber);
                    subscriber.add(raw.subscribe(c -> {
                                if (prefixFilter == null || prefixFilter.call(c.getRowName())) {
                                    try {
                                        windower.onNext(c.getResumeMarker(),
                                                RangeWatchEvent.fromWatchChange(c, tt));
                                    } catch (final VException e) {
                                        windower.onError(c.getResumeMarker(), e);
                                    }
                                }
                                if (!c.isContinued()) {
                                    windower.onBatchEnd();
                                }
                            },
                            subscriber::onError,
                            subscriber::onCompleted
                    ));
                }
        );
    }
private static < T > Observable < RangeWatchBatch < T > > observeWatchStream ( final InputChannel < WatchChange > s , @ Nullable final Func1 < String , Boolean > prefixFilter , final TypeToken < T > tt ) { final Observable < WatchChange > raw = RxInputChannel . wrap ( s ) . autoConnect ( ) ; return Observable . create ( subscriber -> { final RangeWatchBatchWindower < T > windower = new RangeWatchBatchWindower < > ( subscriber ) ; subscriber . add ( raw . subscribe ( c -> { if ( prefixFilter == null || prefixFilter . call ( c . getRowName ( ) ) ) { try { windower . onNext ( c . getResumeMarker ( ) , RangeWatchEvent . fromWatchChange ( c , tt ) ) ; } catch ( final VException e ) { windower . onError ( c . getResumeMarker ( ) , e ) ; } } if ( ! c . isContinued ( ) ) { windower . onBatchEnd ( ) ; } } , subscriber :: onError , subscriber :: onCompleted ) ) ; } ) ; }
Wraps a watch stream in an observable.
Wraps a watch stream in an observable .
// TODO(rosswang): support other RowRange types