@Test public void testExecuteStreamingSql() throws IOException { ExecuteSqlRequest request = ExecuteSqlRequest.newBuilder().build(); String sql = "select book from library"; SpannerImplBase spannerSpy = doTest(new SpannerImplBase() { @Override public void executeStreamingSql(ExecuteSqlRequest request, StreamObserver<PartialResultSet> responseObserver) { responseObserver.onNext(PartialResultSet.newBuilder().build()); responseObserver.onCompleted(); } }, grpcClient -> grpcClient.executeStreamingSql(this.mockContext, sql).blockFirst()); ArgumentCaptor<ExecuteSqlRequest> requestCaptor = ArgumentCaptor .forClass(ExecuteSqlRequest.class); verify(spannerSpy).executeStreamingSql(requestCaptor.capture(), any()); assertEquals(sql, requestCaptor.getValue().getSql()); assertEquals(SESSION_NAME, requestCaptor.getValue().getSession()); assertEquals(TRANSACTION_ID, requestCaptor.getValue().getTransaction().getId()); }
@Override public Flux<PartialResultSet> executeStreamingSql( StatementExecutionContext ctx, String sql, Struct params, Map<String, Type> types) { return Flux.defer(() -> { Assert.requireNonNull(ctx.getSessionName(), "Session name must not be null"); ExecuteSqlRequest.Builder executeSqlRequest = buildSqlRequest(ctx, sql); if (params != null) { executeSqlRequest .setParams(params) .putAllParamTypes(types); } if (ctx.getTransactionId() != null) { executeSqlRequest.setTransaction( TransactionSelector.newBuilder().setId(ctx.getTransactionId()) .build()); executeSqlRequest.setSeqno(ctx.nextSeqNum()); } return ObservableReactiveUtil.streamingCall( obs -> this.spanner.executeStreamingSql(executeSqlRequest.build(), obs)); }); }
GrpcClient implements Client { @Override public Flux<PartialResultSet> executeStreamingSql( StatementExecutionContext ctx, String sql, Struct params, Map<String, Type> types) { return Flux.defer(() -> { Assert.requireNonNull(ctx.getSessionName(), "Session name must not be null"); ExecuteSqlRequest.Builder executeSqlRequest = buildSqlRequest(ctx, sql); if (params != null) { executeSqlRequest .setParams(params) .putAllParamTypes(types); } if (ctx.getTransactionId() != null) { executeSqlRequest.setTransaction( TransactionSelector.newBuilder().setId(ctx.getTransactionId()) .build()); executeSqlRequest.setSeqno(ctx.nextSeqNum()); } return ObservableReactiveUtil.streamingCall( obs -> this.spanner.executeStreamingSql(executeSqlRequest.build(), obs)); }); } }
GrpcClient implements Client { @Override public Flux<PartialResultSet> executeStreamingSql( StatementExecutionContext ctx, String sql, Struct params, Map<String, Type> types) { return Flux.defer(() -> { Assert.requireNonNull(ctx.getSessionName(), "Session name must not be null"); ExecuteSqlRequest.Builder executeSqlRequest = buildSqlRequest(ctx, sql); if (params != null) { executeSqlRequest .setParams(params) .putAllParamTypes(types); } if (ctx.getTransactionId() != null) { executeSqlRequest.setTransaction( TransactionSelector.newBuilder().setId(ctx.getTransactionId()) .build()); executeSqlRequest.setSeqno(ctx.nextSeqNum()); } return ObservableReactiveUtil.streamingCall( obs -> this.spanner.executeStreamingSql(executeSqlRequest.build(), obs)); }); } GrpcClient(GoogleCredentials credentials); @VisibleForTesting  GrpcClient(SpannerStub spanner, DatabaseAdminStub databaseAdmin, OperationsStub operations); }
GrpcClient implements Client { @Override public Flux<PartialResultSet> executeStreamingSql( StatementExecutionContext ctx, String sql, Struct params, Map<String, Type> types) { return Flux.defer(() -> { Assert.requireNonNull(ctx.getSessionName(), "Session name must not be null"); ExecuteSqlRequest.Builder executeSqlRequest = buildSqlRequest(ctx, sql); if (params != null) { executeSqlRequest .setParams(params) .putAllParamTypes(types); } if (ctx.getTransactionId() != null) { executeSqlRequest.setTransaction( TransactionSelector.newBuilder().setId(ctx.getTransactionId()) .build()); executeSqlRequest.setSeqno(ctx.nextSeqNum()); } return ObservableReactiveUtil.streamingCall( obs -> this.spanner.executeStreamingSql(executeSqlRequest.build(), obs)); }); } GrpcClient(GoogleCredentials credentials); @VisibleForTesting  GrpcClient(SpannerStub spanner, DatabaseAdminStub databaseAdmin, OperationsStub operations); @Override Mono<Transaction> beginTransaction(
      String sessionName, TransactionOptions transactionOptions); @Override Mono<CommitResponse> commitTransaction(String sessionName, Transaction transaction); @Override Mono<Void> rollbackTransaction(String sessionName, Transaction transaction); @Override Mono<Session> createSession(String databaseName); @Override Mono<Void> deleteSession(String sessionName); @Override Flux<ResultSet> executeBatchDml(
      StatementExecutionContext ctx,
      List<Statement> statements); @Override Flux<PartialResultSet> executeStreamingSql(
      StatementExecutionContext ctx,
      String sql,
      Struct params,
      Map<String, Type> types); @Override Mono<Operation> executeDdl(
      String fullyQualifiedDatabaseName,
      List<String> ddlStatements,
      Duration ddlOperationTimeout,
      Duration ddlPollInterval); @Override Mono<Void> close(); @Override Mono<Boolean> healthcheck(StatementExecutionContext ctx); @VisibleForTesting SpannerStub getSpanner(); }
GrpcClient implements Client { @Override public Flux<PartialResultSet> executeStreamingSql( StatementExecutionContext ctx, String sql, Struct params, Map<String, Type> types) { return Flux.defer(() -> { Assert.requireNonNull(ctx.getSessionName(), "Session name must not be null"); ExecuteSqlRequest.Builder executeSqlRequest = buildSqlRequest(ctx, sql); if (params != null) { executeSqlRequest .setParams(params) .putAllParamTypes(types); } if (ctx.getTransactionId() != null) { executeSqlRequest.setTransaction( TransactionSelector.newBuilder().setId(ctx.getTransactionId()) .build()); executeSqlRequest.setSeqno(ctx.nextSeqNum()); } return ObservableReactiveUtil.streamingCall( obs -> this.spanner.executeStreamingSql(executeSqlRequest.build(), obs)); }); } GrpcClient(GoogleCredentials credentials); @VisibleForTesting  GrpcClient(SpannerStub spanner, DatabaseAdminStub databaseAdmin, OperationsStub operations); @Override Mono<Transaction> beginTransaction(
      String sessionName, TransactionOptions transactionOptions); @Override Mono<CommitResponse> commitTransaction(String sessionName, Transaction transaction); @Override Mono<Void> rollbackTransaction(String sessionName, Transaction transaction); @Override Mono<Session> createSession(String databaseName); @Override Mono<Void> deleteSession(String sessionName); @Override Flux<ResultSet> executeBatchDml(
      StatementExecutionContext ctx,
      List<Statement> statements); @Override Flux<PartialResultSet> executeStreamingSql(
      StatementExecutionContext ctx,
      String sql,
      Struct params,
      Map<String, Type> types); @Override Mono<Operation> executeDdl(
      String fullyQualifiedDatabaseName,
      List<String> ddlStatements,
      Duration ddlOperationTimeout,
      Duration ddlPollInterval); @Override Mono<Void> close(); @Override Mono<Boolean> healthcheck(StatementExecutionContext ctx); @VisibleForTesting SpannerStub getSpanner();  }