syncbase/src/main/java/io/v/syncbase/Database.java
/**
     * Notifies {@code h} of initial state, and of all subsequent changes to this database.
     */
Notifies  h of initial state, and of all subsequent changes to this database.
Notifies h of initial state and of all subsequent changes to this database .
public void addWatchChangeHandler(final WatchChangeHandler h, final AddWatchChangeHandlerOptions opts) {
        if (opts.resumeMarker != null && opts.resumeMarker.length != 0) {
            throw new UnsupportedOperationException("Specifying resumeMarker is not yet supported");
        }
        final AtomicBoolean canceled = new AtomicBoolean(false);
        mCoreDatabase.watch(null, ImmutableList.of(opts.getCollectionRowPattern()),
                new io.v.syncbase.core.Database.WatchPatternsCallbacks() {
                    private boolean mGotFirstBatch = false;
                    private final List<WatchChange> mBatch = new ArrayList<>();
                    @Override
                    public void onChange(io.v.syncbase.core.WatchChange coreWatchChange) {
                        if (canceled.get()) {
                            return;
                        }
                        boolean isRoot = coreWatchChange.entityType ==
                                io.v.syncbase.core.WatchChange.EntityType.ROOT;
                        boolean isUserdataInternalRow =
                                coreWatchChange.entityType ==
                                        io.v.syncbase.core.WatchChange.EntityType.ROW &&
                                coreWatchChange.collection.name.equals(Syncbase.USERDATA_NAME) &&
                                coreWatchChange.row.startsWith(Syncbase.USERDATA_INTERNAL_PREFIX);
                        if (!isRoot && (opts.showUserdataInternalRow || !isUserdataInternalRow)) {
                            mBatch.add(new WatchChange(coreWatchChange));
                        }
                        if (!coreWatchChange.continued) {
                            final SettableFuture<Boolean> setFuture = SettableFuture.create();
                            if (!mGotFirstBatch) {
                                mGotFirstBatch = true;
                                final List<WatchChange> cpBatch = mBatch;
                                Syncbase.sOpts.mCallbackExecutor.execute(new Runnable() {
                                    @Override
                                    public void run() {
                                        h.onInitialState(cpBatch.iterator());
                                        setFuture.set(true);
                                    }
                                });
                            } else {
                                final List<WatchChange> cpBatch = mBatch;
                                Syncbase.sOpts.mCallbackExecutor.execute(new Runnable() {
                                    @Override
                                    public void run() {
                                        h.onChangeBatch(cpBatch.iterator());
                                        setFuture.set(true);
                                    }
                                });
                            }
                            try {
                                setFuture.get();
                            } catch (InterruptedException | ExecutionException e) {
                                e.printStackTrace();
                                System.err.println(e.toString());
                            }
                            mBatch.clear();
                        }
                    }
                    @Override
                    public void onError(VError vError) {
                        h.onError(vError);
                    }
                });
        synchronized (mWatchChangeHandlersMu) {
            mWatchChangeHandlers.put(h, new Runnable() {
                @Override
                public void run() {
                    canceled.set(true);
                }
            });
        }
    }
public void addWatchChangeHandler ( final WatchChangeHandler h , final AddWatchChangeHandlerOptions opts ) { if ( opts . resumeMarker != null && opts . resumeMarker . length != 0 ) { throw new UnsupportedOperationException ( "Specifying resumeMarker is not yet supported" ) ; } final AtomicBoolean canceled = new AtomicBoolean ( false ) ; mCoreDatabase . watch ( null , ImmutableList . of ( opts . getCollectionRowPattern ( ) ) , new io . v . syncbase . core . Database . WatchPatternsCallbacks ( ) { private boolean mGotFirstBatch = false ; private final List < WatchChange > mBatch = new ArrayList < > ( ) ; @ Override public void onChange ( io . v . syncbase . core . WatchChange coreWatchChange ) { if ( canceled . get ( ) ) { return ; } boolean isRoot = coreWatchChange . entityType == io . v . syncbase . core . WatchChange . EntityType . ROOT ; boolean isUserdataInternalRow = coreWatchChange . entityType == io . v . syncbase . core . WatchChange . EntityType . ROW && coreWatchChange . collection . name . equals ( Syncbase . USERDATA_NAME ) && coreWatchChange . row . startsWith ( Syncbase . USERDATA_INTERNAL_PREFIX ) ; if ( ! isRoot && ( opts . showUserdataInternalRow || ! isUserdataInternalRow ) ) { mBatch . add ( new WatchChange ( coreWatchChange ) ) ; } if ( ! coreWatchChange . continued ) { final SettableFuture < Boolean > setFuture = SettableFuture . create ( ) ; if ( ! mGotFirstBatch ) { mGotFirstBatch = true ; final List < WatchChange > cpBatch = mBatch ; Syncbase . sOpts . mCallbackExecutor . execute ( new Runnable ( ) { @ Override public void run ( ) { h . onInitialState ( cpBatch . iterator ( ) ) ; setFuture . set ( true ) ; } } ) ; } else { final List < WatchChange > cpBatch = mBatch ; Syncbase . sOpts . mCallbackExecutor . execute ( new Runnable ( ) { @ Override public void run ( ) { h . onChangeBatch ( cpBatch . iterator ( ) ) ; setFuture . set ( true ) ; } } ) ; } try { setFuture . get ( ) ; } catch ( InterruptedException | ExecutionException e ) { e . printStackTrace ( ) ; System . err . println ( e . toString ( ) ) ; } mBatch . clear ( ) ; } } @ Override public void onError ( VError vError ) { h . onError ( vError ) ; } } ) ; synchronized ( mWatchChangeHandlersMu ) { mWatchChangeHandlers . put ( h , new Runnable ( ) { @ Override public void run ( ) { canceled . set ( true ) ; } } ) ; } }
Notifies {@code h} of initial state, and of all subsequent changes to this database.
Notifies { @code h } of initial state and of all subsequent changes to this database .
// Note: Eventually we'll add a watch variant that takes a query, where the query can be // constructed using some sort of query builder API. // TODO(sadovsky): Support specifying resumeMarker. Note, watch-from-resumeMarker may be // problematic in that we don't track the governing ACL for changes in the watch log. // TODO(sadovsky): Make sure cancellations are surfaced as such (or ignored). // TODO(alexfandrianto): Implement properly. AtomicBoolean is only a patch.