
package org.apache.hedwig.zookeeper;

import java.io.IOException;
import java.util.List;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.zookeeper.AsyncCallback;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.KeeperException.Code;
import org.apache.zookeeper.data.ACL;

import org.apache.hedwig.util.PathUtils;

public class ZkUtils {

    static Logger logger = LoggerFactory.getLogger(ZkUtils.class);

    static class SyncObject {
        int rc;
        String path;
        boolean called = false;
    }

    public static void createFullPathOptimistic(final ZooKeeper zk, final String originalPath, final byte[] data,
            final List<ACL> acl, final CreateMode createMode)
    throws KeeperException, IOException, InterruptedException {
        final SyncObject syncObj = new SyncObject();

        createFullPathOptimistic(
            zk, originalPath, data, acl, createMode,
            new SafeAsyncZKCallback.StringCallback() {
                @Override
                public void safeProcessResult(final int rc, String path, Object ctx, String name) {
                    synchronized (syncObj) {
                        syncObj.rc = rc;
                        syncObj.path = path;
                        syncObj.called = true;
                        syncObj.notify();
                    }
                }
            }, syncObj
        );

        synchronized (syncObj) {
            while (!syncObj.called) {
                syncObj.wait();
            }
        }

        if (Code.OK.intValue() != syncObj.rc) {
            throw KeeperException.create(syncObj.rc, syncObj.path);
        }
    }

    public static void createFullPathOptimistic(final ZooKeeper zk, final String originalPath, final byte[] data,
            final List<ACL> acl, final CreateMode createMode, final AsyncCallback.StringCallback callback,
            final Object ctx) {

        zk.create(originalPath, data, acl, createMode, new SafeAsyncZKCallback.StringCallback() {
            @Override
            public void safeProcessResult(int rc, String path, Object ctx, String name) {

                if (rc != Code.NONODE.intValue()) {
                    callback.processResult(rc, path, ctx, name);
                    return;
                }

                // Since I got a nonode, it means that my parents don't exist
                // create mode is persistent since ephemeral nodes can't be
                // parents
                ZkUtils.createFullPathOptimistic(zk, PathUtils.parent(originalPath), new byte[0], acl,
                CreateMode.PERSISTENT, new SafeAsyncZKCallback.StringCallback() {

                    @Override
                    public void safeProcessResult(int rc, String path, Object ctx, String name) {
                        if (rc == Code.OK.intValue() || rc == Code.NODEEXISTS.intValue()) {
                            // succeeded in creating the parent, now
                            // create the original path
                            ZkUtils.createFullPathOptimistic(zk, originalPath, data, acl, createMode, callback,
                                                             ctx);
                        } else {
                            callback.processResult(rc, path, ctx, name);
                        }
                    }
                }, ctx);
            }
        }, ctx);

    }

    public static KeeperException logErrorAndCreateZKException(String msg, String path, int rc) {
        KeeperException ke = KeeperException.create(Code.get(rc), path);
        logger.error(msg + ",zkPath: " + path, ke);
        return ke;
    }

}
