package org.jgroups.tests;

import org.jgroups.Global;
import org.jgroups.JChannel;
import org.jgroups.View;
import org.jgroups.protocols.*;
import org.jgroups.protocols.pbcast.GMS;
import org.jgroups.protocols.pbcast.NAKACK2;
import org.jgroups.protocols.pbcast.STABLE;
import org.jgroups.stack.GossipRouter;
import org.jgroups.util.ResourceManager;
import org.jgroups.util.Util;
import org.testng.annotations.AfterClass;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;

import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.List;

/**
 * Tests the TCPGOSSIP protocol
 * 
 * @author Vladimir Blagojevic
 * 
 **/
@Test(groups = { Global.STACK_INDEPENDENT, Global.GOSSIP_ROUTER, Global.EAP_EXCLUDED }, singleThreaded = true)
public class TCPGOSSIP_Test {
    private JChannel            channel, coordinator;
    private final static String GROUP = "TCPGOSSIP_Test";
    private GossipRouter        gossipRouter;
    private int                 gossip_router_port;
    protected InetAddress       bind_addr;


    @BeforeClass
    void startRouter() throws Exception {
        bind_addr=Util.getLoopback();
        gossip_router_port=ResourceManager.getNextTcpPort(bind_addr);
        gossipRouter=new GossipRouter(bind_addr, gossip_router_port);
        gossipRouter.start();
    }


    @AfterClass(alwaysRun = true)
    void stopRouter() throws Exception {
        gossipRouter.stop();
    }

    @AfterMethod(alwaysRun = true)
    void tearDown() throws Exception {
        Util.close(channel, coordinator);
    }

    /**
     * Tests connect-disconnect-connect sequence for a group with two members (using the default configuration).
     **/
    public void testDisconnectConnectTwo() throws Exception {
        coordinator=createTcpgossipChannel("A");
        channel=createTcpgossipChannel("B");
        coordinator.connect(GROUP);
        channel.connect("DisconnectTest.testgroup-1");
        channel.disconnect();
        channel.connect(GROUP);
        View view = channel.getView();
        assert view.size() == 2;
        assert view.containsMember(channel.getAddress());
        assert view.containsMember(coordinator.getAddress());
    }
    
    public void testAddInitialHosts() throws Exception {
        coordinator=createTcpgossipChannel("A");
        channel=createTcpgossipChannel("B");
        coordinator.connect(GROUP);
        channel.connect(GROUP);
        TCPGOSSIP p =channel.getProtocolStack().findProtocol(TCPGOSSIP.class);
        String tmp_bind_addr = bind_addr.getHostAddress();
        assert p.removeInitialHost(tmp_bind_addr, gossip_router_port);
        p.addInitialHost(tmp_bind_addr, gossip_router_port);
       
        View view = channel.getView();
        assert view.size() == 2;
        assert view.containsMember(channel.getAddress());
        assert view.containsMember(coordinator.getAddress());
    }

    public void testConnectThree() throws Exception {
        JChannel third = null;
        try {
            coordinator=createTcpgossipChannel("A");
            channel=createTcpgossipChannel("B");
            coordinator.connect(GROUP);
            channel.connect(GROUP);
            third=createTcpgossipChannel("C");
            third.connect(GROUP);
            Util.waitUntilAllChannelsHaveSameView(10000, 500, coordinator,channel,third);
            View view = channel.getView();
            assert channel.getView().size() == 3
              : String.format("expected view size of 3 but got %d: %s", channel.getView().size(), channel.getView());
            assert third.getView().size() == 3;
            assert view.containsMember(channel.getAddress());
            assert view.containsMember(coordinator.getAddress());
        } finally {
            Util.close(third);
        }
    }

    public void testConnectThreeChannelsWithGRDown() throws Exception {
        JChannel third = null;
        try {
            coordinator=createTcpgossipChannel("A");
            channel=createTcpgossipChannel("B");
            coordinator.connect("testConnectThreeChannelsWithGRDown");
            channel.connect("testConnectThreeChannelsWithGRDown");

            // kill router
            gossipRouter.stop();
            

            // cannot discover others since GR is down
            third=createTcpgossipChannel("C");
            third.connect("testConnectThreeChannelsWithGRDown");

            // restart and....
            gossipRouter.start();
            Util.waitUntilAllChannelsHaveSameView(60000, 1000, coordinator, channel, third);

            // confirm they found each other
            View view = channel.getView();
            assert channel.getView().size() == 3;
            assert third.getView().size() == 3;
            assert view.containsMember(channel.getAddress());
            assert view.containsMember(coordinator.getAddress());
        } finally {
            Util.close(third);
        }
    }

    public void testConnectThreeChannelsWithGRAlreadyDown() throws Exception {
        JChannel third = null;
        try {
            coordinator=createTcpgossipChannel("A");
            channel=createTcpgossipChannel("B");

            // kill router
            gossipRouter.stop();
            
            // cannot discover others since GR is down
            coordinator.connect("testConnectThreeChannelsWithGRAlreadyDown");          
            channel.connect("testConnectThreeChannelsWithGRAlreadyDown");

            third=createTcpgossipChannel("C");
            third.connect("testConnectThreeChannelsWithGRAlreadyDown");

            // restart and....
            gossipRouter.start();
            Util.waitUntilAllChannelsHaveSameView(60000, 1000, coordinator, channel, third);

            // confirm they found each other
            View view = channel.getView();
            assert channel.getView().size() == 3;
            assert third.getView().size() == 3;
            assert view.containsMember(channel.getAddress());
            assert view.containsMember(coordinator.getAddress());
        } finally {
            Util.close(third);
        }
    }


    protected JChannel createTcpgossipChannel(String name) throws Exception {
        TCPGOSSIP gossip=new TCPGOSSIP();
        List<InetSocketAddress> initial_hosts=new ArrayList<>();
        initial_hosts.add(new InetSocketAddress(bind_addr, gossip_router_port));
        gossip.setInitialHosts(initial_hosts);

        JChannel ch=new JChannel(new TCP().setSockConnTimeout(300).setBindAddress(bind_addr),
                                 gossip,
                                 new MERGE3().setMinInterval(1000).setMaxInterval(3000),
                                 new FD_ALL3().setTimeout(5000).setInterval(1500),
                                 new VERIFY_SUSPECT(),
                                 new NAKACK2().useMcastXmit(false),
                                 new UNICAST3(), new STABLE(), new GMS().setJoinTimeout(1000));
        if(name != null)
            ch.setName(name);
        return ch;
    }



    protected static void changeGossipRouter(JChannel channel, String host, int port) {
        TCPGOSSIP tcp_gossip_prot=channel.getProtocolStack().findProtocol(TCPGOSSIP.class);
        List<InetSocketAddress> initial_hosts=tcp_gossip_prot.getInitialHosts();
        initial_hosts.clear();
        initial_hosts.add(new InetSocketAddress(host, port));
    }


}