DATACASS-599 - Polishing.

Add executor to other Futures.transform(…) and Futures.addCallback(…) invocations.

Original pull request: #137.

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 <T> ListenableFuture<T> query(AsyncPreparedStatementCreator preparedStatementCreator,
-			@Nullable PreparedStatementBinder psb, ResultSetExtractor<T> resultSetExtractor)
-			throws DataAccessException {
+			@Nullable PreparedStatementBinder psb, ResultSetExtractor<T> 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<Void> 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 <T> ListenableFuture<List<T>> query(String cql, @Nullable PreparedStatementBinder psb,
-			RowMapper<T> rowMapper) throws DataAccessException {
+	public <T> ListenableFuture<List<T>> query(String cql, @Nullable PreparedStatementBinder psb, RowMapper<T> rowMapper)
+			throws DataAccessException {
 
 		return query(newAsyncPreparedStatementCreator(cql), psb, newResultSetExtractor(rowMapper));
 	}

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<T> implements ListenableFuture<T> {
 
 				settableFuture.setException(t);
 			}
-		});
+		}, ImmediateExecutor.INSTANCE);
 
 		return settableFuture;
 	}

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;
  * <p>
  *
  * @author Mark Paluch
+ * @author Mateusz Stefek
  * @since 2.0
  * @see Mono
  * @see ReactiveResultSet
@@ -156,12 +157,11 @@ public class DefaultBridgedReactiveSession implements ReactiveSession {
 
 				ListenableFuture<ResultSet> future = this.session.executeAsync(statement);
 
-				ListenableFuture<ReactiveResultSet> resultSetFuture =
-					Futures.transform(future, DefaultReactiveResultSet::new, MoreExecutors.directExecutor());
+				ListenableFuture<ReactiveResultSet> 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<PreparedStatement> 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<ResultSet> 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);
 			}
 		}

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 extends Statement> 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));
 	}
 }