/**
 *
 */
package com.github.zkclient;

import com.github.zkclient.exception.ZkNoNodeException;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.data.Stat;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.Before;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.LockSupport;

import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;

/**
 * @author adyliu (imxylz@gmail.com)
 * @since 2012-12-3
 */
public class ZkClientTest {
    static {
        System.setProperty("zookeeper.preAllocSize", "1024");// 1M data log
    }

    final Logger logger = LoggerFactory.getLogger(ZkClientTest.class);
    private static final AtomicInteger counter = new AtomicInteger();
    //
    private ZkServer server;
    private ZkClient client;
    final int TIMEOUT = 30;//30 second for loop timeout

    //
    private static void deleteFile(File f) throws IOException {
        if (f.isFile()) {
            f.delete();
            //System.out.println("[DELETE FILE] "+f.getPath());
        } else if (f.isDirectory()) {
            File[] files = f.listFiles();
            if (files != null) {
                for (File fs : files) {
                    deleteFile(fs);
                }
            }
            f.delete();
            //System.out.println("[DELETE DIRECTORY] "+f.getPath());
        }
    }

    private static ZkServer startZkServer(String testName, int port) throws IOException {
        String dataPath = "build/test/" + testName + "/data";
        String logPath = "build/test/" + testName + "/log";
        File dataDir = new File(".", dataPath).getCanonicalFile();
        File logDir = new File(".", logPath).getCanonicalFile();
        deleteFile(dataDir);
        deleteFile(logDir);
        //start the server with 100ms session timeout
        ZkServer zkServer = new ZkServer(dataDir.getPath(), logDir.getPath(), port, ZkServer.DEFAULT_TICK_TIME, 100);
        zkServer.start();
        return zkServer;
    }

    @AfterClass
    public static void cleanup() throws IOException {
       deleteFile(new File(".", "build/test").getCanonicalFile());
    }

    @Before
    public void setUp() throws Exception {
        this.server = startZkServer("server_" + counter.incrementAndGet(), 4711);
        this.client = this.server.getZkClient();
        assertTrue(this.client.isConnected());
    }

    @After
    public void tearDown() throws Exception {
        if (this.server != null) {
            this.server.shutdown();
        }
    }

    /**
     * Test method for
     * {@link com.github.zkclient.ZkClient#subscribeChildChanges(java.lang.String, com.github.zkclient.IZkChildListener)}
     * .
     */
    @Test
    public void testSubscribeChildChanges() throws Exception {
        final String path = "/a";
        final AtomicInteger count = new AtomicInteger(0);
        final ArrayList<String> children = new ArrayList<String>();
        IZkChildListener listener = new IZkChildListener() {
            public void handleChildChange(String parentPath, List<String> currentChildren) throws Exception {
                count.incrementAndGet();
                children.clear();
                if (currentChildren != null)
                    children.addAll(currentChildren);
                logger.info("handle childchange " + parentPath + ", " + currentChildren);
            }
        };
        //
        client.subscribeChildChanges(path, listener);
        //
        logger.info("create the watcher node " + path);
        client.createPersistent(path);
        //wait some time to make sure the event was triggered
        TestUtil.waitUntil(1, new Callable<Integer>() {
            @Override
            public Integer call() throws Exception {
                return count.get();
            }
        }, TimeUnit.SECONDS, TIMEOUT);
        //
        assertEquals(1, count.get());
        assertEquals(0, children.size());
        //
        //create a child node
        count.set(0);
        client.createPersistent(path + "/child1");
        logger.info("create the first child node " + path + "/child1");
        TestUtil.waitUntil(1, new Callable<Integer>() {
            @Override
            public Integer call() throws Exception {
                return count.get();
            }
        }, TimeUnit.SECONDS, TIMEOUT);
        //
        assertEquals(1, count.get());
        assertEquals(1, children.size());
        assertEquals("child1", children.get(0));
        //
        // create another child node and delete the node
        count.set(0);
        logger.info("create the second child node " + path + "/child2");
        client.createPersistent(path + "/child2");
        //
        logger.info("delete the watcher node " + path);
        client.deleteRecursive(path);
        //
        Boolean eventReceived = TestUtil.waitUntil(true, new Callable<Boolean>() {
            @Override
            public Boolean call() throws Exception {
                return count.get() > 0 && children.size() == 0;
            }
        }, TimeUnit.SECONDS, TIMEOUT);
        assertTrue(eventReceived);
        assertEquals(0, children.size());
        // ===========================================
        // do it again and check the listener validate
        // ===========================================
        count.set(0);
        //
        logger.info("create the watcher node again " + path);
        client.createPersistent(path);
        //
        eventReceived = TestUtil.waitUntil(true, new Callable<Boolean>() {
            @Override
            public Boolean call() throws Exception {
                return count.get() > 0;
            }
        }, TimeUnit.SECONDS, TIMEOUT);
        assertTrue(eventReceived);
        assertEquals(0, children.size());
        //
        // now create the first node
        count.set(0);
        final String child3 = "/child3";
        client.createPersistent(path + child3);
        logger.info("create the first child node again " + path + child3);
        //
        eventReceived = TestUtil.waitUntil(true, new Callable<Boolean>() {
            @Override
            public Boolean call() throws Exception {
                return count.get() > 0;
            }
        }, TimeUnit.SECONDS, 15);
        assertTrue(eventReceived);
        assertEquals(1, children.size());
        assertEquals("child3", children.get(0));
        //
        // delete root node 
        count.set(0);
        logger.info("delete the watcher node again " + path);
        client.deleteRecursive(path);
        // This will receive two message: (1) child was deleted (2) parent was deleted
        //
        eventReceived = TestUtil.waitUntil(true, new Callable<Boolean>() {
            @Override
            public Boolean call() throws Exception {
                return children.isEmpty();
            }
        }, TimeUnit.SECONDS, TIMEOUT);
        assertTrue(eventReceived);
        assertTrue(children.isEmpty());
    }

    static class Holder<T> {
        T t;

        public void set(T t) {
            this.t = t;
        }

        public T get() {
            return t;
        }
    }

    static byte[] toBytes(String s) {
        try {
            return s != null ? s.getBytes("UTF-8") : null;
        } catch (UnsupportedEncodingException e) {
            throw new RuntimeException(e);
        }
    }

    static String toString(byte[] b) {
        try {
            return b != null ? new String(b, "UTF-8") : null;
        } catch (UnsupportedEncodingException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * Test method for
     * {@link com.github.zkclient.ZkClient#subscribeDataChanges(java.lang.String, com.github.zkclient.IZkDataListener)}
     * .
     */
    @Test
    public void testSubscribeDataChanges() throws Exception {
        String path = "/a";
        final AtomicInteger countChanged = new AtomicInteger(0);
        final AtomicInteger countDeleted = new AtomicInteger(0);
        final Holder<String> holder = new Holder<String>();
        IZkDataListener listener = new IZkDataListener() {
            public void handleDataDeleted(String dataPath) throws Exception {
                countDeleted.incrementAndGet();
                holder.set(null);
            }

            public void handleDataChange(String dataPath, byte[] data) throws Exception {
                countChanged.incrementAndGet();
                holder.set(ZkClientTest.toString(data));
            }
        };
        client.subscribeDataChanges(path, listener);
        //
        // create the node
        client.createPersistent(path, toBytes("aaa"));
        //
        //wait some time to make sure the event was triggered
        TestUtil.waitUntil(1, new Callable<Integer>() {
            @Override
            public Integer call() throws Exception {
                return countChanged.get();
            }
        }, TimeUnit.SECONDS, TIMEOUT);
        assertEquals(1, countChanged.get());
        assertEquals(0, countDeleted.get());
        assertEquals("aaa", holder.get());
        //
        countChanged.set(0);
        countDeleted.set(0);
        //
        client.delete(path);
        TestUtil.waitUntil(1, new Callable<Integer>() {
            @Override
            public Integer call() throws Exception {
                return countDeleted.get();
            }
        }, TimeUnit.SECONDS, TIMEOUT);
        assertEquals(0, countChanged.get());
        assertEquals(1, countDeleted.get());
        assertNull(holder.get());
        // ===========================================
        // do it again and check the listener validate
        // ===========================================
        countChanged.set(0);
        countDeleted.set(0);
        client.createPersistent(path, toBytes("bbb"));
        TestUtil.waitUntil(1, new Callable<Integer>() {
            @Override
            public Integer call() throws Exception {
                return countChanged.get();
            }
        }, TimeUnit.SECONDS, TIMEOUT);
        assertEquals(1, countChanged.get());
        assertEquals("bbb", holder.get());
        //
        countChanged.set(0);
        client.writeData(path, toBytes("ccc"));
        //
        TestUtil.waitUntil(1, new Callable<Integer>() {
            @Override
            public Integer call() throws Exception {
                return countChanged.get();
            }
        }, TimeUnit.SECONDS, TIMEOUT);
        assertEquals(1, countChanged.get());
        assertEquals("ccc", holder.get());
    }


    /**
     * Test method for
     * {@link com.github.zkclient.ZkClient#createPersistent(java.lang.String, boolean)}
     * .
     */
    @Test
    public void testCreatePersistent() {
        final String path = "/a/b";
        try {
            client.createPersistent(path, false);
            fail("should throw exception");
        } catch (ZkNoNodeException e) {
            assertFalse(client.exists(path));
        }
        client.createPersistent(path, true);
        assertTrue(client.exists(path));
    }

    /**
     * Test method for
     * {@link com.github.zkclient.ZkClient#createPersistent(java.lang.String, byte[])}
     * .
     */
    @Test
    public void testCreatePersistentStringByteArray() {
        String path = "/a";
        client.createPersistent(path, toBytes("abc"));
        assertEquals("abc", toString(client.readData(path)));
        //
    }

    /**
     * Test method for
     * {@link com.github.zkclient.ZkClient#createPersistentSequential(java.lang.String, byte[])}
     * .
     */
    @Test
    public void testCreatePersistentSequential() {
        String path = "/a";
        String npath = client.createPersistentSequential(path, toBytes("abc"));
        assertTrue(npath != null && npath.length() > 0);
        npath = client.createPersistentSequential(path, toBytes("abc"));
        assertEquals("abc", toString(client.readData(npath)));
    }

    /**
     * Test method for
     * {@link com.github.zkclient.ZkClient#createEphemeral(java.lang.String)}.
     */
    @Test
    public void testCreateEphemeralString() {
        String path = "/a";
        client.createEphemeral(path);
        Stat stat = new Stat();
        client.readData(path, stat);
        assertTrue(stat.getEphemeralOwner() > 0);
    }

    /**
     * Test method for
     * {@link com.github.zkclient.ZkClient#create(java.lang.String, byte[], org.apache.zookeeper.CreateMode)}
     * .
     */
    @Test
    public void testCreate() {
        String path = "/a";
        client.create(path, toBytes("abc"), CreateMode.PERSISTENT);
        assertEquals("abc", toString(client.readData(path)));
    }


    @Test
    public void testCreateEphemeralSequential() {
        String path = "/a";
        String npath = client.createEphemeralSequential(path, toBytes("abc"));
        assertTrue(npath != null && npath.startsWith("/a"));
        Stat stat = new Stat();
        assertArrayEquals(toBytes("abc"), client.readData(npath, stat));
        assertTrue(stat.getEphemeralOwner() > 0);
    }


    /**
     * Test method for
     * {@link com.github.zkclient.ZkClient#getChildren(java.lang.String)}.
     */
    @Test
    public void testGetChildrenString() {
        String path = "/a";
        client.createPersistent(path + "/ch1", true);
        client.createPersistent(path + "/ch2");
        client.createPersistent(path + "/ch3");
        List<String> children = client.getChildren(path);
        assertEquals(3, children.size());
        assertEquals(3, client.countChildren(path));
        assertNull(client.getChildren("/aaa"));
    }


    /**
     * Test method for
     * {@link com.github.zkclient.ZkClient#exists(java.lang.String)}.
     */
    @Test
    public void testExistsString() {
        String path = "/a";
        assertFalse(client.exists(path));
        client.createPersistent(path);
        assertTrue(client.exists(path));
        client.delete(path);
        assertFalse(client.exists(path));
    }

    /**
     * Test method for
     * {@link com.github.zkclient.ZkClient#deleteRecursive(java.lang.String)}.
     */
    @Test
    public void testDeleteRecursive() {
        String path = "/a/b/c";
        client.createPersistent(path, true);
        assertTrue(client.exists(path));
        assertTrue(client.deleteRecursive("/a"));
        assertFalse(client.exists(path));
        assertFalse(client.exists("/a/b"));
        assertFalse(client.exists("/a"));
    }

    /**
     * Test method for
     * {@link com.github.zkclient.ZkClient#waitUntilExists(java.lang.String, java.util.concurrent.TimeUnit, long)}
     * .
     */
    @Test
    public void testWaitUntilExists() {
        final String path = "/a";
        new Thread() {
            public void run() {
                LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(100));
                client.createPersistent(path);
            }
        }.start();
        assertTrue(client.waitUntilExists(path, TimeUnit.SECONDS, 10));
        assertTrue(client.exists(path));
        //
        assertFalse(client.waitUntilExists("/notexists", TimeUnit.SECONDS, 1));
    }

    /**
     * Test method for {@link com.github.zkclient.ZkClient#waitUntilConnected()}
     * .
     */
    @Test
    public void testWaitUntilConnected() {
        ZkClient client2 = new ZkClient("localhost:4711", 15000);
        assertTrue(client2.waitUntilConnected());
        server.shutdown();
        //
        assertTrue(client2.waitForKeeperState(KeeperState.Disconnected, 1, TimeUnit.SECONDS));
        //
        assertFalse(client2.waitUntilConnected(1, TimeUnit.SECONDS));
        client2.close();
    }


    /**
     * Test method for
     * {@link com.github.zkclient.ZkClient#readData(java.lang.String, org.apache.zookeeper.data.Stat)}
     * .
     */
    @Test
    public void testReadDataStringStat() {
        client.createPersistent("/a", "data".getBytes());
        Stat stat = new Stat();
        client.readData("/a", stat);
        assertEquals(0, stat.getVersion());
        assertTrue(stat.getDataLength() > 0);
    }


    /**
     * Test method for {@link com.github.zkclient.ZkClient#numberOfListeners()}.
     */
    @Test
    public void testNumberOfListeners() {
        IZkChildListener zkChildListener = new AbstractListener() {
        };
        client.subscribeChildChanges("/", zkChildListener);
        assertEquals(1, client.numberOfListeners());
        //
        IZkDataListener zkDataListener = new AbstractListener() {
        };
        client.subscribeDataChanges("/a", zkDataListener);
        assertEquals(2, client.numberOfListeners());
        //
        client.subscribeDataChanges("/b", zkDataListener);
        assertEquals(3, client.numberOfListeners());
        //
        IZkStateListener zkStateListener = new AbstractListener() {
        };
        client.subscribeStateChanges(zkStateListener);
        assertEquals(4, client.numberOfListeners());
        //
        client.unsubscribeChildChanges("/", zkChildListener);
        assertEquals(3, client.numberOfListeners());
        //
        client.unsubscribeAll();
        assertEquals(0, client.numberOfListeners());
    }

    /**
     * Test method for {@link com.github.zkclient.ZkClient#getZooKeeper()}.
     */
    @Test
    public void testGetZooKeeper() {
        assertTrue(client.getZooKeeper() != null);
    }

    @Test
    public void testRetryUnitConnected_SessionExpiredException() {
        int sessionTimeout = 200;
        int port = PortUtils.checkAvailablePort(4712);
        int dport = this.server.getPort();
        Gateway gateway = new Gateway(port, dport);
        gateway.start();
        //
        final ZkClient client2 = new ZkClient("localhost:" + port, sessionTimeout, 15000);
        gateway.stop();
        //
        //start the server after 600ms
        new DeferredGatewayStarter(gateway, sessionTimeout * 3).start();
        //
        final Boolean connected = client2.retryUntilConnected(new Callable<Boolean>() {
            @Override
            public Boolean call() throws Exception {
                client2.createPersistent("/abc");
                return Boolean.TRUE;
            }
        });
        assertTrue(connected);
        assertTrue(client2.exists("/abc"));
        client2.close();
        gateway.stop();
        //
    }


    @Test
    public void testChildListenerAfterSessionExpiredException() throws Exception {
        final int sessionTimeout = 200;
        ZkClient connectedClient = server.getZkClient();
        connectedClient.createPersistent("/root");
        //
        int port = PortUtils.checkAvailablePort(4712);
        int dport = this.server.getPort();
        Gateway gateway = new Gateway(port, dport);
        gateway.start();
        //
        final ZkClient disconnectedClient = new ZkClient("localhost:" + port, sessionTimeout, 15000);
        final Holder<List<String>> children = new Holder<List<String>>();
        disconnectedClient.subscribeChildChanges("/root", new IZkChildListener() {

            @Override
            public void handleChildChange(String parentPath, List<String> currentChildren) throws Exception {
                children.set(currentChildren);
            }
        });
        gateway.stop();//
        //
        // the connected client created a new child node
        connectedClient.createPersistent("/root/node1");
        //
        // wait for 3x sessionTImeout, the session should have expired
        Thread.sleep(3 * sessionTimeout);
        //
        // now start the gateway
        gateway.start();
        //
        Boolean hasOneChild = TestUtil.waitUntil(true, new Callable<Boolean>() {
            @Override
            public Boolean call() throws Exception {
                return children.get() != null && children.get().size() == 1;
            }
        }, TimeUnit.SECONDS, TIMEOUT);
        //
        assertTrue(hasOneChild);
        assertEquals("node1", children.get().get(0));
        assertEquals("node1", disconnectedClient.getChildren("/root").get(0));
        //
        disconnectedClient.close();
        gateway.stop();
    }

}
