DATAREDIS-659 - Provide ReactiveServerCommands for basic server interaction.
Original Pull Request: #253
This commit is contained in:
Mark Paluch
committed by
Christoph Strobl
Christoph Strobl
parent
772d412ba8
commit
1f2610ba91
@@ -0,0 +1,125 @@
|
||||
/*
|
||||
* Copyright 2017 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
package org.springframework.data.redis.connection;
|
||||
|
||||
import reactor.core.publisher.Flux;
|
||||
import reactor.core.publisher.Mono;
|
||||
|
||||
import java.util.Properties;
|
||||
|
||||
import org.springframework.data.redis.core.types.RedisClientInfo;
|
||||
|
||||
/**
|
||||
* @author Mark Paluch
|
||||
* @since 2.0
|
||||
*/
|
||||
public interface ReactiveClusterServerCommands extends ReactiveServerCommands {
|
||||
|
||||
/**
|
||||
* @param node must not be {@literal null}.
|
||||
* @see RedisServerCommands#bgReWriteAof()
|
||||
*/
|
||||
Mono<String> bgReWriteAof(RedisClusterNode node);
|
||||
|
||||
/**
|
||||
* @param node must not be {@literal null}.
|
||||
* @see RedisServerCommands#bgSave()
|
||||
*/
|
||||
Mono<String> bgSave(RedisClusterNode node);
|
||||
|
||||
/**
|
||||
* @param node must not be {@literal null}.
|
||||
* @return
|
||||
* @see RedisServerCommands#lastSave()
|
||||
*/
|
||||
Mono<Long> lastSave(RedisClusterNode node);
|
||||
|
||||
/**
|
||||
* @param node must not be {@literal null}.
|
||||
* @see RedisServerCommands#save()
|
||||
*/
|
||||
Mono<String> save(RedisClusterNode node);
|
||||
|
||||
/**
|
||||
* @param node must not be {@literal null}.
|
||||
* @return
|
||||
* @see RedisServerCommands#dbSize()
|
||||
*/
|
||||
Mono<Long> dbSize(RedisClusterNode node);
|
||||
|
||||
/**
|
||||
* @param node must not be {@literal null}.
|
||||
* @see RedisServerCommands#flushDb()
|
||||
*/
|
||||
Mono<String> flushDb(RedisClusterNode node);
|
||||
|
||||
/**
|
||||
* @param node must not be {@literal null}.
|
||||
* @see RedisServerCommands#flushAll()
|
||||
*/
|
||||
Mono<String> flushAll(RedisClusterNode node);
|
||||
|
||||
/**
|
||||
* @param node must not be {@literal null}.
|
||||
* @return
|
||||
* @see RedisServerCommands#info()
|
||||
*/
|
||||
Mono<Properties> info(RedisClusterNode node);
|
||||
|
||||
/**
|
||||
* @param node must not be {@literal null}.
|
||||
* @param section
|
||||
* @return
|
||||
* @see RedisServerCommands#info(String)
|
||||
*/
|
||||
Mono<Properties> info(RedisClusterNode node, String section);
|
||||
|
||||
/**
|
||||
* @param node must not be {@literal null}.
|
||||
* @param pattern
|
||||
* @return
|
||||
* @see RedisServerCommands#getConfig(String)
|
||||
*/
|
||||
Mono<Properties> getConfig(RedisClusterNode node, String pattern);
|
||||
|
||||
/**
|
||||
* @param node must not be {@literal null}.
|
||||
* @param param
|
||||
* @param value
|
||||
* @see RedisServerCommands#setConfig(String, String)
|
||||
*/
|
||||
Mono<String> setConfig(RedisClusterNode node, String param, String value);
|
||||
|
||||
/**
|
||||
* @param node must not be {@literal null}.
|
||||
* @see RedisServerCommands#resetConfigStats()
|
||||
*/
|
||||
Mono<String> resetConfigStats(RedisClusterNode node);
|
||||
|
||||
/**
|
||||
* @param node must not be {@literal null}.
|
||||
* @return
|
||||
* @see RedisServerCommands#time()
|
||||
*/
|
||||
Mono<Long> time(RedisClusterNode node);
|
||||
|
||||
/**
|
||||
* @param node must not be {@literal null}.
|
||||
* @return
|
||||
* @see RedisServerCommands#getClientList()
|
||||
*/
|
||||
Flux<RedisClientInfo> getClientList(RedisClusterNode node);
|
||||
}
|
||||
@@ -15,12 +15,22 @@
|
||||
*/
|
||||
package org.springframework.data.redis.connection;
|
||||
|
||||
import reactor.core.publisher.Mono;
|
||||
|
||||
/**
|
||||
* @author Christoph Strobl
|
||||
* @author Mark Paluch
|
||||
* @since 2.0
|
||||
*/
|
||||
public interface ReactiveRedisClusterConnection extends ReactiveRedisConnection {
|
||||
|
||||
/**
|
||||
* @param node must not be {@literal null}.
|
||||
* @return
|
||||
* @see RedisConnectionCommands#ping()
|
||||
*/
|
||||
Mono<String> ping(RedisClusterNode node);
|
||||
|
||||
@Override
|
||||
ReactiveClusterKeyCommands keyCommands();
|
||||
|
||||
@@ -47,4 +57,7 @@ public interface ReactiveRedisClusterConnection extends ReactiveRedisConnection
|
||||
|
||||
@Override
|
||||
ReactiveClusterHyperLogLogCommands hyperLogLogCommands();
|
||||
|
||||
@Override
|
||||
ReactiveClusterServerCommands serverCommands();
|
||||
}
|
||||
|
||||
@@ -16,6 +16,7 @@
|
||||
package org.springframework.data.redis.connection;
|
||||
|
||||
import lombok.Data;
|
||||
import reactor.core.publisher.Mono;
|
||||
|
||||
import java.io.Closeable;
|
||||
import java.nio.ByteBuffer;
|
||||
@@ -114,6 +115,21 @@ public interface ReactiveRedisConnection extends Closeable {
|
||||
*/
|
||||
ReactiveHyperLogLogCommands hyperLogLogCommands();
|
||||
|
||||
/**
|
||||
* Get {@link ReactiveServerCommands}.
|
||||
*
|
||||
* @return never {@literal null}.
|
||||
*/
|
||||
ReactiveServerCommands serverCommands();
|
||||
|
||||
/**
|
||||
* Test connection.
|
||||
*
|
||||
* @return Server response message - usually {@literal PONG}.
|
||||
* @see <a href="http://redis.io/commands/ping">Redis Documentation: PING</a>
|
||||
*/
|
||||
Mono<String> ping();
|
||||
|
||||
/**
|
||||
* Base interface for Redis commands executed with a reactive infrastructure.
|
||||
*
|
||||
|
||||
@@ -0,0 +1,174 @@
|
||||
/*
|
||||
* Copyright 2017 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
package org.springframework.data.redis.connection;
|
||||
|
||||
import reactor.core.publisher.Flux;
|
||||
import reactor.core.publisher.Mono;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.Properties;
|
||||
|
||||
import org.springframework.data.redis.core.types.RedisClientInfo;
|
||||
|
||||
/**
|
||||
* Redis Server commands executed using reactive infrastructure.
|
||||
*
|
||||
* @author Mark Paluch
|
||||
* @since 2.0
|
||||
*/
|
||||
public interface ReactiveServerCommands {
|
||||
|
||||
/**
|
||||
* Start an {@literal Append Only File} rewrite process on server.
|
||||
*
|
||||
* @see <a href="http://redis.io/commands/bgrewriteaof">Redis Documentation: BGREWRITEAOF</a>
|
||||
*/
|
||||
Mono<String> bgReWriteAof();
|
||||
|
||||
/**
|
||||
* Start background saving of db on server.
|
||||
*
|
||||
* @see <a href="http://redis.io/commands/bgsave">Redis Documentation: BGSAVE</a>
|
||||
*/
|
||||
Mono<String> bgSave();
|
||||
|
||||
/**
|
||||
* Get time of last {@link #bgSave()} operation in seconds.
|
||||
*
|
||||
* @return
|
||||
* @see <a href="http://redis.io/commands/lastsave">Redis Documentation: LASTSAVE</a>
|
||||
*/
|
||||
Mono<Long> lastSave();
|
||||
|
||||
/**
|
||||
* Synchronous save current db snapshot on server.
|
||||
*
|
||||
* @see <a href="http://redis.io/commands/save">Redis Documentation: SAVE</a>
|
||||
*/
|
||||
Mono<String> save();
|
||||
|
||||
/**
|
||||
* Get the total number of available keys in currently selected database.
|
||||
*
|
||||
* @return
|
||||
* @see <a href="http://redis.io/commands/dbsize">Redis Documentation: DBSIZE</a>
|
||||
*/
|
||||
Mono<Long> dbSize();
|
||||
|
||||
/**
|
||||
* Delete all keys of the currently selected database.
|
||||
*
|
||||
* @see <a href="http://redis.io/commands/flushdb">Redis Documentation: FLUSHDB</a>
|
||||
*/
|
||||
Mono<String> flushDb();
|
||||
|
||||
/**
|
||||
* Delete all <b>all keys</b> from <b>all databases</b>.
|
||||
*
|
||||
* @see <a href="http://redis.io/commands/flushall">Redis Documentation: FLUSHALL</a>
|
||||
*/
|
||||
Mono<String> flushAll();
|
||||
|
||||
/**
|
||||
* Load {@literal default} server information like
|
||||
* <ul>
|
||||
* <li>memory</li>
|
||||
* <li>cpu utilization</li>
|
||||
* <li>replication</li>
|
||||
* </ul>
|
||||
* <p>
|
||||
*
|
||||
* @return
|
||||
* @see <a href="http://redis.io/commands/info">Redis Documentation: INFO</a>
|
||||
*/
|
||||
Mono<Properties> info();
|
||||
|
||||
/**
|
||||
* Load server information for given {@code selection}.
|
||||
*
|
||||
* @param section
|
||||
* @return
|
||||
* @see <a href="http://redis.io/commands/info">Redis Documentation: INFO</a>
|
||||
*/
|
||||
Mono<Properties> info(String section);
|
||||
|
||||
/**
|
||||
* Load configuration parameters for given {@code pattern} from server.
|
||||
*
|
||||
* @param pattern
|
||||
* @return
|
||||
* @see <a href="http://redis.io/commands/config-get">Redis Documentation: CONFIG GET</a>
|
||||
*/
|
||||
Mono<Properties> getConfig(String pattern);
|
||||
|
||||
/**
|
||||
* Set server configuration for {@code param} to {@code value}.
|
||||
*
|
||||
* @param param
|
||||
* @param value
|
||||
* @see <a href="http://redis.io/commands/config-set">Redis Documentation: CONFIG SET</a>
|
||||
*/
|
||||
Mono<String> setConfig(String param, String value);
|
||||
|
||||
/**
|
||||
* Reset statistic counters on server. <br>
|
||||
* Counters can be retrieved using {@link #info()}.
|
||||
*
|
||||
* @see <a href="http://redis.io/commands/config-resetstat">Redis Documentation: CONFIG RESETSTAT</a>
|
||||
*/
|
||||
Mono<String> resetConfigStats();
|
||||
|
||||
/**
|
||||
* Request server timestamp using {@code TIME} command.
|
||||
*
|
||||
* @return current server time in milliseconds.
|
||||
* @see <a href="http://redis.io/commands/time">Redis Documentation: TIME</a>
|
||||
*/
|
||||
Mono<Long> time();
|
||||
|
||||
/**
|
||||
* Closes a given client connection identified by {@literal host:port}.
|
||||
*
|
||||
* @param host of connection to close.
|
||||
* @param port of connection to close
|
||||
* @see <a href="http://redis.io/commands/client-kill">Redis Documentation: CLIENT KILL</a>
|
||||
*/
|
||||
Mono<String> killClient(String host, int port);
|
||||
|
||||
/**
|
||||
* Assign given name to current connection.
|
||||
*
|
||||
* @param name
|
||||
* @see <a href="http://redis.io/commands/client-setname">Redis Documentation: CLIENT SETNAME</a>
|
||||
*/
|
||||
Mono<String> setClientName(String name);
|
||||
|
||||
/**
|
||||
* Returns the name of the current connection.
|
||||
*
|
||||
* @see <a href="http://redis.io/commands/client-getname">Redis Documentation: CLIENT GETNAME</a>
|
||||
* @return
|
||||
*/
|
||||
Mono<String> getClientName();
|
||||
|
||||
/**
|
||||
* Request information and statistics about connected clients.
|
||||
*
|
||||
* @return {@link List} of {@link RedisClientInfo} objects.
|
||||
* @see <a href="http://redis.io/commands/client-list">Redis Documentation: CLIENT LIST</a>
|
||||
*/
|
||||
Flux<RedisClientInfo> getClientList();
|
||||
}
|
||||
@@ -29,7 +29,6 @@ import io.lettuce.core.codec.RedisCodec;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collection;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.LinkedHashSet;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
@@ -614,35 +613,4 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Lettuce specific implementation of {@link ClusterTopologyProvider}.
|
||||
*
|
||||
* @author Christoph Strobl
|
||||
* @since 1.7
|
||||
*/
|
||||
static class LettuceClusterTopologyProvider implements ClusterTopologyProvider {
|
||||
|
||||
private final RedisClusterClient client;
|
||||
|
||||
/**
|
||||
* @param client must not be {@literal null}.
|
||||
*/
|
||||
public LettuceClusterTopologyProvider(RedisClusterClient client) {
|
||||
|
||||
Assert.notNull(client, "RedisClusterClient must not be null.");
|
||||
this.client = client;
|
||||
}
|
||||
|
||||
/*
|
||||
* (non-Javadoc)
|
||||
* @see org.springframework.data.redis.connection.ClusterTopologyProvider#getTopology()
|
||||
*/
|
||||
@Override
|
||||
public ClusterTopology getTopology() {
|
||||
return new ClusterTopology(
|
||||
new LinkedHashSet<>(LettuceConverters.partitionsToClusterNodes(client.getPartitions())));
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -0,0 +1,55 @@
|
||||
/*
|
||||
* Copyright 2017 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
package org.springframework.data.redis.connection.lettuce;
|
||||
|
||||
import io.lettuce.core.cluster.RedisClusterClient;
|
||||
|
||||
import java.util.LinkedHashSet;
|
||||
|
||||
import org.springframework.data.redis.connection.ClusterTopology;
|
||||
import org.springframework.data.redis.connection.ClusterTopologyProvider;
|
||||
import org.springframework.util.Assert;
|
||||
|
||||
/**
|
||||
* Lettuce specific implementation of {@link ClusterTopologyProvider}.
|
||||
*
|
||||
* @author Christoph Strobl
|
||||
* @author Mark Paluch
|
||||
* @since 1.7
|
||||
*/
|
||||
class LettuceClusterTopologyProvider implements ClusterTopologyProvider {
|
||||
|
||||
private final RedisClusterClient client;
|
||||
|
||||
/**
|
||||
* @param client must not be {@literal null}.
|
||||
*/
|
||||
LettuceClusterTopologyProvider(RedisClusterClient client) {
|
||||
|
||||
Assert.notNull(client, "RedisClusterClient must not be null.");
|
||||
|
||||
this.client = client;
|
||||
}
|
||||
|
||||
/*
|
||||
* (non-Javadoc)
|
||||
* @see org.springframework.data.redis.connection.ClusterTopologyProvider#getTopology()
|
||||
*/
|
||||
@Override
|
||||
public ClusterTopology getTopology() {
|
||||
return new ClusterTopology(new LinkedHashSet<>(LettuceConverters.partitionsToClusterNodes(client.getPartitions())));
|
||||
}
|
||||
}
|
||||
@@ -41,18 +41,7 @@ import org.springframework.dao.InvalidDataAccessResourceUsageException;
|
||||
import org.springframework.data.redis.ExceptionTranslationStrategy;
|
||||
import org.springframework.data.redis.PassThroughExceptionTranslationStrategy;
|
||||
import org.springframework.data.redis.RedisConnectionFailureException;
|
||||
import org.springframework.data.redis.connection.ClusterCommandExecutor;
|
||||
import org.springframework.data.redis.connection.Pool;
|
||||
import org.springframework.data.redis.connection.ReactiveRedisConnectionFactory;
|
||||
import org.springframework.data.redis.connection.RedisClusterConfiguration;
|
||||
import org.springframework.data.redis.connection.RedisClusterConnection;
|
||||
import org.springframework.data.redis.connection.RedisConnection;
|
||||
import org.springframework.data.redis.connection.RedisConnectionFactory;
|
||||
import org.springframework.data.redis.connection.RedisNode;
|
||||
import org.springframework.data.redis.connection.RedisPassword;
|
||||
import org.springframework.data.redis.connection.RedisSentinelConfiguration;
|
||||
import org.springframework.data.redis.connection.RedisSentinelConnection;
|
||||
import org.springframework.data.redis.connection.RedisStandaloneConfiguration;
|
||||
import org.springframework.data.redis.connection.*;
|
||||
import org.springframework.util.Assert;
|
||||
import org.springframework.util.ClassUtils;
|
||||
|
||||
@@ -772,7 +761,7 @@ public class LettuceConnectionFactory
|
||||
.orElseGet(() -> RedisClusterClient.create(initialUris));
|
||||
|
||||
this.clusterCommandExecutor = new ClusterCommandExecutor(
|
||||
new LettuceClusterConnection.LettuceClusterTopologyProvider(clusterClient),
|
||||
new LettuceClusterTopologyProvider(clusterClient),
|
||||
new LettuceClusterConnection.LettuceClusterNodeResourceProvider(clusterClient), EXCEPTION_TRANSLATION);
|
||||
|
||||
clientConfiguration.getClientOptions() //
|
||||
|
||||
@@ -856,6 +856,27 @@ abstract public class LettuceConverters extends Converters {
|
||||
return geoArgs;
|
||||
}
|
||||
|
||||
/**
|
||||
* Convert {@code CONFIG GET} output from a {@link List} to {@link Properties}.
|
||||
*
|
||||
* @param input must not be {@literal null}.
|
||||
* @return the mapped result.
|
||||
*/
|
||||
public static Properties toProperties(List<String> input) {
|
||||
|
||||
Assert.notNull(input, "Input list must not be null!");
|
||||
Assert.isTrue(input.size() % 2 == 0, "Input list must contain an even number of entries!");
|
||||
|
||||
Properties properties = new Properties();
|
||||
|
||||
for (int i = 0; i < input.size(); i += 2) {
|
||||
|
||||
properties.setProperty(input.get(i), input.get(i + 1));
|
||||
}
|
||||
|
||||
return properties;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get {@link Converter} capable of {@link Set} of {@link Byte} into {@link GeoResults}.
|
||||
*
|
||||
|
||||
@@ -0,0 +1,337 @@
|
||||
/*
|
||||
* Copyright 2017 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
package org.springframework.data.redis.connection.lettuce;
|
||||
|
||||
import io.lettuce.core.api.reactive.RedisServerReactiveCommands;
|
||||
import reactor.core.publisher.Flux;
|
||||
import reactor.core.publisher.Mono;
|
||||
import reactor.util.function.Tuple2;
|
||||
import reactor.util.function.Tuples;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.Arrays;
|
||||
import java.util.Collection;
|
||||
import java.util.Date;
|
||||
import java.util.HashSet;
|
||||
import java.util.List;
|
||||
import java.util.Map.Entry;
|
||||
import java.util.Properties;
|
||||
import java.util.Set;
|
||||
import java.util.function.BiConsumer;
|
||||
import java.util.function.BinaryOperator;
|
||||
import java.util.function.Function;
|
||||
import java.util.function.Supplier;
|
||||
import java.util.stream.Collector;
|
||||
|
||||
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.core.types.RedisClientInfo;
|
||||
import org.springframework.data.redis.util.ByteUtils;
|
||||
|
||||
/**
|
||||
* @author Mark Paluch
|
||||
* @since 2.0
|
||||
*/
|
||||
class LettuceReactiveClusterServerCommands extends LettuceReactiveServerCommands
|
||||
implements ReactiveClusterServerCommands {
|
||||
|
||||
private final LettuceReactiveRedisClusterConnection connection;
|
||||
private final ClusterTopologyProvider topologyProvider;
|
||||
|
||||
/**
|
||||
* Create new {@link LettuceReactiveGeoCommands}.
|
||||
*
|
||||
* @param connection must not be {@literal null}.
|
||||
*/
|
||||
public LettuceReactiveClusterServerCommands(LettuceReactiveRedisClusterConnection connection,
|
||||
ClusterTopologyProvider topologyProvider) {
|
||||
|
||||
super(connection);
|
||||
|
||||
this.connection = connection;
|
||||
this.topologyProvider = topologyProvider;
|
||||
}
|
||||
|
||||
/* (non-Javadoc)
|
||||
* @see org.springframework.data.redis.connection.ReactiveClusterServerCommands#bgReWriteAof(org.springframework.data.redis.connection.RedisClusterNode)
|
||||
*/
|
||||
@Override
|
||||
public Mono<String> bgReWriteAof(RedisClusterNode node) {
|
||||
return connection.execute(node, RedisServerReactiveCommands::bgrewriteaof).next();
|
||||
}
|
||||
|
||||
/* (non-Javadoc)
|
||||
* @see org.springframework.data.redis.connection.ReactiveClusterServerCommands#bgSave(org.springframework.data.redis.connection.RedisClusterNode)
|
||||
*/
|
||||
@Override
|
||||
public Mono<String> bgSave(RedisClusterNode node) {
|
||||
return connection.execute(node, RedisServerReactiveCommands::bgsave).next();
|
||||
}
|
||||
|
||||
/* (non-Javadoc)
|
||||
* @see org.springframework.data.redis.connection.ReactiveClusterServerCommands#lastSave(org.springframework.data.redis.connection.RedisClusterNode)
|
||||
*/
|
||||
@Override
|
||||
public Mono<Long> lastSave(RedisClusterNode node) {
|
||||
return connection.execute(node, RedisServerReactiveCommands::lastsave).map(Date::getTime).next();
|
||||
}
|
||||
|
||||
/* (non-Javadoc)
|
||||
* @see org.springframework.data.redis.connection.ReactiveClusterServerCommands#save(org.springframework.data.redis.connection.RedisClusterNode)
|
||||
*/
|
||||
@Override
|
||||
public Mono<String> save(RedisClusterNode node) {
|
||||
return connection.execute(node, RedisServerReactiveCommands::save).next();
|
||||
}
|
||||
|
||||
/* (non-Javadoc)
|
||||
* @see org.springframework.data.redis.connection.ReactiveClusterServerCommands#dbSize(org.springframework.data.redis.connection.RedisClusterNode)
|
||||
*/
|
||||
@Override
|
||||
public Mono<Long> dbSize(RedisClusterNode node) {
|
||||
return connection.execute(node, RedisServerReactiveCommands::dbsize).next();
|
||||
}
|
||||
|
||||
/* (non-Javadoc)
|
||||
* @see org.springframework.data.redis.connection.ReactiveClusterServerCommands#flushDb(org.springframework.data.redis.connection.RedisClusterNode)
|
||||
*/
|
||||
@Override
|
||||
public Mono<String> flushDb(RedisClusterNode node) {
|
||||
return connection.execute(node, RedisServerReactiveCommands::flushdb).next();
|
||||
}
|
||||
|
||||
/* (non-Javadoc)
|
||||
* @see org.springframework.data.redis.connection.ReactiveClusterServerCommands#flushAll(org.springframework.data.redis.connection.RedisClusterNode)
|
||||
*/
|
||||
@Override
|
||||
public Mono<String> flushAll(RedisClusterNode node) {
|
||||
return connection.execute(node, RedisServerReactiveCommands::flushall).next();
|
||||
}
|
||||
|
||||
/* (non-Javadoc)
|
||||
* @see org.springframework.data.redis.connection.lettuce.LettuceReactiveServerCommands#info()
|
||||
*/
|
||||
@Override
|
||||
public Mono<Properties> info() {
|
||||
return Flux.merge(executeOnAllNodes(this::info)).collect(PropertiesCollector.INSTANCE);
|
||||
}
|
||||
|
||||
/* (non-Javadoc)
|
||||
* @see org.springframework.data.redis.connection.ReactiveClusterServerCommands#info(org.springframework.data.redis.connection.RedisClusterNode)
|
||||
*/
|
||||
@Override
|
||||
public Mono<Properties> info(RedisClusterNode node) {
|
||||
|
||||
return connection.execute(node, RedisServerReactiveCommands::info) //
|
||||
.map(LettuceConverters::toProperties) //
|
||||
.next();
|
||||
}
|
||||
|
||||
/* (non-Javadoc)
|
||||
* @see org.springframework.data.redis.connection.lettuce.LettuceReactiveServerCommands#info(java.lang.String)
|
||||
*/
|
||||
@Override
|
||||
public Mono<Properties> info(String section) {
|
||||
|
||||
return Flux.merge(executeOnAllNodes(redisClusterNode -> info(redisClusterNode, section)))
|
||||
.collect(PropertiesCollector.INSTANCE);
|
||||
}
|
||||
|
||||
/* (non-Javadoc)
|
||||
* @see org.springframework.data.redis.connection.ReactiveClusterServerCommands#info(org.springframework.data.redis.connection.RedisClusterNode, java.lang.String)
|
||||
*/
|
||||
@Override
|
||||
public Mono<Properties> info(RedisClusterNode node, String section) {
|
||||
|
||||
return connection.execute(node, c -> c.info(section)) //
|
||||
.map(LettuceConverters::toProperties).next();
|
||||
}
|
||||
|
||||
/* (non-Javadoc)
|
||||
* @see org.springframework.data.redis.connection.lettuce.LettuceReactiveServerCommands#getConfig(java.lang.String)
|
||||
*/
|
||||
@Override
|
||||
public Mono<Properties> getConfig(String pattern) {
|
||||
|
||||
return Flux.merge(executeOnAllNodes(node -> getConfig(node, pattern))) //
|
||||
.collect(PropertiesCollector.INSTANCE);
|
||||
}
|
||||
|
||||
/* (non-Javadoc)
|
||||
* @see org.springframework.data.redis.connection.ReactiveClusterServerCommands#getConfig(org.springframework.data.redis.connection.RedisClusterNode, java.lang.String)
|
||||
*/
|
||||
@Override
|
||||
public Mono<Properties> getConfig(RedisClusterNode node, String pattern) {
|
||||
|
||||
return connection.execute(node, c -> c.configGet(pattern).collectList()) //
|
||||
.map(LettuceConverters::toProperties) //
|
||||
.next();
|
||||
}
|
||||
|
||||
/* (non-Javadoc)
|
||||
* @see org.springframework.data.redis.connection.lettuce.LettuceReactiveServerCommands#setConfig(java.lang.String, java.lang.String)
|
||||
*/
|
||||
@Override
|
||||
public Mono<String> setConfig(String param, String value) {
|
||||
return Flux.merge(executeOnAllNodes(node -> setConfig(node, param, value))).map(Tuple2::getT2).last();
|
||||
}
|
||||
|
||||
/* (non-Javadoc)
|
||||
* @see org.springframework.data.redis.connection.ReactiveClusterServerCommands#setConfig(org.springframework.data.redis.connection.RedisClusterNode, java.lang.String, java.lang.String)
|
||||
*/
|
||||
@Override
|
||||
public Mono<String> setConfig(RedisClusterNode node, String param, String value) {
|
||||
return connection.execute(node, c -> c.configSet(param, value)).next();
|
||||
}
|
||||
|
||||
/* (non-Javadoc)
|
||||
* @see org.springframework.data.redis.connection.lettuce.LettuceReactiveServerCommands#resetConfigStats()
|
||||
*/
|
||||
@Override
|
||||
public Mono<String> resetConfigStats() {
|
||||
return Flux.merge(executeOnAllNodes(this::resetConfigStats)).map(Tuple2::getT2).last();
|
||||
}
|
||||
|
||||
/* (non-Javadoc)
|
||||
* @see org.springframework.data.redis.connection.ReactiveClusterServerCommands#resetConfigStats(org.springframework.data.redis.connection.RedisClusterNode)
|
||||
*/
|
||||
@Override
|
||||
public Mono<String> resetConfigStats(RedisClusterNode node) {
|
||||
return connection.execute(node, RedisServerReactiveCommands::configResetstat).next();
|
||||
}
|
||||
|
||||
/* (non-Javadoc)
|
||||
* @see org.springframework.data.redis.connection.ReactiveClusterServerCommands#time(org.springframework.data.redis.connection.RedisClusterNode)
|
||||
*/
|
||||
@Override
|
||||
public Mono<Long> time(RedisClusterNode node) {
|
||||
|
||||
return connection.execute(node, RedisServerReactiveCommands::time) //
|
||||
.map(ByteUtils::getBytes) //
|
||||
.collectList() //
|
||||
.map(LettuceConverters.toTimeConverter()::convert);
|
||||
}
|
||||
|
||||
/* (non-Javadoc)
|
||||
* @see org.springframework.data.redis.connection.lettuce.LettuceReactiveServerCommands#getClientList()
|
||||
*/
|
||||
@Override
|
||||
public Flux<RedisClientInfo> getClientList() {
|
||||
return Flux.merge(executeOnAllNodesMany(this::getClientList)).map(Tuple2::getT2);
|
||||
}
|
||||
|
||||
/* (non-Javadoc)
|
||||
* @see org.springframework.data.redis.connection.ReactiveClusterServerCommands#getClientList(org.springframework.data.redis.connection.RedisClusterNode)
|
||||
*/
|
||||
@Override
|
||||
public Flux<RedisClientInfo> getClientList(RedisClusterNode node) {
|
||||
|
||||
return connection.execute(node, RedisServerReactiveCommands::clientList)
|
||||
.flatMapIterable(LettuceConverters.stringToRedisClientListConverter()::convert);
|
||||
}
|
||||
|
||||
private <T> Collection<Publisher<Tuple2<RedisClusterNode, T>>> executeOnAllNodes(
|
||||
Function<RedisClusterNode, Mono<T>> callback) {
|
||||
|
||||
Set<RedisClusterNode> nodes = topologyProvider.getTopology().getNodes();
|
||||
List<Publisher<Tuple2<RedisClusterNode, T>>> pipeline = new ArrayList<>(nodes.size());
|
||||
|
||||
for (RedisClusterNode redisClusterNode : nodes) {
|
||||
pipeline.add(callback.apply(redisClusterNode).map(p -> Tuples.of(redisClusterNode, p)));
|
||||
}
|
||||
|
||||
return pipeline;
|
||||
}
|
||||
|
||||
private <T> Collection<Publisher<Tuple2<RedisClusterNode, T>>> executeOnAllNodesMany(
|
||||
Function<RedisClusterNode, Flux<T>> callback) {
|
||||
|
||||
Set<RedisClusterNode> nodes = topologyProvider.getTopology().getNodes();
|
||||
List<Publisher<Tuple2<RedisClusterNode, T>>> pipeline = new ArrayList<>(nodes.size());
|
||||
|
||||
for (RedisClusterNode redisClusterNode : nodes) {
|
||||
pipeline.add(callback.apply(redisClusterNode).map(p -> Tuples.of(redisClusterNode, p)));
|
||||
}
|
||||
|
||||
return pipeline;
|
||||
}
|
||||
|
||||
/**
|
||||
* Collector to merge {@link Tuple2} of {@link RedisClusterNode} and {@link Properties} into a single
|
||||
* {@link Properties} object by prefixing original the keys with {@link RedisClusterNode#asString()}.
|
||||
*/
|
||||
private enum PropertiesCollector implements Collector<Tuple2<RedisClusterNode, Properties>, Properties, Properties> {
|
||||
|
||||
INSTANCE;
|
||||
|
||||
/* (non-Javadoc)
|
||||
* @see java.util.stream.Collector#supplier()
|
||||
*/
|
||||
@Override
|
||||
public Supplier<Properties> supplier() {
|
||||
return Properties::new;
|
||||
}
|
||||
|
||||
/* (non-Javadoc)
|
||||
* @see java.util.stream.Collector#accumulator()
|
||||
*/
|
||||
@Override
|
||||
public BiConsumer<Properties, Tuple2<RedisClusterNode, Properties>> accumulator() {
|
||||
|
||||
return (properties, tuple) -> {
|
||||
|
||||
for (Entry<Object, Object> entry : tuple.getT2().entrySet()) {
|
||||
properties.put(tuple.getT1().asString() + "." + entry.getKey(), entry.getValue());
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
/* (non-Javadoc)
|
||||
* @see java.util.stream.Collector#combiner()
|
||||
*/
|
||||
@Override
|
||||
public BinaryOperator<Properties> combiner() {
|
||||
|
||||
return (left, right) -> {
|
||||
|
||||
Properties merge = new Properties();
|
||||
|
||||
merge.putAll(left);
|
||||
merge.putAll(right);
|
||||
|
||||
return merge;
|
||||
};
|
||||
}
|
||||
|
||||
/* (non-Javadoc)
|
||||
* @see java.util.stream.Collector#finisher()
|
||||
*/
|
||||
@Override
|
||||
public Function<Properties, Properties> finisher() {
|
||||
return properties -> properties;
|
||||
}
|
||||
|
||||
/* (non-Javadoc)
|
||||
* @see java.util.stream.Collector#characteristics()
|
||||
*/
|
||||
@Override
|
||||
public Set<Characteristics> characteristics() {
|
||||
return new HashSet<>(Arrays.asList(Characteristics.UNORDERED, Characteristics.IDENTITY_FINISH));
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -15,15 +15,19 @@
|
||||
*/
|
||||
package org.springframework.data.redis.connection.lettuce;
|
||||
|
||||
import io.lettuce.core.api.reactive.BaseRedisReactiveCommands;
|
||||
import io.lettuce.core.api.reactive.RedisReactiveCommands;
|
||||
import io.lettuce.core.cluster.RedisClusterClient;
|
||||
import io.lettuce.core.cluster.api.StatefulRedisClusterConnection;
|
||||
import io.lettuce.core.cluster.api.reactive.RedisClusterReactiveCommands;
|
||||
import reactor.core.publisher.Flux;
|
||||
import reactor.core.publisher.Mono;
|
||||
|
||||
import java.nio.ByteBuffer;
|
||||
|
||||
import org.springframework.data.redis.connection.ClusterTopologyProvider;
|
||||
import org.springframework.data.redis.connection.ReactiveRedisClusterConnection;
|
||||
import org.springframework.data.redis.connection.RedisClusterNode;
|
||||
import org.springframework.data.redis.connection.RedisNode;
|
||||
import org.springframework.util.Assert;
|
||||
import org.springframework.util.StringUtils;
|
||||
@@ -36,8 +40,13 @@ import org.springframework.util.StringUtils;
|
||||
class LettuceReactiveRedisClusterConnection extends LettuceReactiveRedisConnection
|
||||
implements ReactiveRedisClusterConnection {
|
||||
|
||||
private final ClusterTopologyProvider topologyProvider;
|
||||
|
||||
public LettuceReactiveRedisClusterConnection(RedisClusterClient client) {
|
||||
|
||||
super(client);
|
||||
|
||||
this.topologyProvider = new LettuceClusterTopologyProvider(client);
|
||||
}
|
||||
|
||||
/* (non-Javadoc)
|
||||
@@ -112,6 +121,22 @@ class LettuceReactiveRedisClusterConnection extends LettuceReactiveRedisConnecti
|
||||
return new LettuceReactiveClusterNumberCommands(this);
|
||||
}
|
||||
|
||||
/* (non-Javadoc)
|
||||
* @see org.springframework.data.redis.connection.lettuce.LettuceReactiveRedisConnection#serverCommands()
|
||||
*/
|
||||
@Override
|
||||
public LettuceReactiveClusterServerCommands serverCommands() {
|
||||
return new LettuceReactiveClusterServerCommands(this, topologyProvider);
|
||||
}
|
||||
|
||||
/* (non-Javadoc)
|
||||
* @see org.springframework.data.redis.connection.ReactiveRedisClusterConnection#ping(org.springframework.data.redis.connection.RedisClusterNode)
|
||||
*/
|
||||
@Override
|
||||
public Mono<String> ping(RedisClusterNode node) {
|
||||
return execute(node, BaseRedisReactiveCommands::ping).next();
|
||||
}
|
||||
|
||||
/**
|
||||
* @param callback
|
||||
* @return
|
||||
@@ -119,8 +144,8 @@ class LettuceReactiveRedisClusterConnection extends LettuceReactiveRedisConnecti
|
||||
public <T> Flux<T> execute(RedisNode node, LettuceReactiveCallback<T> callback) {
|
||||
|
||||
try {
|
||||
Assert.notNull(node, "RedisClusterNode must not be null!");
|
||||
Assert.notNull(callback, "ReactiveCallback must not be null!");
|
||||
Assert.notNull(node, "Node must not be null!");
|
||||
} catch (IllegalArgumentException e) {
|
||||
return Flux.error(e);
|
||||
}
|
||||
@@ -131,7 +156,7 @@ class LettuceReactiveRedisClusterConnection extends LettuceReactiveRedisConnecti
|
||||
/* (non-Javadoc)
|
||||
* @see org.springframework.data.redis.connection.lettuce.LettuceReactiveRedisConnection#getConnection()
|
||||
*/
|
||||
@SuppressWarnings("unchecked")
|
||||
@SuppressWarnings({ "unchecked", "rawtypes" })
|
||||
@Override
|
||||
protected StatefulRedisClusterConnection<ByteBuffer, ByteBuffer> getConnection() {
|
||||
|
||||
@@ -148,7 +173,7 @@ class LettuceReactiveRedisClusterConnection extends LettuceReactiveRedisConnecti
|
||||
return getConnection().reactive();
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
@SuppressWarnings({ "unchecked", "rawtypes" })
|
||||
protected RedisReactiveCommands<ByteBuffer, ByteBuffer> getCommands(RedisNode node) {
|
||||
|
||||
if (!(getConnection() instanceof StatefulRedisClusterConnection)) {
|
||||
|
||||
@@ -19,11 +19,13 @@ import io.lettuce.core.AbstractRedisClient;
|
||||
import io.lettuce.core.RedisClient;
|
||||
import io.lettuce.core.api.StatefulConnection;
|
||||
import io.lettuce.core.api.StatefulRedisConnection;
|
||||
import io.lettuce.core.api.reactive.BaseRedisReactiveCommands;
|
||||
import io.lettuce.core.cluster.RedisClusterClient;
|
||||
import io.lettuce.core.cluster.api.StatefulRedisClusterConnection;
|
||||
import io.lettuce.core.cluster.api.reactive.RedisClusterReactiveCommands;
|
||||
import io.lettuce.core.codec.RedisCodec;
|
||||
import reactor.core.publisher.Flux;
|
||||
import reactor.core.publisher.Mono;
|
||||
|
||||
import java.nio.ByteBuffer;
|
||||
import java.util.function.Function;
|
||||
@@ -31,16 +33,7 @@ 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;
|
||||
|
||||
/**
|
||||
@@ -141,6 +134,22 @@ class LettuceReactiveRedisConnection implements ReactiveRedisConnection {
|
||||
return new LettuceReactiveHyperLogLogCommands(this);
|
||||
}
|
||||
|
||||
/* (non-Javadoc)
|
||||
* @see org.springframework.data.redis.connection.ReactiveRedisConnection#hyperLogLogCommands()
|
||||
*/
|
||||
@Override
|
||||
public ReactiveServerCommands serverCommands() {
|
||||
return new LettuceReactiveServerCommands(this);
|
||||
}
|
||||
|
||||
/* (non-Javadoc)
|
||||
* @see org.springframework.data.redis.connection.ReactiveRedisConnection#ping()
|
||||
*/
|
||||
@Override
|
||||
public Mono<String> ping() {
|
||||
return execute(BaseRedisReactiveCommands::ping).next();
|
||||
}
|
||||
|
||||
/**
|
||||
* @param callback
|
||||
* @return
|
||||
|
||||
@@ -0,0 +1,206 @@
|
||||
/*
|
||||
* Copyright 2017 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
package org.springframework.data.redis.connection.lettuce;
|
||||
|
||||
import io.lettuce.core.api.reactive.RedisServerReactiveCommands;
|
||||
import reactor.core.publisher.Flux;
|
||||
import reactor.core.publisher.Mono;
|
||||
|
||||
import java.nio.ByteBuffer;
|
||||
import java.util.Date;
|
||||
import java.util.Properties;
|
||||
|
||||
import org.springframework.data.redis.connection.ReactiveServerCommands;
|
||||
import org.springframework.data.redis.core.types.RedisClientInfo;
|
||||
import org.springframework.data.redis.util.ByteUtils;
|
||||
import org.springframework.util.Assert;
|
||||
|
||||
/**
|
||||
* @author Mark Paluch
|
||||
*/
|
||||
class LettuceReactiveServerCommands implements ReactiveServerCommands {
|
||||
|
||||
private final LettuceReactiveRedisConnection connection;
|
||||
|
||||
/**
|
||||
* Create new {@link LettuceReactiveGeoCommands}.
|
||||
*
|
||||
* @param connection must not be {@literal null}.
|
||||
*/
|
||||
public LettuceReactiveServerCommands(LettuceReactiveRedisConnection connection) {
|
||||
|
||||
Assert.notNull(connection, "Connection must not be null!");
|
||||
this.connection = connection;
|
||||
}
|
||||
|
||||
/* (non-Javadoc)
|
||||
* @see org.springframework.data.redis.connection.ReactiveServerCommands#bgReWriteAof()
|
||||
*/
|
||||
@Override
|
||||
public Mono<String> bgReWriteAof() {
|
||||
return connection.execute(RedisServerReactiveCommands::bgrewriteaof).next();
|
||||
}
|
||||
|
||||
/* (non-Javadoc)
|
||||
* @see org.springframework.data.redis.connection.ReactiveServerCommands#bgSave()
|
||||
*/
|
||||
@Override
|
||||
public Mono<String> bgSave() {
|
||||
return connection.execute(RedisServerReactiveCommands::bgsave).next();
|
||||
}
|
||||
|
||||
/* (non-Javadoc)
|
||||
* @see org.springframework.data.redis.connection.ReactiveServerCommands#lastSave()
|
||||
*/
|
||||
@Override
|
||||
public Mono<Long> lastSave() {
|
||||
return connection.execute(RedisServerReactiveCommands::lastsave).next().map(Date::getTime);
|
||||
}
|
||||
|
||||
/* (non-Javadoc)
|
||||
* @see org.springframework.data.redis.connection.ReactiveServerCommands#save()
|
||||
*/
|
||||
@Override
|
||||
public Mono<String> save() {
|
||||
return connection.execute(RedisServerReactiveCommands::save).next();
|
||||
}
|
||||
|
||||
/* (non-Javadoc)
|
||||
* @see org.springframework.data.redis.connection.ReactiveServerCommands#dbSize()
|
||||
*/
|
||||
@Override
|
||||
public Mono<Long> dbSize() {
|
||||
return connection.execute(RedisServerReactiveCommands::dbsize).next();
|
||||
}
|
||||
|
||||
/* (non-Javadoc)
|
||||
* @see org.springframework.data.redis.connection.ReactiveServerCommands#flushDb()
|
||||
*/
|
||||
@Override
|
||||
public Mono<String> flushDb() {
|
||||
return connection.execute(RedisServerReactiveCommands::flushdb).next();
|
||||
}
|
||||
|
||||
/* (non-Javadoc)
|
||||
* @see org.springframework.data.redis.connection.ReactiveServerCommands#flushAll()
|
||||
*/
|
||||
@Override
|
||||
public Mono<String> flushAll() {
|
||||
return connection.execute(RedisServerReactiveCommands::flushall).next();
|
||||
}
|
||||
|
||||
/* (non-Javadoc)
|
||||
* @see org.springframework.data.redis.connection.ReactiveServerCommands#info()
|
||||
*/
|
||||
@Override
|
||||
public Mono<Properties> info() {
|
||||
|
||||
return connection.execute(RedisServerReactiveCommands::info) //
|
||||
.map(LettuceConverters::toProperties) //
|
||||
.next();
|
||||
}
|
||||
|
||||
/* (non-Javadoc)
|
||||
* @see org.springframework.data.redis.connection.ReactiveServerCommands#info(java.lang.String)
|
||||
*/
|
||||
@Override
|
||||
public Mono<Properties> info(String section) {
|
||||
|
||||
return connection.execute(c -> c.info(section)) //
|
||||
.map(LettuceConverters::toProperties) //
|
||||
.next();
|
||||
}
|
||||
|
||||
/* (non-Javadoc)
|
||||
* @see org.springframework.data.redis.connection.ReactiveServerCommands#getConfig(java.lang.String)
|
||||
*/
|
||||
@Override
|
||||
public Mono<Properties> getConfig(String pattern) {
|
||||
|
||||
return connection.execute(c -> c.configGet(pattern).collectList()) //
|
||||
.map(LettuceConverters::toProperties).next();
|
||||
}
|
||||
|
||||
/* (non-Javadoc)
|
||||
* @see org.springframework.data.redis.connection.ReactiveServerCommands#setConfig(java.lang.String, java.lang.String)
|
||||
*/
|
||||
@Override
|
||||
public Mono<String> setConfig(String param, String value) {
|
||||
return connection.execute(c -> c.configSet(param, value)).next();
|
||||
}
|
||||
|
||||
/* (non-Javadoc)
|
||||
* @see org.springframework.data.redis.connection.ReactiveServerCommands#resetConfigStats()
|
||||
*/
|
||||
@Override
|
||||
public Mono<String> resetConfigStats() {
|
||||
return connection.execute(RedisServerReactiveCommands::configResetstat).next();
|
||||
}
|
||||
|
||||
/* (non-Javadoc)
|
||||
* @see org.springframework.data.redis.connection.ReactiveServerCommands#time()
|
||||
*/
|
||||
@Override
|
||||
public Mono<Long> time() {
|
||||
|
||||
return connection.execute(RedisServerReactiveCommands::time) //
|
||||
.map(ByteUtils::getBytes) //
|
||||
.collectList() //
|
||||
.map(LettuceConverters.toTimeConverter()::convert);
|
||||
}
|
||||
|
||||
/* (non-Javadoc)
|
||||
* @see org.springframework.data.redis.connection.ReactiveServerCommands#killClient(java.lang.String, int)
|
||||
*/
|
||||
@Override
|
||||
public Mono<String> killClient(String host, int port) {
|
||||
|
||||
Assert.notNull(host, "Host must not be null");
|
||||
|
||||
String client = String.format("%s:%s", host, port);
|
||||
return connection.execute(c -> c.clientKill(client)).next();
|
||||
}
|
||||
|
||||
/* (non-Javadoc)
|
||||
* @see org.springframework.data.redis.connection.ReactiveServerCommands#setClientName(java.lang.String)
|
||||
*/
|
||||
@Override
|
||||
public Mono<String> setClientName(String name) {
|
||||
return connection.execute(c -> c.clientSetname(ByteBuffer.wrap(LettuceConverters.toBytes(name)))).next();
|
||||
}
|
||||
|
||||
/* (non-Javadoc)
|
||||
* @see org.springframework.data.redis.connection.ReactiveServerCommands#getClientName()
|
||||
*/
|
||||
@Override
|
||||
public Mono<String> getClientName() {
|
||||
|
||||
return connection.execute(RedisServerReactiveCommands::clientGetname) //
|
||||
.map(ByteUtils::getBytes) //
|
||||
.map(LettuceConverters::toString) //
|
||||
.next();
|
||||
}
|
||||
|
||||
/* (non-Javadoc)
|
||||
* @see org.springframework.data.redis.connection.ReactiveServerCommands#getClientList()
|
||||
*/
|
||||
@Override
|
||||
public Flux<RedisClientInfo> getClientList() {
|
||||
|
||||
return connection.execute(RedisServerReactiveCommands::clientList)
|
||||
.flatMapIterable(s -> LettuceConverters.stringToRedisClientListConverter().convert(s));
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user