Sentinel and Cluster support. Spring Data Redis integration. #1373
Nikita
7 years ago
parent
a489a82f5d
commit
1ca20f25bb
@ -0,0 +1,42 @@
|
||||
/**
|
||||
* Copyright 2018 Nikita Koksharov
|
||||
*
|
||||
* 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.redisson.client.protocol.decoder;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
import org.redisson.client.handler.State;
|
||||
import org.redisson.client.protocol.Decoder;
|
||||
import org.redisson.client.protocol.decoder.MultiDecoder;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author Nikita Koksharov
|
||||
*
|
||||
*/
|
||||
|
||||
public class TimeLongObjectDecoder implements MultiDecoder<Long> {
|
||||
|
||||
@Override
|
||||
public Decoder<Object> getDecoder(int paramNum, State state) {
|
||||
return null;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Long decode(List<Object> parts, State state) {
|
||||
return ((Long)parts.get(0)) * 1000L + ((Long)parts.get(1)) / 1000L;
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,120 @@
|
||||
/**
|
||||
* Copyright 2018 Nikita Koksharov
|
||||
*
|
||||
* 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.redisson.spring.data.connection;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.net.URI;
|
||||
import java.util.ArrayList;
|
||||
import java.util.EnumSet;
|
||||
import java.util.HashSet;
|
||||
import java.util.List;
|
||||
import java.util.Set;
|
||||
|
||||
import org.redisson.client.handler.State;
|
||||
import org.redisson.client.protocol.Decoder;
|
||||
import org.redisson.misc.URIBuilder;
|
||||
import org.springframework.data.redis.connection.RedisClusterNode;
|
||||
import org.springframework.data.redis.connection.RedisClusterNode.Flag;
|
||||
import org.springframework.data.redis.connection.RedisClusterNode.LinkState;
|
||||
import org.springframework.data.redis.connection.RedisClusterNode.RedisClusterNodeBuilder;
|
||||
import org.springframework.data.redis.connection.RedisClusterNode.SlotRange;
|
||||
import org.springframework.data.redis.connection.RedisNode.NodeType;
|
||||
|
||||
import io.netty.buffer.ByteBuf;
|
||||
import io.netty.util.CharsetUtil;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author Nikita Koksharov
|
||||
*
|
||||
*/
|
||||
public class RedisClusterNodeDecoder implements Decoder<List<RedisClusterNode>> {
|
||||
|
||||
@Override
|
||||
public List<RedisClusterNode> decode(ByteBuf buf, State state) throws IOException {
|
||||
String response = buf.toString(CharsetUtil.UTF_8);
|
||||
|
||||
List<RedisClusterNode> nodes = new ArrayList<RedisClusterNode>();
|
||||
for (String nodeInfo : response.split("\n")) {
|
||||
String[] params = nodeInfo.split(" ");
|
||||
|
||||
String nodeId = params[0];
|
||||
|
||||
String flagsStr = params[2];
|
||||
Set<Flag> flags = EnumSet.noneOf(Flag.class);
|
||||
for (String flag : flagsStr.split(",")) {
|
||||
String flagValue = flag.toUpperCase().replaceAll("\\?", "");
|
||||
flags.add(Flag.valueOf(flagValue));
|
||||
}
|
||||
|
||||
URI address = null;
|
||||
if (!flags.contains(Flag.NOADDR)) {
|
||||
String addr = params[1].split("@")[0];
|
||||
address = URIBuilder.create("redis://" + addr);
|
||||
}
|
||||
|
||||
String masterId = params[3];
|
||||
if ("-".equals(masterId)) {
|
||||
masterId = null;
|
||||
}
|
||||
|
||||
Set<Integer> slotsCollection = new HashSet<Integer>();
|
||||
LinkState linkState = null;
|
||||
if (params.length >= 8 && params[7] != null) {
|
||||
linkState = LinkState.valueOf(params[7].toUpperCase());
|
||||
}
|
||||
if (params.length > 8) {
|
||||
for (int i = 0; i < params.length - 8; i++) {
|
||||
String slots = params[i + 8];
|
||||
if (slots.indexOf("-<-") != -1 || slots.indexOf("->-") != -1) {
|
||||
continue;
|
||||
}
|
||||
|
||||
String[] parts = slots.split("-");
|
||||
if(parts.length == 1) {
|
||||
slotsCollection.add(Integer.valueOf(parts[0]));
|
||||
} else if(parts.length == 2) {
|
||||
for (int j = Integer.valueOf(parts[0]); j < Integer.valueOf(parts[1]) + 1; j++) {
|
||||
slotsCollection.add(j);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
NodeType type = null;
|
||||
if (flags.contains(Flag.MASTER)) {
|
||||
type = NodeType.MASTER;
|
||||
} else if (flags.contains(Flag.SLAVE)) {
|
||||
type = NodeType.SLAVE;
|
||||
}
|
||||
|
||||
RedisClusterNodeBuilder builder = RedisClusterNode.newRedisClusterNode()
|
||||
.linkState(linkState)
|
||||
.slaveOf(masterId)
|
||||
.serving(new SlotRange(slotsCollection))
|
||||
.withId(nodeId)
|
||||
.promotedAs(type)
|
||||
.withFlags(flags);
|
||||
|
||||
if (address != null) {
|
||||
builder.listeningAt(address.getHost(), address.getPort());
|
||||
}
|
||||
nodes.add(builder.build());
|
||||
}
|
||||
return nodes;
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,91 @@
|
||||
/**
|
||||
* Copyright 2018 Nikita Koksharov
|
||||
*
|
||||
* 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.redisson.spring.data.connection;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collection;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
import org.redisson.client.RedisConnection;
|
||||
import org.redisson.client.codec.StringCodec;
|
||||
import org.redisson.client.protocol.RedisCommands;
|
||||
import org.springframework.data.redis.connection.NamedNode;
|
||||
import org.springframework.data.redis.connection.RedisSentinelConnection;
|
||||
import org.springframework.data.redis.connection.RedisServer;
|
||||
import org.springframework.data.redis.connection.convert.Converters;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author Nikita Koksharov
|
||||
*
|
||||
*/
|
||||
public class RedissonSentinelConnection implements RedisSentinelConnection {
|
||||
|
||||
private final RedisConnection connection;
|
||||
|
||||
public RedissonSentinelConnection(RedisConnection connection) {
|
||||
this.connection = connection;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void failover(NamedNode master) {
|
||||
connection.sync(RedisCommands.SENTINEL_FAILOVER, master.getName());
|
||||
}
|
||||
|
||||
private static List<RedisServer> toRedisServersList(List<Map<String, String>> source) {
|
||||
List<RedisServer> servers = new ArrayList<RedisServer>(source.size());
|
||||
for (Map<String, String> info : source) {
|
||||
servers.add(RedisServer.newServerFrom(Converters.toProperties(info)));
|
||||
}
|
||||
return servers;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Collection<RedisServer> masters() {
|
||||
List<Map<String, String>> masters = connection.sync(StringCodec.INSTANCE, RedisCommands.SENTINEL_MASTERS);
|
||||
return toRedisServersList(masters);
|
||||
}
|
||||
|
||||
@Override
|
||||
public Collection<RedisServer> slaves(NamedNode master) {
|
||||
List<Map<String, String>> slaves = connection.sync(StringCodec.INSTANCE, RedisCommands.SENTINEL_SLAVES, master.getName());
|
||||
return toRedisServersList(slaves);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void remove(NamedNode master) {
|
||||
connection.sync(RedisCommands.SENTINEL_REMOVE, master.getName());
|
||||
}
|
||||
|
||||
@Override
|
||||
public void monitor(RedisServer master) {
|
||||
connection.sync(RedisCommands.SENTINEL_MONITOR, master.getName(), master.getHost(),
|
||||
master.getPort().intValue(), master.getQuorum().intValue());
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() throws IOException {
|
||||
connection.closeAsync();
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean isOpen() {
|
||||
return !connection.isClosed();
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,40 @@
|
||||
/**
|
||||
* Copyright 2018 Nikita Koksharov
|
||||
*
|
||||
* 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.redisson.spring.data.connection;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
|
||||
import org.redisson.client.protocol.convertor.SingleConvertor;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author Nikita Koksharov
|
||||
*
|
||||
*/
|
||||
public class StringToListConvertor extends SingleConvertor<List<String>> {
|
||||
|
||||
@Override
|
||||
public List<String> convert(Object obj) {
|
||||
String value = (String) obj;
|
||||
List<String> result = new ArrayList<String>();
|
||||
for (String entry : value.split("\r\n|\n")) {
|
||||
result.add(entry);
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,22 @@
|
||||
package org.redisson.spring.data.connection;
|
||||
|
||||
import org.junit.Before;
|
||||
import org.redisson.BaseTest;
|
||||
import org.redisson.RedisRunner;
|
||||
import org.springframework.data.redis.connection.RedisConnection;
|
||||
import org.springframework.data.redis.connection.jedis.JedisConnection;
|
||||
|
||||
import redis.clients.jedis.Jedis;
|
||||
|
||||
public abstract class BaseConnectionTest extends BaseTest {
|
||||
|
||||
RedisConnection connection;
|
||||
|
||||
@Before
|
||||
public void init() {
|
||||
Jedis jedis = new Jedis(RedisRunner.getDefaultRedisServerBindAddressAndPort());
|
||||
connection = new JedisConnection(jedis);
|
||||
// connection = new RedissonConnection(redisson);
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,222 @@
|
||||
package org.redisson.spring.data.connection;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.Collection;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Properties;
|
||||
|
||||
import static org.assertj.core.api.Assertions.*;
|
||||
import org.junit.After;
|
||||
import org.junit.AfterClass;
|
||||
import org.junit.Before;
|
||||
import org.junit.BeforeClass;
|
||||
import org.junit.Test;
|
||||
import org.redisson.ClusterRunner;
|
||||
import org.redisson.RedisRunner;
|
||||
import org.redisson.RedisRunner.FailedToStartRedisException;
|
||||
import org.redisson.Redisson;
|
||||
import org.redisson.ClusterRunner.ClusterProcesses;
|
||||
import org.redisson.api.RedissonClient;
|
||||
import org.redisson.config.Config;
|
||||
import org.redisson.config.SubscriptionMode;
|
||||
import org.redisson.connection.MasterSlaveConnectionManager;
|
||||
import org.redisson.connection.balancer.RandomLoadBalancer;
|
||||
import org.springframework.data.redis.connection.ClusterInfo;
|
||||
import org.springframework.data.redis.connection.RedisClusterNode;
|
||||
import org.springframework.data.redis.connection.RedisNode.NodeType;
|
||||
import org.springframework.data.redis.core.types.RedisClientInfo;
|
||||
|
||||
public class RedissonClusterConnectionTest {
|
||||
|
||||
static RedissonClient redisson;
|
||||
static RedissonClusterConnection connection;
|
||||
static ClusterProcesses process;
|
||||
|
||||
@BeforeClass
|
||||
public static void before() throws FailedToStartRedisException, IOException, InterruptedException {
|
||||
RedisRunner master1 = new RedisRunner().randomPort().randomDir().nosave();
|
||||
RedisRunner master2 = new RedisRunner().randomPort().randomDir().nosave();
|
||||
RedisRunner master3 = new RedisRunner().randomPort().randomDir().nosave();
|
||||
RedisRunner slave1 = new RedisRunner().randomPort().randomDir().nosave();
|
||||
RedisRunner slave2 = new RedisRunner().randomPort().randomDir().nosave();
|
||||
RedisRunner slave3 = new RedisRunner().randomPort().randomDir().nosave();
|
||||
|
||||
|
||||
ClusterRunner clusterRunner = new ClusterRunner()
|
||||
.addNode(master1, slave1)
|
||||
.addNode(master2, slave2)
|
||||
.addNode(master3, slave3);
|
||||
process = clusterRunner.run();
|
||||
|
||||
Config config = new Config();
|
||||
config.useClusterServers()
|
||||
.setSubscriptionMode(SubscriptionMode.SLAVE)
|
||||
.setLoadBalancer(new RandomLoadBalancer())
|
||||
.addNodeAddress(process.getNodes().stream().findAny().get().getRedisServerAddressAndPort());
|
||||
|
||||
redisson = Redisson.create(config);
|
||||
connection = new RedissonClusterConnection(redisson);
|
||||
}
|
||||
|
||||
@AfterClass
|
||||
public static void after() {
|
||||
process.shutdown();
|
||||
redisson.shutdown();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testClusterGetNodes() {
|
||||
Iterable<RedisClusterNode> nodes = connection.clusterGetNodes();
|
||||
assertThat(nodes).hasSize(6);
|
||||
for (RedisClusterNode redisClusterNode : nodes) {
|
||||
assertThat(redisClusterNode.getLinkState()).isNotNull();
|
||||
assertThat(redisClusterNode.getFlags()).isNotEmpty();
|
||||
assertThat(redisClusterNode.getHost()).isNotNull();
|
||||
assertThat(redisClusterNode.getPort()).isNotNull();
|
||||
assertThat(redisClusterNode.getId()).isNotNull();
|
||||
assertThat(redisClusterNode.getType()).isNotNull();
|
||||
if (redisClusterNode.getType() == NodeType.MASTER) {
|
||||
assertThat(redisClusterNode.getSlotRange().getSlots()).isNotEmpty();
|
||||
} else {
|
||||
assertThat(redisClusterNode.getMasterId()).isNotNull();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testClusterGetNodesMaster() {
|
||||
Iterable<RedisClusterNode> nodes = connection.clusterGetNodes();
|
||||
for (RedisClusterNode redisClusterNode : nodes) {
|
||||
if (redisClusterNode.getType() == NodeType.MASTER) {
|
||||
Collection<RedisClusterNode> slaves = connection.clusterGetSlaves(redisClusterNode);
|
||||
assertThat(slaves).hasSize(1);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testClusterGetMasterSlaveMap() {
|
||||
Map<RedisClusterNode, Collection<RedisClusterNode>> map = connection.clusterGetMasterSlaveMap();
|
||||
assertThat(map).hasSize(3);
|
||||
for (Collection<RedisClusterNode> slaves : map.values()) {
|
||||
assertThat(slaves).hasSize(1);
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testClusterGetSlotForKey() {
|
||||
Integer slot = connection.clusterGetSlotForKey("123".getBytes());
|
||||
assertThat(slot).isNotNull();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testClusterGetNodeForSlot() {
|
||||
RedisClusterNode node1 = connection.clusterGetNodeForSlot(1);
|
||||
RedisClusterNode node2 = connection.clusterGetNodeForSlot(16000);
|
||||
assertThat(node1.getId()).isNotEqualTo(node2.getId());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testClusterGetNodeForKey() {
|
||||
RedisClusterNode node = connection.clusterGetNodeForKey("123".getBytes());
|
||||
assertThat(node).isNotNull();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testClusterGetClusterInfo() {
|
||||
ClusterInfo info = connection.clusterGetClusterInfo();
|
||||
assertThat(info.getSlotsFail()).isEqualTo(0);
|
||||
assertThat(info.getSlotsOk()).isEqualTo(MasterSlaveConnectionManager.MAX_SLOT);
|
||||
assertThat(info.getSlotsAssigned()).isEqualTo(MasterSlaveConnectionManager.MAX_SLOT);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testClusterAddRemoveSlots() {
|
||||
RedisClusterNode master = getFirstMaster();
|
||||
Integer slot = master.getSlotRange().getSlots().iterator().next();
|
||||
connection.clusterDeleteSlots(master, slot);
|
||||
connection.clusterAddSlots(master, slot);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testClusterCountKeysInSlot() {
|
||||
Long t = connection.clusterCountKeysInSlot(1);
|
||||
assertThat(t).isZero();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testClusterMeetForget() {
|
||||
RedisClusterNode master = getFirstMaster();
|
||||
connection.clusterForget(master);
|
||||
connection.clusterMeet(master);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testClusterGetKeysInSlot() {
|
||||
List<byte[]> keys = connection.clusterGetKeysInSlot(12, 10);
|
||||
assertThat(keys).isEmpty();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testClusterPing() {
|
||||
RedisClusterNode master = getFirstMaster();
|
||||
String res = connection.ping(master);
|
||||
assertThat(res).isEqualTo("PONG");
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testDbSize() {
|
||||
RedisClusterNode master = getFirstMaster();
|
||||
Long size = connection.dbSize(master);
|
||||
assertThat(size).isZero();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testInfo() {
|
||||
RedisClusterNode master = getFirstMaster();
|
||||
Properties info = connection.info(master);
|
||||
assertThat(info.size()).isGreaterThan(10);
|
||||
}
|
||||
|
||||
|
||||
@Test
|
||||
public void testResetConfigStats() {
|
||||
RedisClusterNode master = getFirstMaster();
|
||||
connection.resetConfigStats(master);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testTime() {
|
||||
RedisClusterNode master = getFirstMaster();
|
||||
Long time = connection.time(master);
|
||||
assertThat(time).isGreaterThan(1000);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testGetClientList() {
|
||||
RedisClusterNode master = getFirstMaster();
|
||||
List<RedisClientInfo> list = connection.getClientList(master);
|
||||
assertThat(list.size()).isGreaterThan(10);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testSetConfig() {
|
||||
RedisClusterNode master = getFirstMaster();
|
||||
connection.setConfig(master, "timeout", "10");
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testGetConfig() {
|
||||
RedisClusterNode master = getFirstMaster();
|
||||
List<String> config = connection.getConfig(master, "*");
|
||||
assertThat(config.size()).isGreaterThan(20);
|
||||
}
|
||||
|
||||
protected RedisClusterNode getFirstMaster() {
|
||||
Map<RedisClusterNode, Collection<RedisClusterNode>> map = connection.clusterGetMasterSlaveMap();
|
||||
RedisClusterNode master = map.keySet().iterator().next();
|
||||
return master;
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,26 @@
|
||||
package org.redisson.spring.data.connection;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
import org.junit.Test;
|
||||
|
||||
public class RedissonConnectionTest extends BaseConnectionTest {
|
||||
|
||||
@Test
|
||||
public void testEcho() {
|
||||
assertThat(connection.echo("test".getBytes())).isEqualTo("test".getBytes());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testSetGet() {
|
||||
connection.set("key".getBytes(), "value".getBytes());
|
||||
assertThat(connection.get("key".getBytes())).isEqualTo("value".getBytes());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testHSetGet() {
|
||||
assertThat(connection.hSet("key".getBytes(), "field".getBytes(), "value".getBytes())).isTrue();
|
||||
assertThat(connection.hGet("key".getBytes(), "field".getBytes())).isEqualTo("value".getBytes());
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,45 @@
|
||||
package org.redisson.spring.data.connection;
|
||||
|
||||
import static org.assertj.core.api.Assertions.*;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
import org.junit.Test;
|
||||
import org.redisson.BaseTest;
|
||||
|
||||
public class RedissonMultiConnectionTest extends BaseConnectionTest {
|
||||
|
||||
@Test
|
||||
public void testEcho() {
|
||||
RedissonConnection connection = new RedissonConnection(redisson);
|
||||
connection.multi();
|
||||
assertThat(connection.echo("test".getBytes())).isNull();
|
||||
assertThat(connection.exec().iterator().next()).isEqualTo("test".getBytes());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testSetGet() {
|
||||
RedissonConnection connection = new RedissonConnection(redisson);
|
||||
connection.multi();
|
||||
assertThat(connection.isQueueing()).isTrue();
|
||||
connection.set("key".getBytes(), "value".getBytes());
|
||||
assertThat(connection.get("key".getBytes())).isNull();
|
||||
|
||||
List<Object> result = connection.exec();
|
||||
assertThat(connection.isQueueing()).isFalse();
|
||||
assertThat(result.get(0)).isEqualTo("value".getBytes());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testHSetGet() {
|
||||
RedissonConnection connection = new RedissonConnection(redisson);
|
||||
connection.multi();
|
||||
assertThat(connection.hSet("key".getBytes(), "field".getBytes(), "value".getBytes())).isNull();
|
||||
assertThat(connection.hGet("key".getBytes(), "field".getBytes())).isNull();
|
||||
|
||||
List<Object> result = connection.exec();
|
||||
assertThat((Boolean)result.get(0)).isTrue();
|
||||
assertThat(result.get(1)).isEqualTo("value".getBytes());
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,46 @@
|
||||
package org.redisson.spring.data.connection;
|
||||
|
||||
import static org.assertj.core.api.Assertions.*;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
import org.junit.Test;
|
||||
import org.redisson.BaseTest;
|
||||
|
||||
public class RedissonPipelineConnectionTest extends BaseConnectionTest {
|
||||
|
||||
@Test
|
||||
public void testEcho() {
|
||||
RedissonConnection connection = new RedissonConnection(redisson);
|
||||
connection.openPipeline();
|
||||
assertThat(connection.echo("test".getBytes())).isNull();
|
||||
assertThat(connection.closePipeline().iterator().next()).isEqualTo("test".getBytes());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testSetGet() {
|
||||
RedissonConnection connection = new RedissonConnection(redisson);
|
||||
connection.openPipeline();
|
||||
assertThat(connection.isPipelined()).isTrue();
|
||||
connection.set("key".getBytes(), "value".getBytes());
|
||||
assertThat(connection.get("key".getBytes())).isNull();
|
||||
|
||||
List<Object> result = connection.closePipeline();
|
||||
assertThat(connection.isPipelined()).isFalse();
|
||||
assertThat(result.get(0)).isEqualTo("value".getBytes());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testHSetGet() {
|
||||
RedissonConnection connection = new RedissonConnection(redisson);
|
||||
connection.openPipeline();
|
||||
assertThat(connection.hSet("key".getBytes(), "field".getBytes(), "value".getBytes())).isNull();
|
||||
assertThat(connection.hGet("key".getBytes(), "field".getBytes())).isNull();
|
||||
|
||||
List<Object> result = connection.closePipeline();
|
||||
assertThat((Boolean)result.get(0)).isTrue();
|
||||
assertThat(result.get(1)).isEqualTo("value".getBytes());
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
@ -0,0 +1,132 @@
|
||||
package org.redisson.spring.data.connection;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.Collection;
|
||||
|
||||
import org.junit.After;
|
||||
import org.junit.Before;
|
||||
import org.junit.Test;
|
||||
import org.redisson.RedisRunner;
|
||||
import org.redisson.RedisRunner.FailedToStartRedisException;
|
||||
import org.redisson.Redisson;
|
||||
import org.redisson.api.RedissonClient;
|
||||
import org.redisson.config.Config;
|
||||
import org.redisson.connection.balancer.RandomLoadBalancer;
|
||||
import org.springframework.data.redis.connection.RedisSentinelConnection;
|
||||
import org.springframework.data.redis.connection.RedisServer;
|
||||
|
||||
public class RedissonSentinelConnectionTest {
|
||||
|
||||
RedissonClient redisson;
|
||||
RedisSentinelConnection connection;
|
||||
RedisRunner.RedisProcess master;
|
||||
RedisRunner.RedisProcess slave1;
|
||||
RedisRunner.RedisProcess slave2;
|
||||
RedisRunner.RedisProcess sentinel1;
|
||||
RedisRunner.RedisProcess sentinel2;
|
||||
RedisRunner.RedisProcess sentinel3;
|
||||
|
||||
@Before
|
||||
public void before() throws FailedToStartRedisException, IOException, InterruptedException {
|
||||
master = new RedisRunner()
|
||||
.nosave()
|
||||
.randomDir()
|
||||
.run();
|
||||
slave1 = new RedisRunner()
|
||||
.port(6380)
|
||||
.nosave()
|
||||
.randomDir()
|
||||
.slaveof("127.0.0.1", 6379)
|
||||
.run();
|
||||
slave2 = new RedisRunner()
|
||||
.port(6381)
|
||||
.nosave()
|
||||
.randomDir()
|
||||
.slaveof("127.0.0.1", 6379)
|
||||
.run();
|
||||
sentinel1 = new RedisRunner()
|
||||
.nosave()
|
||||
.randomDir()
|
||||
.port(26379)
|
||||
.sentinel()
|
||||
.sentinelMonitor("myMaster", "127.0.0.1", 6379, 2)
|
||||
.run();
|
||||
sentinel2 = new RedisRunner()
|
||||
.nosave()
|
||||
.randomDir()
|
||||
.port(26380)
|
||||
.sentinel()
|
||||
.sentinelMonitor("myMaster", "127.0.0.1", 6379, 2)
|
||||
.run();
|
||||
sentinel3 = new RedisRunner()
|
||||
.nosave()
|
||||
.randomDir()
|
||||
.port(26381)
|
||||
.sentinel()
|
||||
.sentinelMonitor("myMaster", "127.0.0.1", 6379, 2)
|
||||
.run();
|
||||
|
||||
Thread.sleep(5000);
|
||||
|
||||
Config config = new Config();
|
||||
config.useSentinelServers()
|
||||
.setLoadBalancer(new RandomLoadBalancer())
|
||||
.addSentinelAddress(sentinel3.getRedisServerAddressAndPort()).setMasterName("myMaster");
|
||||
redisson = Redisson.create(config);
|
||||
|
||||
RedissonConnectionFactory factory = new RedissonConnectionFactory(redisson);
|
||||
connection = factory.getSentinelConnection();
|
||||
}
|
||||
|
||||
@After
|
||||
public void after() {
|
||||
sentinel1.stop();
|
||||
sentinel2.stop();
|
||||
sentinel3.stop();
|
||||
master.stop();
|
||||
slave1.stop();
|
||||
slave2.stop();
|
||||
|
||||
redisson.shutdown();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testMasters() {
|
||||
Collection<RedisServer> masters = connection.masters();
|
||||
assertThat(masters).hasSize(1);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testSlaves() {
|
||||
Collection<RedisServer> masters = connection.masters();
|
||||
Collection<RedisServer> slaves = connection.slaves(masters.iterator().next());
|
||||
assertThat(slaves).hasSize(2);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testRemove() {
|
||||
Collection<RedisServer> masters = connection.masters();
|
||||
connection.remove(masters.iterator().next());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testMonitor() {
|
||||
Collection<RedisServer> masters = connection.masters();
|
||||
RedisServer master = masters.iterator().next();
|
||||
master.setName(master.getName() + ":");
|
||||
connection.monitor(master);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testFailover() throws InterruptedException {
|
||||
Collection<RedisServer> masters = connection.masters();
|
||||
connection.failover(masters.iterator().next());
|
||||
|
||||
Thread.sleep(10000);
|
||||
|
||||
RedisServer newMaster = connection.masters().iterator().next();
|
||||
assertThat(masters.iterator().next().getPort()).isNotEqualTo(newMaster.getPort());
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,54 @@
|
||||
package org.redisson.spring.data.connection;
|
||||
|
||||
import java.util.Arrays;
|
||||
import java.util.concurrent.atomic.AtomicReference;
|
||||
|
||||
import org.awaitility.Awaitility;
|
||||
import org.awaitility.Duration;
|
||||
import org.junit.Test;
|
||||
import org.springframework.data.redis.connection.Message;
|
||||
import org.springframework.data.redis.connection.MessageListener;
|
||||
|
||||
public class RedissonSubscribeTest extends BaseConnectionTest {
|
||||
|
||||
@Test
|
||||
public void testSubscribe() {
|
||||
RedissonConnection connection = new RedissonConnection(redisson);
|
||||
AtomicReference<byte[]> msg = new AtomicReference<byte[]>();
|
||||
connection.subscribe(new MessageListener() {
|
||||
@Override
|
||||
public void onMessage(Message message, byte[] pattern) {
|
||||
msg.set(message.getBody());
|
||||
}
|
||||
}, "test".getBytes());
|
||||
|
||||
connection.publish("test".getBytes(), "msg".getBytes());
|
||||
Awaitility.await().atMost(Duration.ONE_SECOND)
|
||||
.until(() -> Arrays.equals("msg".getBytes(), msg.get()));
|
||||
|
||||
connection.getSubscription().unsubscribe();
|
||||
|
||||
connection.publish("test".getBytes(), "msg".getBytes());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testUnSubscribe() {
|
||||
RedissonConnection connection = new RedissonConnection(redisson);
|
||||
AtomicReference<byte[]> msg = new AtomicReference<byte[]>();
|
||||
connection.subscribe(new MessageListener() {
|
||||
@Override
|
||||
public void onMessage(Message message, byte[] pattern) {
|
||||
msg.set(message.getBody());
|
||||
}
|
||||
}, "test".getBytes());
|
||||
|
||||
connection.publish("test".getBytes(), "msg".getBytes());
|
||||
Awaitility.await().atMost(Duration.ONE_SECOND)
|
||||
.until(() -> Arrays.equals("msg".getBytes(), msg.get()));
|
||||
|
||||
connection.getSubscription().unsubscribe();
|
||||
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
Loading…
Reference in New Issue