SyncbaseBindingTermini { public static <T> Subscription bindWrite( final RxTable rxTable, final Observable<T> rxData, final String key, final Class<T> type, final T deleteValue, final Action1<Throwable> onError) { return rxData .switchMap(data -> rxTable.once().map(t -> new WriteData<>(t, data))) .onBackpressureLatest() .subscribe(new Subscriber<WriteData<T>>() { @Override public void onStart() { request(1); } @Override public void onNext(WriteData<T> w) { final ListenableFuture<Void> op = Objects.equals(w.data, deleteValue) ? w.t.delete(rxTable.getVContext(), key) : w.t.put(rxTable.getVContext(), key, w.data, type); FluentFutures.from(op) .onSuccess(r -> request(1)) .onFailure(this::onError); } @Override public void onCompleted() { } @Override public void onError(final Throwable t) { onError.call(t); } }); }  static Observable<SingleWatchEvent<T>> bindRead(
            final RxTable rxTable, final String key, final Class<T> type, final T defaultValue); static Subscription bindWrite(
            final RxTable rxTable, final Observable<T> rxData, final String key,
            final Class<T> type, final T deleteValue, final Action1<Throwable> onError); static TwoWayBinding<T> bind(
            final RxTable rxTable, final String key, final Class<T> type, final T defaultValue,
            final T deleteValue, final Action1<Throwable> onError);  }
@Test public void testSequencing() { final ReplaySubject<Table> mockTables = ReplaySubject.createWithSize(1); final RxTable rxTable = mock(RxTable.class); when(rxTable.getObservable()).thenReturn(mockTables); when(rxTable.once()).thenCallRealMethod(); final PublishSubject<Integer> rxData = PublishSubject.create(); final String key = "key"; SyncbaseBindingTermini.bindWrite(rxTable, rxData, key, Integer.class, null, this::catchAsync); rxData.onNext(1); rxData.onNext(2); final Table t1 = mock(Table.class); when(t1.put(any(), any(), any(), any())).thenReturn(Futures.immediateFuture(null)); mockTables.onNext(t1); verify(t1, never()).put(null, key, 1, Integer.class); verify(t1, timeout(BLOCKING_DELAY_MS) ).put(null, key, 2, Integer.class); rxData.onNext(3); verify(t1, timeout(BLOCKING_DELAY_MS)).put(null, key, 3, Integer.class); final Table t2 = mock(Table.class); final SettableFuture<Void> putComplete = SettableFuture.create(); when(t2.put(any(), any(), any(), any())).thenReturn(putComplete); mockTables.onNext(t2); rxData.onNext(4); verify(t2, timeout(BLOCKING_DELAY_MS)).put(null, key, 4, Integer.class); rxData.onNext(5); rxData.onNext(6); putComplete.set(null); verify(t2, never()).put(null, key, 5, Integer.class); verify(t2, timeout(BLOCKING_DELAY_MS)).put(null, key, 6, Integer.class); verifyNoMoreInteractions(t1); verifyNoMoreInteractions(t2); }