Java 类redis.clients.jedis.HostAndPort 实例源码

/**
 * 注意：
 * 这里返回的JedisCluster是单例的，并且可以直接注入到其他类中去使用
 * @return
 */
@Bean
public JedisCluster getJedisCluster() {
    //获取服务器数组(这里要相信自己的输入，所以没有考虑空指针问题)
    String[] serverArray = redisProperties.getClusterNodes().split(",");
    Set<HostAndPort> nodes = new HashSet<>();

    for (String ipPort : serverArray) {
        String[] ipPortPair = ipPort.split(":");
        nodes.add(new HostAndPort(ipPortPair[0].trim(), Integer.valueOf(ipPortPair[1].trim())));
    }

    return new JedisCluster(nodes,10000,
            1000,1,redisProperties.getPassword() ,
            new GenericObjectPoolConfig());
}

@Override
public Jedis getResource() {
    while (true) {
        Jedis jedis = super.getResource();
        jedis.setDataSource(this);

        // get a reference because it can change concurrently
        final HostAndPort master = currentHostMaster;
        final HostAndPort connection = new HostAndPort(jedis.getClient().getHost(), jedis.getClient()
                .getPort());

        if (master.equals(connection)) {
            // connected to the correct master
            return jedis;
        } else {
            returnBrokenResource(jedis);
        }
    }
}

private void forceFailover(JedisSentinelPool pool) throws InterruptedException {
  HostAndPort oldMaster = pool.getCurrentHostMaster();

  // jedis connection should be master
  Jedis beforeFailoverJedis = pool.getResource();
  assertEquals("PONG", beforeFailoverJedis.ping());

  waitForFailover(pool, oldMaster);

  Jedis afterFailoverJedis = pool.getResource();
  assertEquals("PONG", afterFailoverJedis.ping());
  assertEquals("foobared", afterFailoverJedis.configGet("requirepass").get(1));
  assertEquals(2, afterFailoverJedis.getDB().intValue());

  // returning both connections to the pool should not throw
  beforeFailoverJedis.close();
  afterFailoverJedis.close();
}

@Test
public void testCalculateConnectionPerSlot() {
  Set<HostAndPort> jedisClusterNode = new HashSet<HostAndPort>();
  jedisClusterNode.add(new HostAndPort("127.0.0.1", 7379));
  JedisCluster jc = new JedisCluster(jedisClusterNode, DEFAULT_TIMEOUT, DEFAULT_TIMEOUT, DEFAULT_REDIRECTIONS, "cluster", DEFAULT_CONFIG);
  jc.set("foo", "bar");
  jc.set("test", "test");
  assertEquals("bar", node3.get("foo"));
  assertEquals("test", node2.get("test"));

  JedisCluster jc2 = new JedisCluster(new HostAndPort("127.0.0.1", 7379), DEFAULT_TIMEOUT, DEFAULT_TIMEOUT, DEFAULT_REDIRECTIONS, "cluster", DEFAULT_CONFIG);
  jc2.set("foo", "bar");
  jc2.set("test", "test");
  assertEquals("bar", node3.get("foo"));
  assertEquals("test", node2.get("test"));
}

@Test
public void testStableSlotWhenMigratingNodeOrImportingNodeIsNotSpecified()
    throws InterruptedException {
  Set<HostAndPort> jedisClusterNode = new HashSet<HostAndPort>();
  jedisClusterNode.add(new HostAndPort(nodeInfo1.getHost(), nodeInfo1.getPort()));
  JedisCluster jc = new JedisCluster(jedisClusterNode, DEFAULT_TIMEOUT, DEFAULT_TIMEOUT, DEFAULT_REDIRECTIONS, "cluster", DEFAULT_CONFIG);

  int slot51 = JedisClusterCRC16.getSlot("51");
  jc.set("51", "foo");
  // node2 is responsible of taking care of slot51 (7186)

  node3.clusterSetSlotImporting(slot51, JedisClusterTestUtil.getNodeId(node2.clusterNodes()));
  assertEquals("foo", jc.get("51"));
  node3.clusterSetSlotStable(slot51);
  assertEquals("foo", jc.get("51"));

  node2.clusterSetSlotMigrating(slot51, JedisClusterTestUtil.getNodeId(node3.clusterNodes()));
  // assertEquals("foo", jc.get("51")); // it leads Max Redirections
  node2.clusterSetSlotStable(slot51);
  assertEquals("foo", jc.get("51"));
}

@Bean//bean
public JedisCluster getJedisCluster() {
    //解析字符串
    Set<HostAndPort> nodes = new HashSet<>();
    try {
        String[] cNodes = clusterNodes.split(",");
        for (String cNode : cNodes) {
            //192.168.123.1:7001
            String[] hp = cNode.split(":");
            nodes.add(new HostAndPort(hp[0], Integer.parseInt(hp[1])));

        }
        return new JedisCluster(nodes);
    } catch (ArrayIndexOutOfBoundsException | PatternSyntaxException | NullPointerException e) {
        e.printStackTrace();
    }
    return null;


}

@Bean//bean
public JedisCluster getJedisCluster() {
    //解析字符串
    Set<HostAndPort> nodes = new HashSet<>();
    try {
        String[] cNodes = clusterNodes.split(",");
        for (String cNode : cNodes) {
            //192.168.123.1:7001
            String[] hp = cNode.split(":");
            nodes.add(new HostAndPort(hp[0], Integer.parseInt(hp[1])));

        }
        return new JedisCluster(nodes);
    } catch (ArrayIndexOutOfBoundsException | PatternSyntaxException | NullPointerException e) {
        e.printStackTrace();
    }
    return null;


}

/**
 * This method is called by PDI during transformation startup. 
 *
 * @param smi   step meta interface implementation, containing the step settings
 * @param sdi   step data interface implementation, used to store runtime information
 * 
 * @return true if initialization completed successfully, false if there was an error preventing the step from working. 
 *  
 */
public boolean init(StepMetaInterface smi, StepDataInterface sdi) {
    // Casting to step-specific implementation classes is safe
    RedisOutputStepMeta meta = (RedisOutputStepMeta) smi;
    RedisOutputStepData data = (RedisOutputStepData) sdi;
       String url = environmentSubstitute(meta.getUrl());
       logBasic("creating redis session factory, addresses=" + url);
       if (url == null || url.isEmpty()) {
        throw new IllegalArgumentException("redis cluster url set configured");
    }
    String[] redisNodes = url.split(",");
    Set<HostAndPort> jedisClusterNodes = new HashSet<>();
    try {
           for (String redisNode: redisNodes) {
               String[] config = redisNode.split(":");
               jedisClusterNodes.add(new HostAndPort(config[0], Integer.parseInt(config[1])));
           }
    } catch (IllegalArgumentException ex) {
        throw new IllegalArgumentException("redis cluster url not configured correctly");
    }
       client = new JedisCluster(jedisClusterNodes, REDIS_TIMEOUT, new GenericObjectPoolConfig());
    return super.init(meta, data);
}

@Bean
public JedisCluster getJedisCluster() {
    List<String> serverArray = redisProperties.getNodes();
    Set<HostAndPort> nodes = new HashSet<>();

    for (String ipPort : serverArray) {
        String[] ipPortPair = ipPort.split(":");
        nodes.add(new HostAndPort(ipPortPair[0].trim(), Integer.valueOf(ipPortPair[1].trim())));
    }
    JedisPoolConfig config = new JedisPoolConfig();
    config.setMaxTotal(500);
    config.setMaxIdle(1000 * 60);
    config.setMaxWaitMillis(1000 * 10);
    config.setTestOnBorrow(true);
    return new JedisCluster(nodes, config);
}

/**
 * method to get jedis nodes
 * 
 * @param hosts
 * @param clusterEnabled
 * @return
 */
private Collection<? extends Serializable> getJedisNodes(String hosts, boolean clusterEnabled) {
    hosts = hosts.replaceAll("\\s", "");
    String[] hostPorts = hosts.split(",");

    List<String> node = null;
    Set<HostAndPort> nodes = null;

    for (String hostPort : hostPorts) {
        String[] hostPortArr = hostPort.split(":");

        if (clusterEnabled) {
            nodes = (nodes == null) ? new HashSet<HostAndPort>() : nodes;
            nodes.add(new HostAndPort(hostPortArr[0], Integer.valueOf(hostPortArr[1])));
        } else {
            int port = Integer.valueOf(hostPortArr[1]);
            if (!hostPortArr[0].isEmpty() && port > 0) {
                node = (node == null) ? new ArrayList<String>() : node;
                node.add(hostPortArr[0]);
                node.add(String.valueOf(port));
                break;
            }
        }
    }
    return clusterEnabled ? nodes : node;
}

private ScanPage exist(String cluster, ScanPage scanPage) throws Exception {
    scanPage.setHasMore(false);
    List<D_RedisClusterNode> nodes = clusterNodeService.getAllClusterNodes(cluster);
    Set<HostAndPort> masters = new HashSet<HostAndPort>();
    nodes.forEach(node->{
        masters.add(new HostAndPort(node.getHost(), node.getPort()));
    });
    JedisCluster jedis = new JedisCluster(masters);
    try {
        if(jedis.exists(scanPage.getQuery())){
            scanPage.setKeys(new HashSet<String>());
            scanPage.getKeys().add(scanPage.getQuery());
        }
    } finally {
        jedis.close();
    }
    return scanPage;
}

public ScanPage query(String cluster, ScanPage scanPage) throws Exception {
    D_ClusterNode_Tree tree = clusterNodeService.getClusterTree(cluster);
    Set<HostAndPort> masters = new HashSet<HostAndPort>();
    for (D_ClusterNode_Master nodes : tree.getMasters()) {//每一个分片获取一个节点
        D_RedisClusterNode node = nodes.getMaster();
        if(node.getStatus() == RedisNodeStatus.CONNECT){
            masters.add(new HostAndPort(node.getHost(), node.getPort()));
        }else{
            for (D_RedisClusterNode slave : nodes.getSlaves()) {
                if(slave.getStatus() == RedisNodeStatus.CONNECT){
                    masters.add(new HostAndPort(slave.getHost(), slave.getPort()));
                    break;
                }
            }
        }
    }
    RedisClusterScan scan = new RedisClusterScan(masters);
    scanPage.setKeys(null);
    return scan.scan(scanPage);
}

@SuppressWarnings("unchecked")
public M_ClusterSlots(List<Object> clusterSlot, Map<HostAndPort, M_clusterNode> hpNodes){
    hpSlots = new HashMap<HostAndPort, List<Slot>>();
    idSlots = new HashMap<M_clusterNode, List<Slot>>();

    for (Object object : clusterSlot) {
        List<Object> tmp = (List<Object>) object;
        Long start = (Long) tmp.get(0);
        Long end = (Long) tmp.get(1);
        for (int i = 2; i < tmp.size(); i++) {
            List<Object> nodeSlot = (List<Object>)tmp.get(i);
            HostAndPort hp = new HostAndPort(new String((byte[]) nodeSlot.get(0)), ((Long)nodeSlot.get(1)).intValue());
            M_clusterNode node = hpNodes.get(hp);

            List<Slot> slot;
            if(!hpSlots.containsKey(hp)){
                slot = new ArrayList<Slot>();
            }else{
                slot = hpSlots.get(hp);
            }
            slot.add(new Slot(start, end));
            hpSlots.put(hp, slot);
            idSlots.put(node, slot);
        }
    }
}

/**
 * 分配槽
 */
public static void assignSlot(List<HostAndPort> masters, Map<HostAndPort, RedisClusterTerminal> clients, Notify notify) {
    message(notify, ">> AssignSlot Start ... ");
    int slotSize = 16384 / masters.size();
    for (int i = 0; i < masters.size(); i++) {
        int start = i * slotSize;
        int end = (i + 1) * slotSize;
        if((i + 1) == masters.size()){
            end = 16384;
        }
        HostAndPort master = masters.get(i);
        RedisClusterTerminal cluster = clients.get(master);
        message(notify, ">> assignSlot from " + start + " to " + end + " with " + master);
        for(;start < end; start++) {
            if(start % 1000 == 0){
                message(notify, ">> " + start);
            }
            cluster.clusterAddSlots(start);
        }
    }
    message(notify, ">> assignSlot Done!");
}

/**
 * 将当前槽迁移到该节点
 */
public void reshard(int start,int end)throws Exception {
    message("start move slot from " + start + " to " + end);
    List<M_clusterNode> all = getClusterNode_list();
    Map<HostAndPort, M_clusterNode> hpNodes = new HashMap<HostAndPort, M_clusterNode>();
    M_clusterNode myself = null;
    for (M_clusterNode n : all) {
        if(n.getMaster() == null || "".equals(n.getMaster())){
            hpNodes.put(new HostAndPort(n.getHost(), n.getPort()), n);
        }
        if(n.getMyself()!= null && n.getMyself()){
            myself = n;
        }
    }
    M_ClusterSlots clusterSlots = new M_ClusterSlots(clusterSlots(), hpNodes);
    int i = start;
    while(i <= end){
        M_clusterNode sourceNode = clusterSlots.getNodeBySlot(i);
        if(sourceNode == null || !sourceNode.getNode().equals(myself.getNode())){
            message(">> move :" + i);
            moveSlot(myself, sourceNode, i);
        }
        i++;
    }
}

/**
 * 选取Slave节点
 */
private void selectSlave(Map<HostAndPort, List<HostAndPort>> tree, List<HostAndPort> list) {
    List<HostAndPort> keySets = new ArrayList<HostAndPort>(tree.keySet());
    while (list.size() > 0) {
        Collections.shuffle(list);
        Collections.shuffle(keySets);
        for (HostAndPort ipPort : keySets) {
            List<HostAndPort> slaves = tree.get(ipPort);
            List<HostAndPort> exist = new ArrayList<HostAndPort>(slaves);
            exist.add(ipPort);
            HostAndPort tmp = getOtherHost(list, exist);
            if(tmp != null){
                slaves.add(tmp);
                tree.put(ipPort, slaves);
            }
        }
    }
}

/**测试redis集群方案*/
@Test
public void testCluster(){

    Set<HostAndPort> jedisClusterNodes = new HashSet<HostAndPort>();
    //Jedis Cluster will attempt to discover cluster nodes automatically
    jedisClusterNodes.add(new HostAndPort("192.168.12.90", 7001));
    JedisCluster jc = new JedisCluster(jedisClusterNodes);
    jc.set("foo", "bar");
    String value = jc.get("foo");

    System.out.println(value);
    try {
        jc.close();
    } catch (Exception e) {
        e.printStackTrace();
    }
}

@Bean
public JedisCluster jedisCluster(JedisPoolConfig jedisPoolConfig){
    Set<HostAndPort> sets=new HashSet<>();
    for (String node:clusterNodes){
        String host=node.split(":")[0];
        int port=Integer.valueOf(node.split(":")[1]);
        sets.add(new HostAndPort(host,port));
    }
    Assert.notNull(sets, "jedisClusterNode object must not be null");
    return new JedisCluster(sets,this.clusterTimeout,
            this.clusterMaxAttempts,jedisPoolConfig);

}

public JaRedisFactory(final String host, final int port, final int connectionTimeout,
                      final int soTimeout, final String password, final int database, final String clientName) {
    this.hostAndPort.set(new HostAndPort(host, port));
    this.connectionTimeout = connectionTimeout;
    this.soTimeout = soTimeout;
    this.password = password;
    this.database = database;
    this.clientName = clientName;
}

public JaRedisFactory(final URI uri, final int connectionTimeout, final int soTimeout,
                      final String clientName) {
    if (!JedisURIHelper.isValid(uri)) {
        throw new InvalidURIException(String.format(
                "Cannot open Redis connection due invalid URI. %s", uri.toString()));
    }
    this.hostAndPort.set(new HostAndPort(uri.getHost(), uri.getPort()));
    this.connectionTimeout = connectionTimeout;
    this.soTimeout = soTimeout;
    this.password = JedisURIHelper.getPassword(uri);
    this.database = JedisURIHelper.getDBIndex(uri);
    this.clientName = clientName;
}

@Override
public PooledObject<Jedis> makeObject() throws Exception {
    final HostAndPort hostAndPort = this.hostAndPort.get();
    JaRedis.Builder builder = new JaRedis.Builder();
    builder
            .host(hostAndPort.getHost())
            .port(hostAndPort.getPort())
            .connectionTimeout(connectionTimeout)
            .soTimeout(soTimeout);
    Jedis jedis = builder.build();
    try {
        jedis.connect();
        if (null != this.password) {
            jedis.auth(this.password);
        }
        if (database != 0) {
            jedis.select(database);
        }
        if (clientName != null) {
            jedis.clientSetname(clientName);
        }
    } catch (JedisException je) {
        jedis.close();
        throw je;
    }
    return new DefaultPooledObject<>(jedis);
}

public TracingJedisCluster(Set<HostAndPort> jedisClusterNode, int connectionTimeout,
    int soTimeout,
    int maxAttempts, String password, GenericObjectPoolConfig poolConfig, Tracer tracer,
    boolean traceWithActiveSpanOnly) {
  super(jedisClusterNode, connectionTimeout, soTimeout, maxAttempts, password, poolConfig);
  this.helper = new TracingHelper(tracer, traceWithActiveSpanOnly);
}

public TracingJedisCluster(HostAndPort node, int connectionTimeout, int soTimeout,
    int maxAttempts,
    String password, GenericObjectPoolConfig poolConfig, Tracer tracer,
    boolean traceWithActiveSpanOnly) {
  super(node, connectionTimeout, soTimeout, maxAttempts, password, poolConfig);
  this.helper = new TracingHelper(tracer, traceWithActiveSpanOnly);
}

private void initPool(HostAndPort master) {
    if (!master.equals(currentHostMaster)) {
        currentHostMaster = master;
        if (factory == null) {
            factory = new JaRedisFactory(master.getHost(), master.getPort(), connectionTimeout,
                    soTimeout, password, database, clientName);
            initPool(poolConfig, factory);
        } else {
            factory.setHostAndPort(currentHostMaster);
            internalPool.clear();
        }
        log.info("Created JedisPool to master at " + master);
    }
}

@Test(expected = JedisConnectionException.class)
public void initializeWithNotAvailableSentinelsShouldThrowException() {
  Set<String> wrongSentinels = new HashSet<String>();
  wrongSentinels.add(new HostAndPort("localhost", 65432).toString());
  wrongSentinels.add(new HostAndPort("localhost", 65431).toString());

  JedisSentinelPool pool = new JedisSentinelPool(MASTER_NAME, wrongSentinels);
  pool.destroy();
}

private void waitForFailover(JedisSentinelPool pool, HostAndPort oldMaster)
    throws InterruptedException {
  HostAndPort newMaster = JedisSentinelTestUtil.waitForNewPromotedMaster(MASTER_NAME,
    sentinelJedis1, sentinelJedis2);

  waitForJedisSentinelPoolRecognizeNewMaster(pool, newMaster);
}

public static String getNodeId(String infoOutput, HostAndPort node) {

    for (String infoLine : infoOutput.split("\n")) {
      if (infoLine.contains(node.toString())) {
        return infoLine.split(" ")[0];
      }
    }
    return "";
  }

@Test
public void testMovedExceptionParameters() {
  try {
    node1.set("foo", "bar");
  } catch (JedisMovedDataException jme) {
    assertEquals(12182, jme.getSlot());
    assertEquals(new HostAndPort("127.0.0.1", 7381), jme.getTargetNode());
    return;
  }
  fail();
}

@Test
public void testRecalculateSlotsWhenMoved() throws InterruptedException {
  Set<HostAndPort> jedisClusterNode = new HashSet<HostAndPort>();
  jedisClusterNode.add(new HostAndPort("127.0.0.1", 7379));
  JedisCluster jc = new JedisCluster(jedisClusterNode, DEFAULT_TIMEOUT, DEFAULT_TIMEOUT, DEFAULT_REDIRECTIONS, "cluster", DEFAULT_CONFIG);
  int slot51 = JedisClusterCRC16.getSlot("51");
  node2.clusterDelSlots(slot51);
  node3.clusterDelSlots(slot51);
  node3.clusterAddSlots(slot51);

  JedisClusterTestUtil.waitForClusterReady(node1, node2, node3);
  jc.set("51", "foo");
  assertEquals("foo", jc.get("51"));
}

@Test
public void testAskResponse() throws InterruptedException {
  Set<HostAndPort> jedisClusterNode = new HashSet<HostAndPort>();
  jedisClusterNode.add(new HostAndPort("127.0.0.1", 7379));
  JedisCluster jc = new JedisCluster(jedisClusterNode, DEFAULT_TIMEOUT, DEFAULT_TIMEOUT, DEFAULT_REDIRECTIONS, "cluster", DEFAULT_CONFIG);
  int slot51 = JedisClusterCRC16.getSlot("51");
  node3.clusterSetSlotImporting(slot51, JedisClusterTestUtil.getNodeId(node2.clusterNodes()));
  node2.clusterSetSlotMigrating(slot51, JedisClusterTestUtil.getNodeId(node3.clusterNodes()));
  jc.set("51", "foo");
  assertEquals("foo", jc.get("51"));
}

@Test(expected = JedisClusterMaxRedirectionsException.class)
public void testRedisClusterMaxRedirections() {
  Set<HostAndPort> jedisClusterNode = new HashSet<HostAndPort>();
  jedisClusterNode.add(new HostAndPort("127.0.0.1", 7379));
  JedisCluster jc = new JedisCluster(jedisClusterNode, DEFAULT_TIMEOUT, DEFAULT_TIMEOUT, DEFAULT_REDIRECTIONS, "cluster", DEFAULT_CONFIG);
  int slot51 = JedisClusterCRC16.getSlot("51");
  // This will cause an infinite redirection loop
  node2.clusterSetSlotMigrating(slot51, JedisClusterTestUtil.getNodeId(node3.clusterNodes()));
  jc.set("51", "foo");
}

@Test
public void testClusterCountKeysInSlot() {
  Set<HostAndPort> jedisClusterNode = new HashSet<HostAndPort>();
  jedisClusterNode.add(new HostAndPort(nodeInfo1.getHost(), nodeInfo1.getPort()));
  JedisCluster jc = new JedisCluster(jedisClusterNode, DEFAULT_TIMEOUT, DEFAULT_TIMEOUT, DEFAULT_REDIRECTIONS, "cluster", DEFAULT_CONFIG);

  for (int index = 0; index < 5; index++) {
    jc.set("foo{bar}" + index, "hello");
  }

  int slot = JedisClusterCRC16.getSlot("foo{bar}");
  assertEquals(DEFAULT_REDIRECTIONS, node1.clusterCountKeysInSlot(slot).intValue());
}

@Test(expected = JedisException.class)
public void testIfPoolConfigAppliesToClusterPools() {
  GenericObjectPoolConfig config = new GenericObjectPoolConfig();
  config.setMaxTotal(0);
  config.setMaxWaitMillis(DEFAULT_TIMEOUT);
  Set<HostAndPort> jedisClusterNode = new HashSet<HostAndPort>();
  jedisClusterNode.add(new HostAndPort("127.0.0.1", 7379));
  JedisCluster jc = new JedisCluster(jedisClusterNode, DEFAULT_TIMEOUT, DEFAULT_TIMEOUT, DEFAULT_REDIRECTIONS, "cluster", config);
  jc.set("52", "poolTestValue");
}

@Test
public void testJedisClusterTimeout() {
  Set<HostAndPort> jedisClusterNode = new HashSet<HostAndPort>();
  jedisClusterNode.add(new HostAndPort(nodeInfo1.getHost(), nodeInfo1.getPort()));

  JedisCluster jc = new JedisCluster(jedisClusterNode, 4000, 4000, DEFAULT_REDIRECTIONS, "cluster", DEFAULT_CONFIG);

  for (JedisPool pool : jc.getClusterNodes().values()) {
    Jedis jedis = pool.getResource();
    assertEquals(jedis.getClient().getConnectionTimeout(), 4000);
    assertEquals(jedis.getClient().getSoTimeout(), 4000);
    jedis.close();
  }

}

@Test
public void testJedisClusterRunsWithMultithreaded() throws InterruptedException,
    ExecutionException, IOException {
  Set<HostAndPort> jedisClusterNode = new HashSet<HostAndPort>();
  jedisClusterNode.add(new HostAndPort("127.0.0.1", 7379));
  final JedisCluster jc = new JedisCluster(jedisClusterNode, DEFAULT_TIMEOUT, DEFAULT_TIMEOUT, DEFAULT_REDIRECTIONS, "cluster", DEFAULT_CONFIG);
  jc.set("foo", "bar");

  ThreadPoolExecutor executor = new ThreadPoolExecutor(10, 100, 0, TimeUnit.SECONDS,
      new ArrayBlockingQueue<Runnable>(10));
  List<Future<String>> futures = new ArrayList<Future<String>>();
  for (int i = 0; i < 50; i++) {
    executor.submit(new Callable<String>() {
      @Override
      public String call() throws Exception {
        // FIXME : invalidate slot cache from JedisCluster to test
        // random connection also does work
        return jc.get("foo");
      }
    });
  }

  for (Future<String> future : futures) {
    String value = future.get();
    assertEquals("bar", value);
  }

  jc.close();
}

@Test(timeout = DEFAULT_TIMEOUT)
public void testReturnConnectionOnJedisConnectionException() throws InterruptedException {
  Set<HostAndPort> jedisClusterNode = new HashSet<HostAndPort>();
  jedisClusterNode.add(new HostAndPort("127.0.0.1", 7379));
  JedisPoolConfig config = DEFAULT_CONFIG;
  config.setMaxTotal(1);
  JedisCluster jc = new JedisCluster(jedisClusterNode, DEFAULT_TIMEOUT, DEFAULT_TIMEOUT, DEFAULT_REDIRECTIONS, "cluster", config);

  Jedis j = jc.getClusterNodes().get("127.0.0.1:7380").getResource();
  ClientKillerUtil.tagClient(j, "DEAD");
  ClientKillerUtil.killClient(j, "DEAD");
  j.close();

  jc.get("test");
}

@Test(expected = JedisClusterMaxRedirectionsException.class, timeout = DEFAULT_TIMEOUT)
public void testReturnConnectionOnRedirection() {
  Set<HostAndPort> jedisClusterNode = new HashSet<HostAndPort>();
  jedisClusterNode.add(new HostAndPort("127.0.0.1", 7379));
  JedisPoolConfig config = DEFAULT_CONFIG;
  config.setMaxTotal(1);
  JedisCluster jc = new JedisCluster(jedisClusterNode, 0, 2, DEFAULT_REDIRECTIONS, "cluster", config);

  // This will cause an infinite redirection between node 2 and 3
  node3.clusterSetSlotMigrating(15363, JedisClusterTestUtil.getNodeId(node2.clusterNodes()));
  jc.get("e");
}

@Test
public void testLocalhostNodeNotAddedWhen127Present() {
  HostAndPort localhost = new HostAndPort("localhost", 7379);
  Set<HostAndPort> jedisClusterNode = new HashSet<HostAndPort>();
  // cluster node is defined as 127.0.0.1; adding localhost should work,
  // but shouldn't show up.
  jedisClusterNode.add(localhost);
  JedisPoolConfig config = DEFAULT_CONFIG;
  config.setMaxTotal(1);
  JedisCluster jc = new JedisCluster(jedisClusterNode, 0, 2, DEFAULT_REDIRECTIONS, "cluster", DEFAULT_CONFIG);
  Map<String, JedisPool> clusterNodes = jc.getClusterNodes();
  assertEquals(3, clusterNodes.size());
  assertFalse(clusterNodes.containsKey(JedisClusterInfoCache.getNodeKey(localhost)));
}

@Test
public void testInvalidStartNodeNotAdded() {
  HostAndPort invalidHost = new HostAndPort("not-a-real-host", 7379);
  Set<HostAndPort> jedisClusterNode = new HashSet<HostAndPort>();
  jedisClusterNode.add(new HostAndPort("127.0.0.1", 7379));
  jedisClusterNode.add(invalidHost);
  JedisPoolConfig config = DEFAULT_CONFIG;
  config.setMaxTotal(1);
  JedisCluster jc = new JedisCluster(jedisClusterNode,  0, 2, DEFAULT_REDIRECTIONS, "cluster", config);
  Map<String, JedisPool> clusterNodes = jc.getClusterNodes();
  assertEquals(3, clusterNodes.size());
  assertFalse(clusterNodes.containsKey(JedisClusterInfoCache.getNodeKey(invalidHost)));
}

@Test
public void sentinel() {
  Jedis j = new Jedis(sentinel.getHost(), sentinel.getPort());

  try {
    List<Map<String, String>> masters = j.sentinelMasters();

    boolean inMasters = false;
    for (Map<String, String> master : masters)
      if (MASTER_NAME.equals(master.get("name"))) inMasters = true;

    assertTrue(inMasters);

    List<String> masterHostAndPort = j.sentinelGetMasterAddrByName(MASTER_NAME);
    HostAndPort masterFromSentinel = new HostAndPort(masterHostAndPort.get(0),
        Integer.parseInt(masterHostAndPort.get(1)));
    assertEquals(master, masterFromSentinel);

    List<Map<String, String>> slaves = j.sentinelSlaves(MASTER_NAME);
    assertTrue(!slaves.isEmpty());
    assertEquals(master.getPort(), Integer.parseInt(slaves.get(0).get("master-port")));

    // DO NOT RE-RUN TEST TOO FAST, RESET TAKES SOME TIME TO... RESET
    assertEquals(Long.valueOf(1), j.sentinelReset(MASTER_NAME));
    assertEquals(Long.valueOf(0), j.sentinelReset("woof" + MASTER_NAME));
  } finally {
    j.close();
  }
}