Add support for flush modes using FLUSHDB and FLUSHALL commands.

Closes #2187
Original pull request: #2190.
This commit is contained in:
Napster
2021-11-02 23:58:56 +01:00
committed by Mark Paluch
parent b895229f7c
commit 3191def956
24 changed files with 737 additions and 8 deletions

View File

@@ -75,6 +75,7 @@ import org.springframework.util.ObjectUtils;
* @author Andrey Shlykov
* @author dengliming
* @author ihaohong
* @author Dennis Neufeld
*/
public class DefaultStringRedisConnection implements StringRedisConnection, DecoratedRedisConnection {
@@ -313,11 +314,21 @@ public class DefaultStringRedisConnection implements StringRedisConnection, Deco
delegate.flushAll();
}
@Override
public void flushAll(FlushOption option) {
delegate.flushAll(option);
}
@Override
public void flushDb() {
delegate.flushDb();
}
@Override
public void flushDb(FlushOption option) {
delegate.flushDb(option);
}
@Override
public byte[] get(byte[] key) {
return convertAndReturn(delegate.get(key), Converters.identityConverter());

View File

@@ -27,6 +27,7 @@ import org.springframework.util.Assert;
/**
* @author Christoph Strobl
* @author Mark Paluch
* @author Dennis Neufeld
* @since 2.0
*/
public interface DefaultedRedisClusterConnection extends RedisClusterConnection, DefaultedRedisConnection {
@@ -73,6 +74,13 @@ public interface DefaultedRedisClusterConnection extends RedisClusterConnection,
serverCommands().flushDb(node);
}
/** @deprecated in favor of {@link RedisConnection#serverCommands()}. */
@Override
@Deprecated
default void flushDb(RedisClusterNode node, FlushOption option) {
serverCommands().flushDb(node, option);
}
/** @deprecated in favor of {@link RedisConnection#serverCommands()}. */
@Override
@Deprecated
@@ -80,6 +88,13 @@ public interface DefaultedRedisClusterConnection extends RedisClusterConnection,
serverCommands().flushAll(node);
}
/** @deprecated in favor of {@link RedisConnection#serverCommands()}. */
@Override
@Deprecated
default void flushAll(RedisClusterNode node, FlushOption option) {
serverCommands().flushAll(node, option);
}
/** @deprecated in favor of {@link RedisConnection#serverCommands()}. */
@Override
@Deprecated

View File

@@ -60,6 +60,7 @@ import org.springframework.lang.Nullable;
* @author Andrey Shlykov
* @author dengliming
* @author ihaohong
* @author Dennis Neufeld
* @since 2.0
*/
public interface DefaultedRedisConnection extends RedisConnection {
@@ -1632,6 +1633,13 @@ public interface DefaultedRedisConnection extends RedisConnection {
serverCommands().flushDb();
}
/** @deprecated in favor of {@link RedisConnection#serverCommands()}. */
@Override
@Deprecated
default void flushDb(FlushOption option) {
serverCommands().flushDb(option);
}
/** @deprecated in favor of {@link RedisConnection#serverCommands()}. */
@Override
@Deprecated
@@ -1639,6 +1647,13 @@ public interface DefaultedRedisConnection extends RedisConnection {
serverCommands().flushAll();
}
/** @deprecated in favor of {@link RedisConnection#serverCommands()}. */
@Override
@Deprecated
default void flushAll(FlushOption option) {
serverCommands().flushAll(option);
}
/** @deprecated in favor of {@link RedisConnection#serverCommands()}. */
@Override
@Deprecated

View File

@@ -20,6 +20,7 @@ import reactor.core.publisher.Mono;
import java.util.Properties;
import org.springframework.data.redis.connection.RedisServerCommands.FlushOption;
import org.springframework.data.redis.core.types.RedisClientInfo;
/**
@@ -27,6 +28,7 @@ import org.springframework.data.redis.core.types.RedisClientInfo;
*
* @author Mark Paluch
* @author Christoph Strobl
* @author Dennis Neufeld
* @since 2.0
*/
public interface ReactiveClusterServerCommands extends ReactiveServerCommands {
@@ -91,6 +93,17 @@ public interface ReactiveClusterServerCommands extends ReactiveServerCommands {
*/
Mono<String> flushDb(RedisClusterNode node);
/**
* Delete all keys of the currently selected database using the specified flush option.
*
* @param node must not be {@literal null}. {@link Mono} indicating command completion.
* @param option
* @throws IllegalArgumentException when {@code node} is {@literal null}.
* @see RedisServerCommands#flushDb(FlushOption)
* @since 2.6
*/
Mono<String> flushDb(RedisClusterNode node, FlushOption option);
/**
* Delete all <b>all keys</b> from <b>all databases</b>.
*
@@ -101,6 +114,18 @@ public interface ReactiveClusterServerCommands extends ReactiveServerCommands {
*/
Mono<String> flushAll(RedisClusterNode node);
/**
* Delete all <b>all keys</b> from <b>all databases</b> using the specified flush option.
*
* @param node must not be {@literal null}.
* @param option
* @return {@link Mono} indicating command completion.
* @throws IllegalArgumentException when {@code node} is {@literal null}.
* @see RedisServerCommands#flushAll(FlushOption)
* @since 2.6
*/
Mono<String> flushAll(RedisClusterNode node, FlushOption option);
/**
* Load {@literal default} server information like
* <ul>

View File

@@ -21,6 +21,7 @@ import reactor.core.publisher.Mono;
import java.util.Properties;
import java.util.concurrent.TimeUnit;
import org.springframework.data.redis.connection.RedisServerCommands.FlushOption;
import org.springframework.data.redis.core.types.RedisClientInfo;
/**
@@ -28,6 +29,7 @@ import org.springframework.data.redis.core.types.RedisClientInfo;
*
* @author Mark Paluch
* @author Christoph Strobl
* @author Dennis Neufeld
* @since 2.0
*/
public interface ReactiveServerCommands {
@@ -81,6 +83,16 @@ public interface ReactiveServerCommands {
*/
Mono<String> flushDb();
/**
* Delete all keys of the currently selected database using the specified flush option.
*
* @param option
* @return {@link Mono} indicating command completion.
* @see <a href="https://redis.io/commands/flushdb">Redis Documentation: FLUSHDB</a>
* @since 2.6
*/
Mono<String> flushDb(FlushOption option);
/**
* Delete all <b>all keys</b> from <b>all databases</b>.
*
@@ -89,6 +101,16 @@ public interface ReactiveServerCommands {
*/
Mono<String> flushAll();
/**
* Delete all <b>all keys</b> from <b>all databases</b> using the specified flush option.
*
* @param option
* @return {@link Mono} indicating command completion.
* @see <a href="https://redis.io/commands/flushall">Redis Documentation: FLUSHALL</a>
* @since 2.6
*/
Mono<String> flushAll(FlushOption option);
/**
* Load {@literal default} server information like
* <ul>

View File

@@ -23,6 +23,7 @@ import org.springframework.data.redis.core.types.RedisClientInfo;
/**
* @author Mark Paluch
* @author Dennis Neufeld
* @since 2.0
*/
public interface RedisClusterServerCommands extends RedisServerCommands {
@@ -65,12 +66,28 @@ public interface RedisClusterServerCommands extends RedisServerCommands {
*/
void flushDb(RedisClusterNode node);
/**
* @param node must not be {@literal null}.
* @param option
* @see RedisServerCommands#flushDb(FlushOption)
* @since 2.6
*/
void flushDb(RedisClusterNode node, FlushOption option);
/**
* @param node must not be {@literal null}.
* @see RedisServerCommands#flushAll()
*/
void flushAll(RedisClusterNode node);
/**
* @param node must not be {@literal null}.
* @param option
* @see RedisServerCommands#flushAll(FlushOption)
* @since 2.6
*/
void flushAll(RedisClusterNode node, FlushOption option);
/**
* @param node must not be {@literal null}.
* @return

View File

@@ -29,6 +29,7 @@ import org.springframework.lang.Nullable;
* @author Christoph Strobl
* @author Thomas Darimont
* @author Mark Paluch
* @author Dennis Neufeld
*/
public interface RedisServerCommands {
@@ -43,6 +44,13 @@ public interface RedisServerCommands {
COPY, REPLACE
}
/**
* @since 2.6
*/
enum FlushOption {
SYNC, ASYNC
}
/**
* Start an {@literal Append Only File} rewrite process on server.
*
@@ -101,6 +109,15 @@ public interface RedisServerCommands {
*/
void flushDb();
/**
* Delete all keys of the currently selected database using the specified flush option.
*
* @param option
* @see <a href="https://redis.io/commands/flushdb">Redis Documentation: FLUSHDB</a>
* @since 2.6
*/
void flushDb(FlushOption option);
/**
* Delete all <b>all keys</b> from <b>all databases</b>.
*
@@ -108,6 +125,15 @@ public interface RedisServerCommands {
*/
void flushAll();
/**
* Delete all <b>all keys</b> from <b>all databases</b> using the specified flush option.
*
* @param option
* @see <a href="https://redis.io/commands/flushall">Redis Documentation: FLUSHALL</a>
* @since 2.6
*/
void flushAll(FlushOption option);
/**
* Load {@literal default} server information like
* <ul>

View File

@@ -17,6 +17,7 @@ package org.springframework.data.redis.connection.jedis;
import redis.clients.jedis.BinaryJedis;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.args.FlushMode;
import java.util.ArrayList;
import java.util.Collection;
@@ -41,6 +42,7 @@ import org.springframework.util.CollectionUtils;
/**
* @author Mark Paluch
* @author Dennis Neufeld
* @since 2.0
*/
class JedisClusterServerCommands implements RedisClusterServerCommands {
@@ -127,22 +129,43 @@ class JedisClusterServerCommands implements RedisClusterServerCommands {
executeCommandOnAllNodes(BinaryJedis::flushDB);
}
@Override
public void flushDb(FlushOption option) {
executeCommandOnAllNodes(it -> it.flushDB(toFlushMode(option)));
}
@Override
public void flushDb(RedisClusterNode node) {
executeCommandOnSingleNode(BinaryJedis::flushDB, node);
}
@Override
public void flushDb(RedisClusterNode node, FlushOption option) {
executeCommandOnSingleNode(it -> it.flushDB(toFlushMode(option)), node);
}
@Override
public void flushAll() {
connection.getClusterCommandExecutor()
.executeCommandOnAllNodes((JedisClusterCommandCallback<String>) BinaryJedis::flushAll);
}
@Override
public void flushAll(FlushOption option) {
connection.getClusterCommandExecutor()
.executeCommandOnAllNodes((JedisClusterCommandCallback<String>) it -> it.flushAll(toFlushMode(option)));
}
@Override
public void flushAll(RedisClusterNode node) {
executeCommandOnSingleNode(BinaryJedis::flushAll, node);
}
@Override
public void flushAll(RedisClusterNode node, FlushOption option) {
executeCommandOnSingleNode(it -> it.flushAll(toFlushMode(option)), node);
}
@Override
public Properties info() {
@@ -395,4 +418,19 @@ class JedisClusterServerCommands implements RedisClusterServerCommands {
private <T> MultiNodeResult<T> executeCommandOnAllNodes(JedisClusterCommandCallback<T> cmd) {
return connection.getClusterCommandExecutor().executeCommandOnAllNodes(cmd);
}
static FlushMode toFlushMode(@Nullable FlushOption option) {
if (option == null) {
return FlushMode.SYNC;
}
switch (option) {
case ASYNC:
return FlushMode.ASYNC;
case SYNC:
return FlushMode.SYNC;
}
throw new UnsupportedOperationException("Flush option " + option + " is not implemented.");
}
}

View File

@@ -18,6 +18,7 @@ package org.springframework.data.redis.connection.jedis;
import redis.clients.jedis.BinaryJedis;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.MultiKeyPipelineBase;
import redis.clients.jedis.args.FlushMode;
import redis.clients.jedis.args.SaveMode;
import java.util.List;
@@ -33,6 +34,7 @@ import org.springframework.util.Assert;
/**
* @author Mark Paluch
* @author Dennis Neufeld
* @since 2.0
*/
class JedisServerCommands implements RedisServerCommands {
@@ -75,11 +77,25 @@ class JedisServerCommands implements RedisServerCommands {
connection.invokeStatus().just(BinaryJedis::flushDB, MultiKeyPipelineBase::flushDB);
}
@Override
public void flushDb(FlushOption option) {
FlushMode flushMode = JedisClusterServerCommands.toFlushMode(option);
connection.invokeStatus().just(it -> it.flushDB(flushMode), it -> it.flushDB(flushMode));
}
@Override
public void flushAll() {
connection.invokeStatus().just(BinaryJedis::flushAll, MultiKeyPipelineBase::flushAll);
}
@Override
public void flushAll(FlushOption option) {
FlushMode flushMode = JedisClusterServerCommands.toFlushMode(option);
connection.invokeStatus().just(it -> it.flushAll(flushMode), it -> it.flushAll(flushMode));
}
@Override
public Properties info() {
return connection.invoke().from(BinaryJedis::info, MultiKeyPipelineBase::info).get(JedisConverters::toProperties);

View File

@@ -38,6 +38,7 @@ import org.springframework.util.CollectionUtils;
/**
* @author Mark Paluch
* @author Dennis Neufeld
* @since 2.0
*/
class LettuceClusterServerCommands extends LettuceServerCommands implements RedisClusterServerCommands {
@@ -96,21 +97,41 @@ class LettuceClusterServerCommands extends LettuceServerCommands implements Redi
executeCommandOnAllNodes(RedisServerCommands::flushdb);
}
@Override
public void flushDb(FlushOption option) {
executeCommandOnAllNodes(it -> it.flushdb(toFlushMode(option)));
}
@Override
public void flushDb(RedisClusterNode node) {
executeCommandOnSingleNode(RedisServerCommands::flushdb, node);
}
@Override
public void flushDb(RedisClusterNode node, FlushOption option) {
executeCommandOnSingleNode(it -> it.flushdb(toFlushMode(option)), node);
}
@Override
public void flushAll() {
executeCommandOnAllNodes(RedisServerCommands::flushall);
}
@Override
public void flushAll(FlushOption option) {
executeCommandOnAllNodes(it -> it.flushall(toFlushMode(option)));
}
@Override
public void flushAll(RedisClusterNode node) {
executeCommandOnSingleNode(RedisServerCommands::flushall, node);
}
@Override
public void flushAll(RedisClusterNode node, FlushOption option) {
executeCommandOnSingleNode(it -> it.flushall(toFlushMode(option)), node);
}
@Override
public Properties info(RedisClusterNode node) {
return LettuceConverters.toProperties(executeCommandOnSingleNode(RedisServerCommands::info, node).getValue());

View File

@@ -41,6 +41,7 @@ import org.reactivestreams.Publisher;
import org.springframework.data.redis.connection.ClusterTopologyProvider;
import org.springframework.data.redis.connection.ReactiveClusterServerCommands;
import org.springframework.data.redis.connection.RedisClusterNode;
import org.springframework.data.redis.connection.RedisServerCommands.FlushOption;
import org.springframework.data.redis.core.types.RedisClientInfo;
import org.springframework.data.redis.util.ByteUtils;
import org.springframework.util.Assert;
@@ -50,6 +51,7 @@ import org.springframework.util.Assert;
*
* @author Mark Paluch
* @author Christoph Strobl
* @author Dennis Neufeld
* @since 2.0
*/
class LettuceReactiveClusterServerCommands extends LettuceReactiveServerCommands
@@ -105,11 +107,21 @@ class LettuceReactiveClusterServerCommands extends LettuceReactiveServerCommands
return connection.execute(node, RedisServerReactiveCommands::flushdb).next();
}
@Override
public Mono<String> flushDb(RedisClusterNode node, FlushOption option) {
return connection.execute(node, it -> it.flushdb(LettuceServerCommands.toFlushMode(option))).next();
}
@Override
public Mono<String> flushAll(RedisClusterNode node) {
return connection.execute(node, RedisServerReactiveCommands::flushall).next();
}
@Override
public Mono<String> flushAll(RedisClusterNode node, FlushOption option) {
return connection.execute(node, it -> it.flushall(LettuceServerCommands.toFlushMode(option))).next();
}
@Override
public Mono<Properties> info() {
return Flux.merge(executeOnAllNodes(this::info)).collect(PropertiesCollector.INSTANCE);

View File

@@ -25,6 +25,7 @@ import java.util.Properties;
import java.util.concurrent.TimeUnit;
import org.springframework.data.redis.connection.ReactiveServerCommands;
import org.springframework.data.redis.connection.RedisServerCommands.FlushOption;
import org.springframework.data.redis.core.types.RedisClientInfo;
import org.springframework.data.redis.util.ByteUtils;
import org.springframework.util.Assert;
@@ -34,6 +35,7 @@ import org.springframework.util.Assert;
*
* @author Mark Paluch
* @author Christoph Strobl
* @author Dennis Neufeld
*/
class LettuceReactiveServerCommands implements ReactiveServerCommands {
@@ -82,11 +84,21 @@ class LettuceReactiveServerCommands implements ReactiveServerCommands {
return connection.execute(RedisServerReactiveCommands::flushdb).next();
}
@Override
public Mono<String> flushDb(FlushOption option) {
return connection.execute(it -> it.flushdb(LettuceServerCommands.toFlushMode(option))).next();
}
@Override
public Mono<String> flushAll() {
return connection.execute(RedisServerReactiveCommands::flushall).next();
}
@Override
public Mono<String> flushAll(FlushOption option) {
return connection.execute(it -> it.flushall(LettuceServerCommands.toFlushMode(option))).next();
}
@Override
public Mono<Properties> info() {

View File

@@ -15,6 +15,7 @@
*/
package org.springframework.data.redis.connection.lettuce;
import io.lettuce.core.FlushMode;
import io.lettuce.core.LettuceFutures;
import io.lettuce.core.RedisFuture;
import io.lettuce.core.api.async.RedisKeyAsyncCommands;
@@ -34,6 +35,7 @@ import org.springframework.util.Assert;
/**
* @author Mark Paluch
* @author Dennis Neufeld
* @since 2.0
*/
class LettuceServerCommands implements RedisServerCommands {
@@ -74,11 +76,21 @@ class LettuceServerCommands implements RedisServerCommands {
connection.invokeStatus().just(RedisServerAsyncCommands::flushdb);
}
@Override
public void flushDb(FlushOption option) {
connection.invoke().just(it -> it.flushdb(toFlushMode(option)));
}
@Override
public void flushAll() {
connection.invokeStatus().just(RedisServerAsyncCommands::flushall);
}
@Override
public void flushAll(FlushOption option) {
connection.invokeStatus().just(it -> it.flushall(toFlushMode(option)));
}
@Override
public Properties info() {
return connection.invoke().from(RedisServerAsyncCommands::info).get(LettuceConverters.stringToProps());
@@ -221,6 +233,21 @@ class LettuceServerCommands implements RedisServerCommands {
return connection.getConnection();
}
static FlushMode toFlushMode(@Nullable FlushOption option) {
if (option == null) {
return FlushMode.SYNC;
}
switch (option) {
case ASYNC:
return FlushMode.ASYNC;
case SYNC:
return FlushMode.SYNC;
}
throw new UnsupportedOperationException("Flush option " + option + " is not implemented.");
}
static class CompletedRedisFuture<T> extends CompletableFuture<T> implements RedisFuture<T> {
public CompletedRedisFuture(T value) {

View File

@@ -22,6 +22,7 @@ import org.springframework.data.redis.connection.RedisClusterCommands;
import org.springframework.data.redis.connection.RedisClusterNode;
import org.springframework.data.redis.connection.RedisClusterNode.SlotRange;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.RedisServerCommands.FlushOption;
/**
* Redis operations for cluster specific operations. A {@link RedisClusterNode} can be obtained from
@@ -31,6 +32,7 @@ import org.springframework.data.redis.connection.RedisConnection;
*
* @author Christoph Strobl
* @author Mark Paluch
* @author Dennis Neufeld
* @since 1.7
*/
public interface ClusterOperations<K, V> {
@@ -117,6 +119,16 @@ public interface ClusterOperations<K, V> {
*/
void flushDb(RedisClusterNode node);
/**
* Flush db on node using the specified flush option.
*
* @param node must not be {@literal null}.
* @param option
* @see RedisConnection#flushDb(FlushOption)
* @since 2.6
*/
void flushDb(RedisClusterNode node, FlushOption option);
/**
* @param node must not be {@literal null}.
* @return

View File

@@ -23,6 +23,7 @@ import org.springframework.data.redis.connection.RedisClusterCommands.AddSlots;
import org.springframework.data.redis.connection.RedisClusterConnection;
import org.springframework.data.redis.connection.RedisClusterNode;
import org.springframework.data.redis.connection.RedisClusterNode.SlotRange;
import org.springframework.data.redis.connection.RedisServerCommands.FlushOption;
import org.springframework.data.redis.connection.RedisServerCommands.MigrateOption;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
@@ -31,6 +32,7 @@ import org.springframework.util.Assert;
* Default {@link ClusterOperations} implementation.
*
* @author Christoph Strobl
* @author Dennis Neufeld
* @since 1.7
* @param <K>
* @param <V>
@@ -149,6 +151,17 @@ class DefaultClusterOperations<K, V> extends AbstractOperations<K, V> implements
});
}
@Override
public void flushDb(RedisClusterNode node, FlushOption option) {
Assert.notNull(node, "ClusterNode must not be null.");
doInCluster((RedisClusterCallback<Void>) connection -> {
connection.flushDb(node, option);
return null;
});
}
@Override
public Collection<RedisClusterNode> getSlaves(final RedisClusterNode node) {

View File

@@ -486,10 +486,6 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
// Methods dealing with Redis Lua scripts
// -------------------------------------------------------------------------
/*
*(non-Javadoc)
* @see org.springframework.data.redis.core.ReactiveRedisOperations#execute(org.springframework.data.redis.core.script.RedisScript, java.util.List, java.util.List)
*/
@Override
public <T> Flux<T> execute(RedisScript<T> script, List<K> keys, List<?> args) {
return reactiveScriptExecutor.execute(script, keys, args);

View File

@@ -203,10 +203,6 @@ public class DefaultRedisSet<E> extends AbstractRedisCollection<E> implements Re
return DataType.SET;
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.support.collections.RedisSet#scan()
*/
@Override
public Cursor<E> scan() {
return scan(ScanOptions.NONE);