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

DATAREDIS-999 - Use UsingWhen to apply functions to reactive Redis connections.

Browse Source 
We now use Flux.usingWhen() to apply functions on reactive Redis connections and to release connections calling non-blocking methods.

Original Pull Request: #461
This commit is contained in:
Mark Paluch
2019-06-28 11:57:11 +02:00
committed by Christoph Strobl
parent e9b5e6ee8e
commit dfa366d8f7
3 changed files with 90 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveRedisConnection.java
View File

@@ -32,6 +32,7 @@ import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
import org.reactivestreams.Publisher;
import org.springframework.dao.DataAccessException;
import org.springframework.dao.InvalidDataAccessResourceUsageException;
import org.springframework.data.redis.connection.*;
@@ -237,7 +238,7 @@ class LettuceReactiveRedisConnection implements ReactiveRedisConnection {
* @see org.springframework.data.redis.connection.ReactiveRedisConnection#closeLater()
*/
public Mono<Void> closeLater() {
return Mono.fromRunnable(dedicatedConnection::close);
return Flux.mergeDelayError(2, dedicatedConnection.close(), pubSubConnection.close()).then();
}
protected Mono<? extends StatefulConnection<ByteBuffer, ByteBuffer>> getConnection() {
@@ -363,7 +364,7 @@ class LettuceReactiveRedisConnection implements ReactiveRedisConnection {
this.connectionProvider = connectionProvider;
Mono<StatefulConnection> defer = Mono
.defer(() -> Mono.fromCompletionStage(connectionProvider.getConnectionAsync(connectionType)));
.fromCompletionStage(() -> connectionProvider.getConnectionAsync(connectionType));
this.connectionPublisher = defer.doOnNext(it -> {
@@ -403,21 +404,25 @@ class LettuceReactiveRedisConnection implements ReactiveRedisConnection {
}
/**
* Close connection (blocking call).
* Close connection.
*/
void close() {
Mono<Void> close() {
if (state.compareAndSet(State.INITIAL, CLOSING) || state.compareAndSet(State.CONNECTION_REQUESTED, CLOSING)) {
return Mono.defer(() -> {
StatefulConnection<ByteBuffer, ByteBuffer> connection = this.connection;
this.connection = null;
if (state.compareAndSet(State.INITIAL, CLOSING) || state.compareAndSet(State.CONNECTION_REQUESTED, CLOSING)) {
StatefulConnection<ByteBuffer, ByteBuffer> connection = this.connection;
this.connection = null;
state.set(State.CLOSED);
if (connection != null) {
return Mono.fromCompletionStage(connectionProvider.releaseAsync(connection));
}
if (connection != null) {
LettuceFutureUtils.join(connectionProvider.releaseAsync(connection));
}
state.set(State.CLOSED);
}
return Mono.empty();
});
}
private static boolean isClosing(State state) {

46
src/main/java/org/springframework/data/redis/core/ReactiveRedisTemplate.java
View File

@@ -15,9 +15,6 @@
*/
package org.springframework.data.redis.core;
import org.springframework.data.redis.hash.HashMapper;
import org.springframework.data.redis.hash.ObjectHashMapper;
import org.springframework.lang.Nullable;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;
@@ -31,6 +28,7 @@ import java.util.List;
import java.util.stream.Collectors;
import org.reactivestreams.Publisher;
import org.springframework.data.redis.connection.DataType;
import org.springframework.data.redis.connection.ReactiveRedisConnection;
import org.springframework.data.redis.connection.ReactiveRedisConnection.CommandResponse;
@@ -39,11 +37,14 @@ import org.springframework.data.redis.connection.ReactiveSubscription.Message;
import org.springframework.data.redis.core.script.DefaultReactiveScriptExecutor;
import org.springframework.data.redis.core.script.ReactiveScriptExecutor;
import org.springframework.data.redis.core.script.RedisScript;
import org.springframework.data.redis.hash.HashMapper;
import org.springframework.data.redis.hash.ObjectHashMapper;
import org.springframework.data.redis.listener.ReactiveRedisMessageListenerContainer;
import org.springframework.data.redis.listener.Topic;
import org.springframework.data.redis.serializer.RedisElementReader;
import org.springframework.data.redis.serializer.RedisElementWriter;
import org.springframework.data.redis.serializer.RedisSerializationContext;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
@@ -131,22 +132,7 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
public <T> Flux<T> execute(ReactiveRedisCallback<T> action, boolean exposeConnection) {
Assert.notNull(action, "Callback object must not be null");
ReactiveRedisConnectionFactory factory = getConnectionFactory();
ReactiveRedisConnection conn = factory.getReactiveConnection();
try {
ReactiveRedisConnection connToUse = preProcessConnection(conn, false);
ReactiveRedisConnection connToExpose = (exposeConnection ? connToUse : createRedisConnectionProxy(connToUse));
Publisher<T> result = action.doInRedis(connToExpose);
return Flux.from(postProcessResult(result, connToUse, false)).doFinally(signalType -> conn.close());
} catch (RuntimeException e) {
conn.close();
throw e;
}
return Flux.from(doInConnection(action, exposeConnection));
}
/**
@@ -160,7 +146,7 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
Assert.notNull(callback, "ReactiveRedisCallback must not be null!");
return Flux.defer(() -> doInConnection(callback, exposeConnection));
return Flux.from(doInConnection(callback, exposeConnection));
}
/**
@@ -170,11 +156,11 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
* @param callback must not be {@literal null}
* @return a {@link Mono} wrapping the {@link ReactiveRedisCallback}.
*/
public <T> Mono<T> createMono(final ReactiveRedisCallback<T> callback) {
public <T> Mono<T> createMono(ReactiveRedisCallback<T> callback) {
Assert.notNull(callback, "ReactiveRedisCallback must not be null!");
return Mono.defer(() -> Mono.from(doInConnection(callback, exposeConnection)));
return Mono.from(doInConnection(callback, exposeConnection));
}
/**
@@ -190,15 +176,19 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
Assert.notNull(action, "Callback object must not be null");
ReactiveRedisConnectionFactory factory = getConnectionFactory();
ReactiveRedisConnection conn = factory.getReactiveConnection();
return Flux.usingWhen(Mono.fromSupplier(() -> {
ReactiveRedisConnection connToUse = preProcessConnection(conn, false);
ReactiveRedisConnectionFactory factory = getConnectionFactory();
ReactiveRedisConnection conn = factory.getReactiveConnection();
ReactiveRedisConnection connToUse = preProcessConnection(conn, false);
ReactiveRedisConnection connToExpose = (exposeConnection ? connToUse : createRedisConnectionProxy(connToUse));
Publisher<T> result = action.doInRedis(connToExpose);
return (exposeConnection ? connToUse : createRedisConnectionProxy(connToUse));
}), conn -> {
Publisher<T> result = action.doInRedis(conn);
return Flux.from(postProcessResult(result, connToUse, false)).doFinally(signal -> conn.close());
return postProcessResult(result, conn, false);
}, ReactiveRedisConnection::closeLater, ReactiveRedisConnection::closeLater);
}
/*