BigtableDataGrpcClient implements BigtableDataClient { @Override public ResultScanner<FlatRow> readFlatRows(ReadRowsRequest request) { if (shouldOverrideAppProfile(request.getAppProfileId())) { request = request.toBuilder().setAppProfileId(clientDefaultAppProfileId).build(); } final ResponseQueueReader reader = new ResponseQueueReader(); RetryingReadRowsOperation operation = createReadRowsRetryListener(request, reader); operation.setResultObserver( new StreamObserver<ReadRowsResponse>() { @Override public void onNext(ReadRowsResponse value) { reader.addRequestResultMarker(); } @Override public void onError(Throwable t) {} @Override public void onCompleted() {} }); operation.getAsyncResult(); return new ResumingStreamingResultScanner(reader); } BigtableDataGrpcClient(
      Channel channel,
      ScheduledExecutorService retryExecutorService,
      BigtableOptions bigtableOptions); @Override void setCallOptionsFactory(CallOptionsFactory callOptionsFactory); @Override MutateRowResponse mutateRow(MutateRowRequest request); @Override ListenableFuture<MutateRowResponse> mutateRowAsync(MutateRowRequest request); @Override List<MutateRowsResponse> mutateRows(MutateRowsRequest request); @Override ListenableFuture<List<MutateRowsResponse>> mutateRowsAsync(MutateRowsRequest request); @Override CheckAndMutateRowResponse checkAndMutateRow(CheckAndMutateRowRequest request); @Override ListenableFuture<CheckAndMutateRowResponse> checkAndMutateRowAsync(
      CheckAndMutateRowRequest request); @Override ReadModifyWriteRowResponse readModifyWriteRow(ReadModifyWriteRowRequest request); @Override ListenableFuture<ReadModifyWriteRowResponse> readModifyWriteRowAsync(
      ReadModifyWriteRowRequest request); @Override List<SampleRowKeysResponse> sampleRowKeys(SampleRowKeysRequest request); @Override ListenableFuture<List<SampleRowKeysResponse>> sampleRowKeysAsync(
      SampleRowKeysRequest request); @Override ListenableFuture<List<Row>> readRowsAsync(ReadRowsRequest request); @Override ListenableFuture<List<FlatRow>> readFlatRowsAsync(ReadRowsRequest request); @Override List<FlatRow> readFlatRowsList(ReadRowsRequest request); @Override ResultScanner<Row> readRows(ReadRowsRequest request); @Override ResultScanner<FlatRow> readFlatRows(ReadRowsRequest request); @Override ScanHandler readFlatRows(ReadRowsRequest request, StreamObserver<FlatRow> observer); @VisibleForTesting
  static final Predicate<MutateRowRequest> IS_RETRYABLE_MUTATION; }
@Test public void testScanner() throws IOException { ReadRowsRequest.Builder requestBuilder = ReadRowsRequest.newBuilder().setTableName(TABLE_NAME); requestBuilder.getRowsBuilder().addRowKeys(ByteString.EMPTY); ResultScanner<FlatRow> scanner = defaultClient.readFlatRows(requestBuilder.build()); ArgumentCaptor<ClientCall.Listener> listenerCaptor = ArgumentCaptor.forClass(ClientCall.Listener.class); verify(mockClientCall, times(1)).start(listenerCaptor.capture(), any(Metadata.class)); Listener listener = listenerCaptor.getValue(); ByteString key1 = ByteString.copyFromUtf8("Key1"); ByteString key2 = ByteString.copyFromUtf8("Key2"); listener.onMessage(RetryingReadRowsOperationTest.buildResponse(key1)); listener.onMessage(RetryingReadRowsOperationTest.buildResponse(key2)); Assert.assertEquals(2, scanner.available()); Assert.assertEquals(key1, scanner.next().getRowKey()); listener.onClose(Status.OK, new Metadata()); Assert.assertEquals(key2, scanner.next().getRowKey()); Assert.assertNull(scanner.next()); }
@Test public void testScannerIdle() throws IOException { ReadRowsRequest.Builder requestBuilder = ReadRowsRequest.newBuilder().setTableName(TABLE_NAME); requestBuilder.getRowsBuilder().addRowKeys(ByteString.EMPTY); ResultScanner<FlatRow> scanner = defaultClient.readFlatRows(requestBuilder.build()); ArgumentCaptor<ClientCall.Listener> listenerCaptor = ArgumentCaptor.forClass(ClientCall.Listener.class); verify(mockClientCall, times(1)).start(listenerCaptor.capture(), any(Metadata.class)); Listener listener = listenerCaptor.getValue(); ByteString key1 = ByteString.copyFromUtf8("Key1"); ByteString key2 = ByteString.copyFromUtf8("Key2"); listener.onMessage(RetryingReadRowsOperationTest.buildResponse(key1)); listener.onMessage(RetryingReadRowsOperationTest.buildResponse(key2)); listener.onClose( Status.CANCELLED.withCause( new Watchdog.StreamWaitTimeoutException( Watchdog.State.IDLE, TimeUnit.MINUTES.toMillis(10))), new Metadata()); Assert.assertEquals(2, scanner.available()); Assert.assertEquals(key1, scanner.next().getRowKey()); Assert.assertEquals(key2, scanner.next().getRowKey()); ByteString key3 = ByteString.copyFromUtf8("Key3"); listener.onMessage(RetryingReadRowsOperationTest.buildResponse(key3)); Assert.assertEquals(key3, scanner.next().getRowKey()); verify(mochScheduler, times(1)).schedule(any(Runnable.class), anyLong(), any(TimeUnit.class)); }