DataClientClassicApi implements DataClientWrapper { @Override public ApiFuture<List<Result>> readRowsAsync(Query request) { return ApiFutures.transform( ApiFutureUtil.adapt(delegate.readFlatRowsAsync(request.toProto(requestContext))), new ApiFunction<List<FlatRow>, List<Result>>() { @Override public List<Result> apply(List<FlatRow> rows) { ImmutableList.Builder<Result> resultList = ImmutableList.builder(); for (FlatRow flatRow : rows) { resultList.add(Adapters.FLAT_ROW_ADAPTER.adaptResponse(flatRow)); } return resultList.build(); } }, MoreExecutors.directExecutor()); } DataClientClassicApi(BigtableSession session, RequestContext requestContext); @Override BulkMutationWrapper createBulkMutation(String tableId); @Override BulkReadWrapper createBulkRead(String tableId); @Override ApiFuture<Void> mutateRowAsync(RowMutation rowMutation); @Override ApiFuture<Result> readModifyWriteRowAsync(ReadModifyWriteRow readModifyWriteRow); @Override ApiFuture<Boolean> checkAndMutateRowAsync(ConditionalRowMutation conditionalRowMutation); @Override ApiFuture<List<KeyOffset>> sampleRowKeysAsync(String tableId); @Override ApiFuture<Result> readRowAsync(String tableId, ByteString rowKey, Filters.Filter filter); @Override org.apache.hadoop.hbase.client.ResultScanner readRows(Query request); @Override ApiFuture<List<Result>> readRowsAsync(Query request); @Override void readRowsAsync(Query request, final StreamObserver<Result> observer); @Override void close();  }
@Test public void testReadRowsAsync() throws Exception { Query query = Query.create(TABLE_ID); FlatRow anotherFlatRow = FlatRow.newBuilder().withRowKey(ROW_KEY).build(); List<FlatRow> listFlatRows = ImmutableList.of(SAMPLE_FLAT_ROW, anotherFlatRow); when(delegate.readFlatRowsAsync(query.toProto(REQUEST_CONTEXT))) .thenReturn(Futures.immediateFuture(listFlatRows)); List<Result> actualResult = dataClientWrapper.readRowsAsync(query).get(); assertEquals(listFlatRows.size(), actualResult.size()); assertArrayEquals( FLAT_ROW_ADAPTER.adaptResponse(SAMPLE_FLAT_ROW).rawCells(), actualResult.get(0).rawCells()); assertArrayEquals( FLAT_ROW_ADAPTER.adaptResponse(anotherFlatRow).rawCells(), actualResult.get(1).rawCells()); verify(delegate).readFlatRowsAsync(query.toProto(REQUEST_CONTEXT)); }
@Test public void testReadRowsAsyncWithStreamOb() { final Exception readException = new Exception(); Query request = Query.create(TABLE_ID).rowKey(ROW_KEY); StreamObserver<Result> resultStreamOb = new StreamObserver<Result>() { @Override public void onNext(Result result) { assertArrayEquals( FLAT_ROW_ADAPTER.adaptResponse(SAMPLE_FLAT_ROW).rawCells(), result.rawCells()); } @Override public void onError(Throwable throwable) { assertEquals(readException, throwable); } @Override public void onCompleted() {} }; when(delegate.readFlatRows( Mockito.<ReadRowsRequest>any(), Mockito.<StreamObserver<FlatRow>>any())) .thenAnswer( new Answer<ScanHandler>() { @Override public ScanHandler answer(InvocationOnMock invocationOnMock) { StreamObserver<FlatRow> sb = invocationOnMock.getArgument(1); sb.onNext(SAMPLE_FLAT_ROW); sb.onError(readException); sb.onCompleted(); return new ScanHandler() { @Override public void cancel() {} }; } }); dataClientWrapper.readRowsAsync(request, resultStreamOb); verify(delegate) .readFlatRows(Mockito.<ReadRowsRequest>any(), Mockito.<StreamObserver<FlatRow>>any()); }