ZooKeeper distributed lock samples

package com.apogado.test.distlock.test;

import java.util.Arrays;
import java.util.Collections;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 *
 * @author Gabriel
 */
public class LockTest {

    private static final Logger logger = LoggerFactory.getLogger(LockTest.class);
    public static final String CONN_STRING = "localhost:2181";
    ZooKeeper zookeeperServer;
    final Watcher watcher = new TestWatcher();

    @org.junit.Before
    public void setup() {
        try {
            synchronized (watcher) {
                this.zookeeperServer = new ZooKeeper(CONN_STRING, 3000, watcher);
                watcher.wait(3000);
            }
            ZooKeeper.States state = this.zookeeperServer.getState();
            logger.info("state: {}", this.zookeeperServer.getState());
            org.junit.Assert.assertEquals(ZooKeeper.States.CONNECTED, state);
        } catch (Exception ex) {
            logger.error("setup", ex);
            org.junit.Assert.fail(ex.toString());
        }

    }

    @org.junit.After
    public void cleanup() {
        try {
            if (this.zookeeperServer != null) {
                this.zookeeperServer.close();
            }
        } catch (Exception ex) {
            logger.warn("cleanup", ex);
        }
    }

    @org.junit.Test
    public void testLock() {
        try {
            String parentNode = "/lock";
            String lockNodePath = "/lock/test";
            logger.info("Node lock path: {}", lockNodePath);

            Stat stat;
            stat = this.zookeeperServer.exists("/cxf-locator", false);
            org.junit.Assert.assertNotNull(stat);
            logger.info("stat immediate: {}", stat);


            try {
                stat = this.zookeeperServer.exists("/nonexisting", false);
                org.junit.Assert.assertNull(stat);
            } catch (KeeperException ex) {
                logger.info("Node doesn't exist: " + ex.getMessage());
            }

            // create parent lock
            logger.info("Checking for the parent node");
            stat = this.zookeeperServer.exists(parentNode, false);
            if (stat == null) {
                logger.info("Parent node doesn't exist, creating one");
                this.zookeeperServer.create(parentNode, null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            } else {
                logger.info("Parent node exists");
            }

            // create a lock node
            String path = this.zookeeperServer.create(lockNodePath, null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
            logger.info("Created lock path: {}", path);

            // check the lock is created
            stat = this.zookeeperServer.exists(lockNodePath, false);
            org.junit.Assert.assertNotNull(stat);
            logger.info("Lock data: {}", stat.toString());

//            logger.info("Sleeping to allow manual check");
//            Thread.sleep(10000);

            // add node watcher
            TestWatcher nodeWatcher = new TestWatcher();
            stat = this.zookeeperServer.exists(lockNodePath, nodeWatcher);

            // set data
            this.zookeeperServer.setData(lockNodePath, "stop".getBytes(), stat.getVersion());
            synchronized (nodeWatcher) {
                logger.info("Waiting to be notified");
                nodeWatcher.wait(1000);
                logger.info("Notified: {}", nodeWatcher.isNotified());
                org.junit.Assert.assertTrue(nodeWatcher.isNotified());
            }

            // delete lock
            try {
                this.zookeeperServer.delete(path, -1);
                stat = this.zookeeperServer.exists(lockNodePath, false);
                org.junit.Assert.assertNull(stat);
                logger.info("lock node deleted");
            } catch (KeeperException ke) {
                logger.error("failed to delete a lock node", ke);
                org.junit.Assert.fail(ke.toString());
            }

        } catch (Exception ex) {
            logger.warn("testLock: ", ex);
            org.junit.Assert.fail(ex.toString());
        }
    }

    public class TestWatcher implements Watcher {

        private boolean notified;

        public void process(WatchedEvent we) {
            logger.info("watcher event: {}", we.toString());
            synchronized (this) {
                setNotified(true);
                this.notify();
            }
        }

        public boolean isNotified() {
            return notified;
        }

        public void setNotified(boolean notified) {
            this.notified = notified;
        }
    }
}