#include <string.h>
#include <errno.h>

#include "zookeeper.h"

static zhandle_t *zh;

/**
 * In this example this method gets the cert for your
 *   environment -- you must provide
 */
char *foo_get_cert_once(char* id) { return 0; }

/** Watcher function -- empty for this example, not something you should
 * do in real code */
void watcher(zhandle_t *zzh, int type, int state, const char *path,
      void *watcherCtx) {}

int main(int argc, char argv) {
  char buffer[512];
  char p[2048];
  char *cert=0;
  char appId[64];

  strcpy(appId, "example.foo_test");
  cert = foo_get_cert_once(appId);
  if(cert!=0) {
      fprintf(stderr,
              "Certificate for appid [%s] is [%s]\n",appId,cert);
      strncpy(p,cert, sizeof(p)-1);
      free(cert);
  } else {
      fprintf(stderr, "Certificate for appid [%s] not found\n",appId);
      strcpy(p, "dummy");
  }

  zoo_set_debug_level(ZOO_LOG_LEVEL_DEBUG);

  zh = zookeeper_init("localhost:3181", watcher, 10000, 0, 0, 0);
  if (!zh) {
      return errno;
  }
  if(zoo_add_auth(zh,"foo",p,strlen(p),0,0)!=ZOK)
      return 2;

  struct ACL CREATE_ONLY_ACL[] = ;
  struct ACL_vector CREATE_ONLY = {1, CREATE_ONLY_ACL};
  int rc = zoo_create(zh,"/xyz","value", 5, &CREATE_ONLY, ZOO_EPHEMERAL,
          buffer, sizeof(buffer)-1);

  /** this operation will fail with a ZNOAUTH error */
  int buflen= sizeof(buffer);
  struct Stat stat;
  rc = zoo_get(zh, "/xyz", 0, buffer, &buflen, &stat);
  if (rc) {
      fprintf(stderr, "Error %d for %s\n", rc, __LINE__);
  }

  zookeeper_close(zh);
  return 0;
}

/**
 * A simple example program to use DataMonitor to start and
 * stop executables based on a znode. The program watches the
 * specified znode and saves the data that corresponds to the
 * znode in the filesystem. It also starts the specified program
 * with the specified arguments when the znode exists and kills
 * the program if the znode goes away.
 */
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;

import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;

public class Executor
implements Watcher, Runnable, DataMonitor.DataMonitorListener
{
  String znode;

  DataMonitor dm;

  ZooKeeper zk;

  String filename;

  String exec[];

  Process child;

  public Executor(String hostPort, String znode, String filename,
          String exec[]) throws KeeperException, IOException {
      this.filename = filename;
      this.exec = exec;
      zk = new ZooKeeper(hostPort, 3000, this);
      dm = new DataMonitor(zk, znode, null, this);
  }

  /**
  * @param args
  */
  public static void main(String[] args) {
      if (args.length < 4) {
          System.err
              .println("USAGE: Executor hostPort znode filename program [args ...]");
          System.exit(2);
      }
      String hostPort = args[0];
      String znode = args[1];
      String filename = args[2];
      String exec[] = new String[args.length - 3];
      System.arraycopy(args, 3, exec, 0, exec.length);
      try {
          new Executor(hostPort, znode, filename, exec).run();
      } catch (Exception e) {
          e.printStackTrace();
      }
  }

  /***************************************************************************
  * We do process any events ourselves, we just need to forward them on.
  *
  * @see org.apache.zookeeper.Watcher#process(org.apache.zookeeper.proto.WatcherEvent)
  */
  public void process(WatchedEvent event) {
      dm.process(event);
  }

  public void run() {
      try {
          synchronized (this) {
              while (!dm.dead) {
                  wait();
              }
          }
      } catch (InterruptedException e) {
      }
  }

  public void closing(int rc) {
      synchronized (this) {
          notifyAll();
      }
  }

  static class StreamWriter extends Thread {
      OutputStream os;

      InputStream is;

      StreamWriter(InputStream is, OutputStream os) {
          this.is = is;
          this.os = os;
          start();
      }

      public void run() {
          byte b[] = new byte[80];
          int rc;
          try {
              while ((rc = is.read(b)) > 0) {
                  os.write(b, 0, rc);
              }
          } catch (IOException e) {
          }

      }
  }

  public void exists(byte[] data) {
      if (data == null) {
          if (child != null) {
              System.out.println("Killing process");
              child.destroy();
              try {
                  child.waitFor();
              } catch (InterruptedException e) {
              }
          }
          child = null;
      } else {
          if (child != null) {
              System.out.println("Stopping child");
              child.destroy();
              try {
                  child.waitFor();
              } catch (InterruptedException e) {
                  e.printStackTrace();
              }
          }
          try {
              FileOutputStream fos = new FileOutputStream(filename);
              fos.write(data);
              fos.close();
          } catch (IOException e) {
              e.printStackTrace();
          }
          try {
              System.out.println("Starting child");
              child = Runtime.getRuntime().exec(exec);
              new StreamWriter(child.getInputStream(), System.out);
              new StreamWriter(child.getErrorStream(), System.err);
          } catch (IOException e) {
              e.printStackTrace();
          }
      }
  }
}

/**
 * A simple class that monitors the data and existence of a ZooKeeper
 * node. It uses asynchronous ZooKeeper APIs.
 */
import java.util.Arrays;

import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.AsyncCallback.StatCallback;
import org.apache.zookeeper.KeeperException.Code;
import org.apache.zookeeper.data.Stat;

public class DataMonitor implements Watcher, StatCallback {

  ZooKeeper zk;

  String znode;

  Watcher chainedWatcher;

  boolean dead;

  DataMonitorListener listener;

  byte prevData[];

  public DataMonitor(ZooKeeper zk, String znode, Watcher chainedWatcher,
          DataMonitorListener listener) {
      this.zk = zk;
      this.znode = znode;
      this.chainedWatcher = chainedWatcher;
      this.listener = listener;
      // Get things started by checking if the node exists. We are going
      // to be completely event driven
      zk.exists(znode, true, this, null);
  }

  /**
  * Other classes use the DataMonitor by implementing this method
  */
  public interface DataMonitorListener {
      /**
      * The existence status of the node has changed.
      */
      void exists(byte data[]);

      /**
      * The ZooKeeper session is no longer valid.
      *
      * @param rc
      *                the ZooKeeper reason code
      */
      void closing(int rc);
  }

  public void process(WatchedEvent event) {
      String path = event.getPath();
      if (event.getType() == Event.EventType.None) {
          // We are are being told that the state of the
          // connection has changed
          switch (event.getState()) {
              case SyncConnected:
                  // In this particular example we don't need to do anything
                  // here - watches are automatically re-registered with 
                  // server and any watches triggered while the client was 
                  // disconnected will be delivered (in order of course)
                  break;
              case Expired:
                  // It's all over
                  dead = true;
                  listener.closing(KeeperException.Code.SessionExpired);
                  break;
          }
      } else {
          if (path != null && path.equals(znode)) {
              // Something has changed on the node, let's find out
              zk.exists(znode, true, this, null);
          }
      }
      if (chainedWatcher != null) {
          chainedWatcher.process(event);
      }
  }

  public void processResult(int rc, String path, Object ctx, Stat stat) {
      boolean exists;
      switch (rc) {
          case Code.Ok:
              exists = true;
              break;
          case Code.NoNode:
              exists = false;
              break;
          case Code.SessionExpired:
          case Code.NoAuth:
              dead = true;
              listener.closing(rc);
              return;
          default:
              // Retry errors
              zk.exists(znode, true, this, null);
              return;
      }

      byte b[] = null;
      if (exists) {
          try {
              b = zk.getData(znode, false, null);
          } catch (KeeperException e) {
              // We don't need to worry about recovering now. The watch
              // callbacks will kick off any exception handling
              e.printStackTrace();
          } catch (InterruptedException e) {
              return;
          }
      }
      if ((b == null && b != prevData)
              || (b != null && !Arrays.equals(prevData, b))) {
          listener.exists(b);
          prevData = b;
      }
  }
}