Spring/spring-data-cassandra
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

DATACASS-509 - Polishing.

Browse Source 
Replace Mono creation via Future and CompletableFuture with Mono sink creation. Using sink directly does not require creation of a CompletableFuture.
This commit is contained in:
Mark Paluch
2017-11-08 16:45:58 +01:00
committed by John Blum
parent 8ab816b919
commit 65b4cd3c8c
2 changed files with 39 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

70
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/session/DefaultBridgedReactiveSession.java
View File

@@ -18,11 +18,12 @@ package org.springframework.data.cassandra.core.cql.session;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.MonoProcessor;
import reactor.core.publisher.MonoSink;
import reactor.core.scheduler.Scheduler;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -31,6 +32,7 @@ import org.springframework.data.cassandra.ReactiveSession;
import org.springframework.util.Assert;
import com.datastax.driver.core.*;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
/**
@@ -128,30 +130,19 @@ public class DefaultBridgedReactiveSession implements ReactiveSession {
Assert.notNull(statement, "Statement must not be null");
return Mono.defer(() -> {
return Mono.create(sink -> {
try {
if (logger.isDebugEnabled()) {
logger.debug("Executing Statement [{}]", statement);
}
CompletableFuture<ReactiveResultSet> future = new CompletableFuture<>();
ResultSetFuture resultSetFuture = session.executeAsync(statement);
ListenableFuture<ResultSet> future = session.executeAsync(statement);
ListenableFuture<ReactiveResultSet> resultSetFuture = Futures.transform(future, DefaultReactiveResultSet::new);
resultSetFuture.addListener(() -> {
if (resultSetFuture.isDone()) {
try {
future.complete(new DefaultReactiveResultSet(resultSetFuture.getUninterruptibly()));
} catch (Exception e) {
future.completeExceptionally(e);
}
}
}, Runnable::run);
return Mono.fromFuture(future);
adaptFuture(resultSetFuture, sink);
} catch (Exception e) {
return Mono.error(e);
sink.error(e);
}
});
}
@@ -175,29 +166,18 @@ public class DefaultBridgedReactiveSession implements ReactiveSession {
Assert.notNull(statement, "Statement must not be null");
return Mono.defer(() -> {
return Mono.create(sink -> {
try {
if (logger.isDebugEnabled()) {
logger.debug("Preparing Statement [{}]", statement);
}
CompletableFuture<PreparedStatement> future = new CompletableFuture<>();
ListenableFuture<PreparedStatement> resultSetFuture = session.prepareAsync(statement);
resultSetFuture.addListener(() -> {
if (resultSetFuture.isDone()) {
try {
future.complete(resultSetFuture.get());
} catch (Exception e) {
future.completeExceptionally(e);
}
}
}, Runnable::run);
return Mono.fromFuture(future);
adaptFuture(resultSetFuture, sink);
} catch (Exception e) {
return Mono.error(e);
sink.error(e);
}
});
}
@@ -226,6 +206,28 @@ public class DefaultBridgedReactiveSession implements ReactiveSession {
return session.getCluster();
}
/**
* Adapt {@link ListenableFuture} signals (completion, error) and propagate these to {@link MonoSink}.
*
* @param future the originating future.
* @param sink the sink receiving signals.
*/
private static <T> void adaptFuture(ListenableFuture<T> future, MonoSink<T> sink) {
future.addListener(() -> {
if (future.isDone()) {
try {
sink.success(future.get());
} catch (ExecutionException e) {
sink.error(e.getCause());
} catch (Exception e) {
sink.error(e);
}
}
}, Runnable::run);
}
static class DefaultReactiveResultSet implements ReactiveResultSet {
private final ResultSet resultSet;
@@ -275,6 +277,8 @@ public class DefaultBridgedReactiveSession implements ReactiveSession {
sink.onNext(future.get());
sink.onComplete();
} catch (ExecutionException e) {
sink.onError(e.getCause());
} catch (Exception e) {
sink.onError(e);
}

8
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/cql/DefaultBridgedReactiveSessionUnitTests.java
View File

@@ -168,8 +168,7 @@ public class DefaultBridgedReactiveSessionUnitTests {
return null;
}).when(future).addListener(any(), any());
when(future.getUninterruptibly()).thenReturn(resultSet);
when(future.isDone()).thenReturn(true);
when(future.get()).thenReturn(resultSet);
when(resultSet.isFullyFetched()).thenReturn(true);
reactiveSession.execute(new SimpleStatement("")).flatMapMany(ReactiveResultSet::rows).collectList().subscribe();
@@ -199,8 +198,7 @@ public class DefaultBridgedReactiveSessionUnitTests {
return null;
}).when(future).addListener(any(), any());
when(future.getUninterruptibly()).thenReturn(resultSet);
when(future.isDone()).thenReturn(true);
when(future.get()).thenReturn(resultSet);
when(resultSet.isFullyFetched()).thenReturn(false, true);
when(resultSet.fetchMoreResults()).thenReturn(Futures.immediateFuture(emptyResultSet));
@@ -228,9 +226,7 @@ public class DefaultBridgedReactiveSessionUnitTests {
return null;
}).when(future).addListener(any(), any());
when(future.getUninterruptibly()).thenReturn(resultSet);
when(future.get()).thenReturn(resultSet);
when(future.isDone()).thenReturn(true);
when(resultSet.isFullyFetched()).thenReturn(false, false, true);
when(resultSet.fetchMoreResults()).thenReturn(future);