@Test
public void peerConnectTest() throws InterruptedException {

    final String localhost = "127.0.0.1";
    final ConstNodeDiscovery discovery = new ConstNodeDiscovery(
            new NodeAddress(localhost,8000),
            new NodeAddress(localhost,8001),
            new NodeAddress(localhost,8002));
    final JoinCountClusterListener listenerA = new JoinCountClusterListener();
    final JoinCountClusterListener listenerB = new JoinCountClusterListener();
    final JoinCountClusterListener listenerC = new JoinCountClusterListener();
    final ClusterClient repA = new ClusterClient(listenerA,localhost, 8000, discovery, 2000, Executors.newSingleThreadScheduledExecutor());
    final ClusterClient repB = new ClusterClient(listenerB,localhost, 8001, discovery, 2000, Executors.newSingleThreadScheduledExecutor());
    final ClusterClient repC = new ClusterClient(listenerC,localhost, 8002, discovery, 2000, Executors.newSingleThreadScheduledExecutor());
    try {
        // spin until 3 nodes are found
        for (int i = 0; i < 10; i++) {
            Thread.sleep(500);
            if (repA.getNodeCount() >= 3
                    && repB.getNodeCount() >= 3
                    && repC.getNodeCount() >= 3) {
                break;
            }
        }
        Assert.assertEquals(3, repA.getNodeCount());
        Assert.assertEquals(3, repB.getNodeCount());
        Assert.assertEquals(3, repC.getNodeCount());
        // spin until 3 nodes are process in the callback
        for (int i = 0; i < 10; i++) {
            Thread.sleep(500);
            if (listenerA.getJoinCount() == 3
                    && listenerB.getJoinCount() == 3
                    && listenerC.getJoinCount() == 3) {
                break;
            }
        }
        Assert.assertEquals(3, listenerA.getJoinCount());
        Assert.assertEquals(3, listenerB.getJoinCount());
        Assert.assertEquals(3, listenerC.getJoinCount());
        // send message to all nodes
        for (NodeId nid : listenerA.getNodes()) {
            repA.sendMessage(nid, new PingRequest());
        }
    } finally {
        repA.shutdown();
        repB.shutdown();
        repC.shutdown();
    }
}