From a23d14277c783d04b2a3971cd7cdaaba85eccb31 Mon Sep 17 00:00:00 2001 From: Mark Paluch Date: Wed, 6 Feb 2019 13:18:08 +0100 Subject: [PATCH] DATACASS-599 - Polishing. MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add executor to other Futures.transform(…) and Futures.addCallback(…) invocations. Original pull request: #137. --- .../cassandra/core/cql/AsyncCqlTemplate.java | 17 +++++----- .../cql/GuavaListenableFutureAdapter.java | 4 ++- .../DefaultBridgedReactiveSession.java | 33 +++++++------------ ...efaultBridgedReactiveSessionUnitTests.java | 27 +++++---------- 4 files changed, 32 insertions(+), 49 deletions(-) diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/AsyncCqlTemplate.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/AsyncCqlTemplate.java index 19315a457..162e1914a 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/AsyncCqlTemplate.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/AsyncCqlTemplate.java @@ -15,6 +15,8 @@ */ package org.springframework.data.cassandra.core.cql; +import io.netty.util.concurrent.ImmediateExecutor; + import java.util.List; import java.util.Map; import java.util.concurrent.ExecutionException; @@ -506,8 +508,7 @@ public class AsyncCqlTemplate extends CassandraAccessor implements AsyncCqlOpera */ @Override public ListenableFuture query(AsyncPreparedStatementCreator preparedStatementCreator, - @Nullable PreparedStatementBinder psb, ResultSetExtractor resultSetExtractor) - throws DataAccessException { + @Nullable PreparedStatementBinder psb, ResultSetExtractor resultSetExtractor) throws DataAccessException { Assert.notNull(preparedStatementCreator, "AsyncPreparedStatementCreator must not be null"); Assert.notNull(resultSetExtractor, "ResultSetExtractor object must not be null"); @@ -553,7 +554,7 @@ public class AsyncCqlTemplate extends CassandraAccessor implements AsyncCqlOpera settableListenableFuture.setException(ex); } } - }), ex -> { + }, ImmediateExecutor.INSTANCE), ex -> { if (ex instanceof DriverException) { settableListenableFuture.setException(translateException(exceptionTranslator, (DriverException) ex)); } else { @@ -574,11 +575,9 @@ public class AsyncCqlTemplate extends CassandraAccessor implements AsyncCqlOpera */ @Override public ListenableFuture query(AsyncPreparedStatementCreator preparedStatementCreator, - @Nullable PreparedStatementBinder psb, RowCallbackHandler rowCallbackHandler) - throws DataAccessException { + @Nullable PreparedStatementBinder psb, RowCallbackHandler rowCallbackHandler) throws DataAccessException { - ListenableFuture results = query(preparedStatementCreator, psb, - newResultSetExtractor(rowCallbackHandler)); + ListenableFuture results = query(preparedStatementCreator, psb, newResultSetExtractor(rowCallbackHandler)); return new ExceptionTranslatingListenableFutureAdapter<>(new MappingListenableFutureAdapter<>(results, o -> null), getExceptionTranslator()); @@ -661,8 +660,8 @@ public class AsyncCqlTemplate extends CassandraAccessor implements AsyncCqlOpera * @see org.springframework.data.cassandra.core.cql.AsyncCqlOperations#query(java.lang.String, org.springframework.data.cassandra.core.cql.PreparedStatementBinder, org.springframework.data.cassandra.core.cql.RowMapper) */ @Override - public ListenableFuture> query(String cql, @Nullable PreparedStatementBinder psb, - RowMapper rowMapper) throws DataAccessException { + public ListenableFuture> query(String cql, @Nullable PreparedStatementBinder psb, RowMapper rowMapper) + throws DataAccessException { return query(newAsyncPreparedStatementCreator(cql), psb, newResultSetExtractor(rowMapper)); } diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/GuavaListenableFutureAdapter.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/GuavaListenableFutureAdapter.java index 8f9252f37..1f7c36850 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/GuavaListenableFutureAdapter.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/GuavaListenableFutureAdapter.java @@ -15,6 +15,8 @@ */ package org.springframework.data.cassandra.core.cql; +import io.netty.util.concurrent.ImmediateExecutor; + import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; @@ -88,7 +90,7 @@ public class GuavaListenableFutureAdapter implements ListenableFuture { settableFuture.setException(t); } - }); + }, ImmediateExecutor.INSTANCE); return settableFuture; } diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/session/DefaultBridgedReactiveSession.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/session/DefaultBridgedReactiveSession.java index 0277b9262..1eb12f763 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/session/DefaultBridgedReactiveSession.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/session/DefaultBridgedReactiveSession.java @@ -15,6 +15,7 @@ */ package org.springframework.data.cassandra.core.cql.session; +import io.netty.util.concurrent.ImmediateExecutor; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import reactor.core.publisher.MonoProcessor; @@ -34,7 +35,6 @@ import org.springframework.util.Assert; import com.datastax.driver.core.*; import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; -import com.google.common.util.concurrent.MoreExecutors; /** * Default implementation of a {@link ReactiveSession}. This implementation bridges asynchronous {@link Session} methods @@ -52,6 +52,7 @@ import com.google.common.util.concurrent.MoreExecutors; *

* * @author Mark Paluch + * @author Mateusz Stefek * @since 2.0 * @see Mono * @see ReactiveResultSet @@ -156,12 +157,11 @@ public class DefaultBridgedReactiveSession implements ReactiveSession { ListenableFuture future = this.session.executeAsync(statement); - ListenableFuture resultSetFuture = - Futures.transform(future, DefaultReactiveResultSet::new, MoreExecutors.directExecutor()); + ListenableFuture resultSetFuture = Futures.transform(future, DefaultReactiveResultSet::new, + ImmediateExecutor.INSTANCE); adaptFuture(resultSetFuture, sink); - } - catch (Exception cause) { + } catch (Exception cause) { sink.error(cause); } }); @@ -196,8 +196,7 @@ public class DefaultBridgedReactiveSession implements ReactiveSession { ListenableFuture resultSetFuture = this.session.prepareAsync(statement); adaptFuture(resultSetFuture, sink); - } - catch (Exception cause) { + } catch (Exception cause) { sink.error(cause); } }); @@ -224,11 +223,9 @@ public class DefaultBridgedReactiveSession implements ReactiveSession { if (future.isDone()) { try { sink.success(future.get()); - } - catch (ExecutionException cause) { + } catch (ExecutionException cause) { sink.error(cause.getCause()); - } - catch (Exception cause) { + } catch (Exception cause) { sink.error(cause); } } @@ -243,7 +240,6 @@ public class DefaultBridgedReactiveSession implements ReactiveSession { this.resultSet = resultSet; } - /* (non-Javadoc) * @see org.springframework.data.cassandra.ReactiveResultSet#rows() */ @@ -272,9 +268,7 @@ public class DefaultBridgedReactiveSession implements ReactiveSession { MonoProcessor processor = MonoProcessor.create(); - return rows - .doOnComplete(() -> fetchMore(it.fetchMoreResults(), processor)) - .concatWith(getRows(processor)); + return rows.doOnComplete(() -> fetchMore(it.fetchMoreResults(), processor)).concatWith(getRows(processor)); }); } @@ -294,17 +288,14 @@ public class DefaultBridgedReactiveSession implements ReactiveSession { try { sink.onNext(future.get()); sink.onComplete(); - } - catch (ExecutionException cause) { + } catch (ExecutionException cause) { sink.onError(cause.getCause()); - } - catch (Exception cause) { + } catch (Exception cause) { sink.onError(cause); } }, Runnable::run); - } - catch (Exception cause) { + } catch (Exception cause) { sink.onError(cause); } } diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/cql/DefaultBridgedReactiveSessionUnitTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/cql/DefaultBridgedReactiveSessionUnitTests.java index 0e0f5185b..0c69a1ed6 100644 --- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/cql/DefaultBridgedReactiveSessionUnitTests.java +++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/cql/DefaultBridgedReactiveSessionUnitTests.java @@ -176,10 +176,7 @@ public class DefaultBridgedReactiveSessionUnitTests { when(future.get()).thenReturn(resultSet); when(resultSet.isFullyFetched()).thenReturn(true); - reactiveSession.execute(new SimpleStatement("")) - .flatMapMany(ReactiveResultSet::rows) - .collectList() - .subscribe(); + reactiveSession.execute(new SimpleStatement("")).flatMapMany(ReactiveResultSet::rows).collectList().subscribe(); verify(rows, times(10)).next(); verify(resultSet, never()).fetchMoreResults(); @@ -259,30 +256,24 @@ public class DefaultBridgedReactiveSessionUnitTests { StepVerifier.create(flux, 0) // .then(() -> runnables.poll().run()) // complete the first future from executeAsync() - .thenRequest(9).expectNextCount(9) - .then(() -> { + .thenRequest(9).expectNextCount(9).then(() -> { // feed the 9 elements from the initial ResultSet verify(resultSet, never()).fetchMoreResults(); - }).thenRequest(1).expectNextCount(1) - .then(() -> { + }).thenRequest(1).expectNextCount(1).then(() -> { // initial ResultSet exhausted, fetch next chunk verify(resultSet).fetchMoreResults(); runnables.poll().run(); - }).thenRequest(1).expectNextCount(1) - .then(() -> { + }).thenRequest(1).expectNextCount(1).then(() -> { // first element from the second ResultSet received, no subsequent fetch assertThat(runnables).isEmpty(); - }).thenRequest(19).expectNextCount(9) - .then(() -> { + }).thenRequest(19).expectNextCount(9).then(() -> { // second ResultSet exhausted assertThat(runnables).hasSize(1); runnables.poll().run(); - }) - .thenRequest(10).expectNextCount(10) - .verifyComplete(); + }).thenRequest(10).expectNextCount(10).verifyComplete(); } @SuppressWarnings("unchecked") @@ -301,8 +292,8 @@ public class DefaultBridgedReactiveSessionUnitTests { @SuppressWarnings("all") private static T eq(T value) { - return ArgumentMatchers.argThat(argument -> argument instanceof Statement - ? value.toString().equals(argument.toString()) - : value.equals(argument)); + return ArgumentMatchers + .argThat(argument -> argument instanceof Statement ? value.toString().equals(argument.toString()) + : value.equals(argument)); } }