@Test public void partialResultSetFetchSizePropagatesAsDemand() { ConnectionFactoryOptions options = ConnectionFactoryOptions.builder() .option(PARTIAL_RESULT_SET_FETCH_SIZE, 4) .option(DRIVER, DRIVER_NAME) .option(PROJECT, "project-id") .option(INSTANCE, "an-instance") .option(DATABASE, "db") .option(GOOGLE_CREDENTIALS, mock(GoogleCredentials.class)) .build(); SpannerConnectionFactory connectionFactory = (SpannerConnectionFactory)this.spannerConnectionFactoryProvider.create(options); TestPublisher<PartialResultSet> partialResultSetPublisher = TestPublisher.create(); when(this.mockClient.executeStreamingSql( any(StatementExecutionContext.class), any(), any(), any())) .thenReturn(partialResultSetPublisher.flux()); Session session = Session.newBuilder().setName("session-name").build(); when(this.mockClient.createSession(any())) .thenReturn(Mono.just(session)); StepVerifier.create(connectionFactory.create() .flatMapMany(c -> c.createStatement("SELECT * from table").execute()) .flatMap(result -> result.map((row,meta) -> row.get(0, String.class))) ) .then(() -> { partialResultSetPublisher.assertMinRequested(1); partialResultSetPublisher.next(makeBook("Odyssey")); partialResultSetPublisher.assertMinRequested(3); }) .expectNext("Odyssey") .then(() -> { partialResultSetPublisher.next(makeBook("Illiad"), makeBook("Margites")); partialResultSetPublisher.assertMinRequested(4); }) .expectNext("Illiad") .expectNext("Margites") .then(() -> { partialResultSetPublisher.complete(); }) .verifyComplete(); }
@Override public ConnectionFactory create(ConnectionFactoryOptions connectionFactoryOptions) { SpannerConnectionConfiguration config = createConfiguration(connectionFactoryOptions); if (connectionFactoryOptions.hasOption(Option.valueOf("client-implementation")) && connectionFactoryOptions.getValue(Option.valueOf("client-implementation")) .equals("client-library")) { return new SpannerClientLibraryConnectionFactory(config); } else { if (this.client == null) { this.client = new GrpcClient(config.getCredentials()); } return new SpannerConnectionFactory(this.client, config); } }
SpannerConnectionFactoryProvider implements ConnectionFactoryProvider { @Override public ConnectionFactory create(ConnectionFactoryOptions connectionFactoryOptions) { SpannerConnectionConfiguration config = createConfiguration(connectionFactoryOptions); if (connectionFactoryOptions.hasOption(Option.valueOf("client-implementation")) && connectionFactoryOptions.getValue(Option.valueOf("client-implementation")) .equals("client-library")) { return new SpannerClientLibraryConnectionFactory(config); } else { if (this.client == null) { this.client = new GrpcClient(config.getCredentials()); } return new SpannerConnectionFactory(this.client, config); } } }
SpannerConnectionFactoryProvider implements ConnectionFactoryProvider { @Override public ConnectionFactory create(ConnectionFactoryOptions connectionFactoryOptions) { SpannerConnectionConfiguration config = createConfiguration(connectionFactoryOptions); if (connectionFactoryOptions.hasOption(Option.valueOf("client-implementation")) && connectionFactoryOptions.getValue(Option.valueOf("client-implementation")) .equals("client-library")) { return new SpannerClientLibraryConnectionFactory(config); } else { if (this.client == null) { this.client = new GrpcClient(config.getCredentials()); } return new SpannerConnectionFactory(this.client, config); } }  }
SpannerConnectionFactoryProvider implements ConnectionFactoryProvider { @Override public ConnectionFactory create(ConnectionFactoryOptions connectionFactoryOptions) { SpannerConnectionConfiguration config = createConfiguration(connectionFactoryOptions); if (connectionFactoryOptions.hasOption(Option.valueOf("client-implementation")) && connectionFactoryOptions.getValue(Option.valueOf("client-implementation")) .equals("client-library")) { return new SpannerClientLibraryConnectionFactory(config); } else { if (this.client == null) { this.client = new GrpcClient(config.getCredentials()); } return new SpannerConnectionFactory(this.client, config); } }  @Override ConnectionFactory create(ConnectionFactoryOptions connectionFactoryOptions); @Override boolean supports(ConnectionFactoryOptions connectionFactoryOptions); @Override String getDriver(); }
SpannerConnectionFactoryProvider implements ConnectionFactoryProvider { @Override public ConnectionFactory create(ConnectionFactoryOptions connectionFactoryOptions) { SpannerConnectionConfiguration config = createConfiguration(connectionFactoryOptions); if (connectionFactoryOptions.hasOption(Option.valueOf("client-implementation")) && connectionFactoryOptions.getValue(Option.valueOf("client-implementation")) .equals("client-library")) { return new SpannerClientLibraryConnectionFactory(config); } else { if (this.client == null) { this.client = new GrpcClient(config.getCredentials()); } return new SpannerConnectionFactory(this.client, config); } }  @Override ConnectionFactory create(ConnectionFactoryOptions connectionFactoryOptions); @Override boolean supports(ConnectionFactoryOptions connectionFactoryOptions); @Override String getDriver(); static final String DRIVER_NAME; static final Option<String> PROJECT; static final Option<String> INSTANCE; static final Option<String> URL; static final Option<Integer> THREAD_POOL_SIZE; static final Option<Integer> PARTIAL_RESULT_SET_FETCH_SIZE; static final Option<Duration> DDL_OPERATION_TIMEOUT; static final Option<Duration> DDL_OPERATION_POLL_INTERVAL; static final Option<GoogleCredentials> GOOGLE_CREDENTIALS; }