DATAREDIS-629 - Adopt changed Mono and Flux error handling API.
Replace Flux/Mono.onErrorResumeWith(…) with Flux/Mono.onErrorMap(…) and turn translateException into a method returning a mapping function instead of a reactive type emitting the mapped exception. Replace otherwiseIfEmpty(…) with switchIfEmpty(…) and otherwiseReturn(…) with onErrorReturn(…).
This commit is contained in:
@@ -87,7 +87,7 @@ public class LettuceReactiveClusterKeyCommands extends LettuceReactiveKeyCommand
|
||||
}
|
||||
|
||||
Mono<Boolean> result = cmd.dump(command.getKey())
|
||||
.otherwiseIfEmpty(Mono.error(new RedisSystemException("Cannot rename key that does not exist",
|
||||
.switchIfEmpty(Mono.error(new RedisSystemException("Cannot rename key that does not exist",
|
||||
new RedisException("ERR no such key."))))
|
||||
.flatMap(value -> cmd.restore(command.getNewName(), 0, value).flatMap(res -> cmd.del(command.getKey())))
|
||||
.map(LettuceConverters.longToBooleanConverter()::convert);
|
||||
@@ -119,7 +119,7 @@ public class LettuceReactiveClusterKeyCommands extends LettuceReactiveKeyCommand
|
||||
}
|
||||
|
||||
return cmd.dump(command.getKey())
|
||||
.otherwiseIfEmpty(Mono.error(new RedisSystemException("Cannot rename key that does not exist",
|
||||
.switchIfEmpty(Mono.error(new RedisSystemException("Cannot rename key that does not exist",
|
||||
new RedisException("ERR no such key."))))
|
||||
.flatMap(value -> cmd.restore(command.getNewName(), 0, value).flatMap(res -> cmd.del(command.getKey())))
|
||||
.map(LettuceConverters.longToBooleanConverter()::convert);
|
||||
|
||||
@@ -105,7 +105,7 @@ public class LettuceReactiveHashCommands implements ReactiveHashCommands {
|
||||
if (command.getFields().size() == 1) {
|
||||
ByteBuffer key = command.getFields().iterator().next();
|
||||
result = cmd.hget(command.getKey(), key.duplicate()).map(value -> KeyValue.fromNullable(key, value))
|
||||
.map(Collections::singletonList).otherwiseReturn(Collections.emptyList());
|
||||
.map(Collections::singletonList).onErrorReturn(Collections.emptyList());
|
||||
} else {
|
||||
result = cmd.hmget(command.getKey(), command.getFields().stream().toArray(ByteBuffer[]::new)).collectList();
|
||||
}
|
||||
|
||||
@@ -125,7 +125,7 @@ public class LettuceReactiveRedisClusterConnection extends LettuceReactiveRedisC
|
||||
return Flux.error(e);
|
||||
}
|
||||
|
||||
return Flux.defer(() -> callback.doWithCommands(getCommands(node))).onErrorResumeWith(translateExeception());
|
||||
return Flux.defer(() -> callback.doWithCommands(getCommands(node))).onErrorMap(translateException());
|
||||
}
|
||||
|
||||
/* (non-Javadoc)
|
||||
|
||||
@@ -31,20 +31,12 @@ 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.ReactiveGeoCommands;
|
||||
import org.springframework.data.redis.connection.ReactiveHashCommands;
|
||||
import org.springframework.data.redis.connection.ReactiveHyperLogLogCommands;
|
||||
import org.springframework.data.redis.connection.ReactiveKeyCommands;
|
||||
import org.springframework.data.redis.connection.ReactiveListCommands;
|
||||
import org.springframework.data.redis.connection.ReactiveNumberCommands;
|
||||
import org.springframework.data.redis.connection.ReactiveRedisConnection;
|
||||
import org.springframework.data.redis.connection.ReactiveSetCommands;
|
||||
import org.springframework.data.redis.connection.ReactiveStringCommands;
|
||||
import org.springframework.data.redis.connection.ReactiveZSetCommands;
|
||||
import org.springframework.data.redis.connection.*;
|
||||
import org.springframework.util.Assert;
|
||||
|
||||
/**
|
||||
* @author Christoph Strobl
|
||||
* @author Mark Paluch
|
||||
* @since 2.0
|
||||
*/
|
||||
public class LettuceReactiveRedisConnection implements ReactiveRedisConnection {
|
||||
@@ -145,7 +137,7 @@ public class LettuceReactiveRedisConnection implements ReactiveRedisConnection {
|
||||
* @return
|
||||
*/
|
||||
public <T> Flux<T> execute(LettuceReactiveCallback<T> callback) {
|
||||
return Flux.defer(() -> callback.doWithCommands(getCommands())).onErrorResumeWith(translateExeception());
|
||||
return Flux.defer(() -> callback.doWithCommands(getCommands())).onErrorMap(translateException());
|
||||
}
|
||||
|
||||
/* (non-Javadoc)
|
||||
@@ -171,20 +163,18 @@ public class LettuceReactiveRedisConnection implements ReactiveRedisConnection {
|
||||
throw new RuntimeException("o.O unknown connection type " + connection);
|
||||
}
|
||||
|
||||
<T> Function<Throwable, Publisher<? extends T>> translateExeception() {
|
||||
<T> Function<Throwable, Throwable> translateException() {
|
||||
|
||||
return throwable -> {
|
||||
|
||||
if (throwable instanceof RuntimeException) {
|
||||
|
||||
DataAccessException convertedException = null;
|
||||
if (throwable instanceof RuntimeException) {
|
||||
convertedException = LettuceConverters.exceptionConverter().convert((RuntimeException) throwable);
|
||||
}
|
||||
return Flux.error(convertedException != null ? convertedException : throwable);
|
||||
DataAccessException convertedException = LettuceConverters.exceptionConverter()
|
||||
.convert((RuntimeException) throwable);
|
||||
return convertedException != null ? convertedException : throwable;
|
||||
}
|
||||
|
||||
return Flux.error(throwable);
|
||||
return throwable;
|
||||
};
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user