@Test public void testAdd() throws Exception { dataClient = BigtableDataClient.create(settingsBuilder.build()); BulkReadWrapper bulkReadWrapper = new BulkReadVeneerApi(dataClient, TABLE_ID); ApiFuture<Result> resultFuture1_1 = bulkReadWrapper.add(ByteString.copyFromUtf8("one"), null); ApiFuture<Result> resultFuture1_2 = bulkReadWrapper.add(ByteString.copyFromUtf8("two"), null); assertFalse(resultFuture1_1.isDone()); Filters.Filter filter = Filters.FILTERS.key().regex("cf"); ApiFuture<Result> secBatchResult1 = bulkReadWrapper.add(ByteString.copyFromUtf8("1"), filter); ApiFuture<Result> secBatchResult2 = bulkReadWrapper.add(ByteString.copyFromUtf8("2"), filter); bulkReadWrapper.sendOutstanding(); resultFuture1_1.get(); resultFuture1_2.get(); secBatchResult1.get(); secBatchResult2.get(); assertTrue(resultFuture1_1.isDone()); assertTrue(resultFuture1_2.isDone()); assertTrue(secBatchResult1.isDone()); assertTrue(secBatchResult2.isDone()); assertEquals(2, fakeDataService.getReadRowsBatchCount()); }
@Override public ApiFuture<Result> add(ByteString rowKey, @Nullable Filters.Filter filter) { cleanupBarrier.incrementAndGet(); ApiFuture<Row> rowFuture = getOrCreateBatcher(filter).add(rowKey); rowFuture.addListener( new Runnable() { @Override public void run() { notifyArrival(); } }, CLEANUP_EXECUTOR); return ApiFutures.transform( rowFuture, new ApiFunction<Row, Result>() { @Override public Result apply(Row row) { return Adapters.ROW_ADAPTER.adaptResponse(row); } }, MoreExecutors.directExecutor()); }
BulkReadVeneerApi implements BulkReadWrapper { @Override public ApiFuture<Result> add(ByteString rowKey, @Nullable Filters.Filter filter) { cleanupBarrier.incrementAndGet(); ApiFuture<Row> rowFuture = getOrCreateBatcher(filter).add(rowKey); rowFuture.addListener( new Runnable() { @Override public void run() { notifyArrival(); } }, CLEANUP_EXECUTOR); return ApiFutures.transform( rowFuture, new ApiFunction<Row, Result>() { @Override public Result apply(Row row) { return Adapters.ROW_ADAPTER.adaptResponse(row); } }, MoreExecutors.directExecutor()); } }
BulkReadVeneerApi implements BulkReadWrapper { @Override public ApiFuture<Result> add(ByteString rowKey, @Nullable Filters.Filter filter) { cleanupBarrier.incrementAndGet(); ApiFuture<Row> rowFuture = getOrCreateBatcher(filter).add(rowKey); rowFuture.addListener( new Runnable() { @Override public void run() { notifyArrival(); } }, CLEANUP_EXECUTOR); return ApiFutures.transform( rowFuture, new ApiFunction<Row, Result>() { @Override public Result apply(Row row) { return Adapters.ROW_ADAPTER.adaptResponse(row); } }, MoreExecutors.directExecutor()); } BulkReadVeneerApi(BigtableDataClient client, String tableId); }
BulkReadVeneerApi implements BulkReadWrapper { @Override public ApiFuture<Result> add(ByteString rowKey, @Nullable Filters.Filter filter) { cleanupBarrier.incrementAndGet(); ApiFuture<Row> rowFuture = getOrCreateBatcher(filter).add(rowKey); rowFuture.addListener( new Runnable() { @Override public void run() { notifyArrival(); } }, CLEANUP_EXECUTOR); return ApiFutures.transform( rowFuture, new ApiFunction<Row, Result>() { @Override public Result apply(Row row) { return Adapters.ROW_ADAPTER.adaptResponse(row); } }, MoreExecutors.directExecutor()); } BulkReadVeneerApi(BigtableDataClient client, String tableId); @Override ApiFuture<Result> add(ByteString rowKey, @Nullable Filters.Filter filter); @Override void sendOutstanding(); }
BulkReadVeneerApi implements BulkReadWrapper { @Override public ApiFuture<Result> add(ByteString rowKey, @Nullable Filters.Filter filter) { cleanupBarrier.incrementAndGet(); ApiFuture<Row> rowFuture = getOrCreateBatcher(filter).add(rowKey); rowFuture.addListener( new Runnable() { @Override public void run() { notifyArrival(); } }, CLEANUP_EXECUTOR); return ApiFutures.transform( rowFuture, new ApiFunction<Row, Result>() { @Override public Result apply(Row row) { return Adapters.ROW_ADAPTER.adaptResponse(row); } }, MoreExecutors.directExecutor()); } BulkReadVeneerApi(BigtableDataClient client, String tableId); @Override ApiFuture<Result> add(ByteString rowKey, @Nullable Filters.Filter filter); @Override void sendOutstanding();  }