 (ns onyx.log.curator
   (:require [clojure.core.async :refer [chan >!! <!! close! thread]]
             [taoensso.timbre :refer [fatal warn trace]]
-            [onyx.static.default-vals :refer [defaults]]
-            [zookeeper :as zk]
-            [zookeeper.internal :as zi])
+            [onyx.static.default-vals :refer [defaults]])
   (:import [org.apache.zookeeper CreateMode]
-           [org.apache.zookeeper KeeperException$NoNodeException KeeperException$NodeExistsException]
+           [org.apache.zookeeper KeeperException$NoNodeException KeeperException$NodeExistsException Watcher]
            [org.apache.curator.test TestingServer]
            [org.apache.zookeeper.data Stat]
            [org.apache.curator.framework CuratorFrameworkFactory CuratorFramework]
-           [org.apache.curator.framework.api CuratorWatcher PathAndBytesable Versionable]
+           [org.apache.curator.framework.api CuratorWatcher PathAndBytesable Versionable GetDataBuilder 
+            SetDataBuilder DeleteBuilder ExistsBuilder GetChildrenBuilder Pathable Watchable]
            [org.apache.curator.framework.state ConnectionStateListener ConnectionState]
            [org.apache.curator.framework.imps CuratorFrameworkState]
            [org.apache.curator RetryPolicy]
            [org.apache.curator.retry BoundedExponentialBackoffRetry]))
+
+;; Thanks to zookeeper-clj
+(defn stat-to-map
+  ([^org.apache.zookeeper.data.Stat stat]
+   ;;(long czxid, long mzxid, long ctime, long mtime, int version, int cversion, int aversion, long ephemeralOwner, int dataLength, int numChildren, long pzxid)
+   (when stat
+     {:czxid (.getCzxid stat)
+      :mzxid (.getMzxid stat)
+      :ctime (.getCtime stat)
+      :mtime (.getMtime stat)
+      :version (.getVersion stat)
+      :cversion (.getCversion stat)
+      :aversion (.getAversion stat)
+      :ephemeralOwner (.getEphemeralOwner stat)
+      :dataLength (.getDataLength stat)
+      :numChildren (.getNumChildren stat)
+      :pzxid (.getPzxid stat)})))
+
+(defn event-to-map
+  ([^org.apache.zookeeper.WatchedEvent event]
+     (when event
+       {:event-type (keyword (.name (.getType event)))
+        :keeper-state (keyword (.name (.getState event)))
+        :path (.getPath event)})))
+
+;; Watcher
+
+(defn make-watcher
+  ([handler]
+     (reify Watcher
+       (process [this event]
+         (handler (event-to-map event))))))
 
 (defn ^CuratorFramework connect
   ([connection-string]

                 creatingParentsIfNeeded 
                 (withMode (create-mode opts)))]
      (if data 
        (.forPath cr path data)
        (.forPath cr path)))
    (catch org.apache.zookeeper.KeeperException$NodeExistsException e
      false) ))


(defn create
  [^CuratorFramework client path & {:keys [data] :as opts}]
  (try 
    (let [cr ^SetDataBuilder (.. client 
                 create 
                 (withMode (create-mode opts)))]
      (if data 
        (.forPath ^SetDataBuilder cr path data)
        (.forPath ^SetDataBuilder cr path)))
    (catch org.apache.zookeeper.KeeperException$NodeExistsException e
      false)))

