implement restore

pom.xml

@@ -15,6 +15,11 @@
             <artifactId>commons-cli</artifactId>
             <version>1.2</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-io</artifactId>
+            <version>1.3.2</version>
+        </dependency>
         <dependency>
             <groupId>org.apache.zookeeper</groupId>
             <artifactId>zookeeper</artifactId>

src/main/java/com/d2fn/guano/FSVisitor.java

@@ -0,0 +1,11 @@
+package com.d2fn.guano;
+
+import java.io.File;
+
+/**
+ * FSVisitor
+ * @author Dietrich Featherston
+ */
+public interface FSVisitor {
+    public void visit(File f, byte[] data, String znode);
+}

src/main/java/com/d2fn/guano/FSWalker.java

@@ -0,0 +1,50 @@
+package com.d2fn.guano;
+
+import org.apache.commons.io.FileUtils;
+
+import java.io.*;
+
+/**
+ * FSWalker
+ * @author Dietrich Featherston
+ */
+public class FSWalker {
+
+    private String localRoot;
+    private String znodeRoot;
+
+    public FSWalker(String localRoot, String znodeRoot) {
+        this.localRoot = localRoot;
+        this.znodeRoot = znodeRoot;
+    }
+
+    public void go(FSVisitor visitor) {
+        File f = new File(localRoot);
+        walk(visitor, f, znodeRoot);
+    }
+
+    private void walk(FSVisitor visitor, File f, String znode) {
+
+        byte[] data = new byte[0];
+
+        if(znode.endsWith("_znode")) {
+            znode = znode.substring(0, znode.length()-6);
+        }
+        if(znode.endsWith("/") && znode.length() > 1) {
+            znode = znode.substring(0, znode.length()-1);
+        }
+
+        try {
+            data = FileUtils.readFileToByteArray(f);
+        } catch (IOException e) {}
+
+        visitor.visit(f, data, znode);
+
+        if(f.isDirectory()) {
+            File[] children = f.listFiles();
+            for(File child : children) {
+                walk(visitor, child, (znode.equals("/") ? "" : znode) + "/" + child.getName());
+            }
+        }
+    }
+}

src/main/java/com/d2fn/guano/RestoreJob.java

@@ -1,15 +1,24 @@
 package com.d2fn.guano;
 
+import org.apache.zookeeper.*;
+import org.apache.zookeeper.data.ACL;
+
+import java.io.File;
+import java.util.ArrayList;
+import java.util.Arrays;
+
 /**
  * RestoreJob
  * @author Dietrich Featherston
  */
-public class RestoreJob implements Job {
+public class RestoreJob implements Job, Watcher, FSVisitor {
 
     private String zkServer;
     private String znode;
     private String inputDir;
 
+    private ZooKeeper zk;
+
     public RestoreJob(String zkServer, String znode, String inputDir) {
         this.zkServer = zkServer;
         this.znode = znode;
@@ -21,5 +30,59 @@ public void go() {
         System.out.println("zookeeper server: " + zkServer);
         System.out.println("reading from local directory: " + inputDir);
         System.out.println("restoring to zookeeper path: " + znode);
+
+        try {
+            zk = new ZooKeeper(zkServer + znode, 10000, this);
+            while(!zk.getState().isConnected()) {
+                System.out.println("connecting to " + zkServer + " with chroot " + znode);
+                Thread.sleep(1000L);
+            }
+            FSWalker walker = new FSWalker(inputDir, "/");
+            walker.go(this);
+
+        } catch (Exception e) {
+            System.err.println(e.getMessage());
+            System.exit(1);
+        }
+    }
+
+    /**
+     * on the left we have data read from a file on the local file system
+     *
+     * @param f
+     * @param data
+     * @param znode
+     */
+    @Override
+    public void visit(File f, byte[] data, String znode) {
+//        System.out.println(f.getPath());
+//        System.out.println(" -> " + znode);
+        createOrSetZnode(data, znode);
+    }
+
+    private void createOrSetZnode(byte[] data, String znode) {
+        try {
+            String s = zk.create(znode, data, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+        } catch (KeeperException.NoNodeException e) {
+            e.printStackTrace(System.err);
+        } catch (KeeperException.NodeExistsException e) {
+            try {
+                byte[] zdata = zk.getData(znode, false, null);
+                if(!Arrays.equals(data, zdata)) {
+                    zk.setData(znode, data, -1);
+                }
+            } catch (Exception e2) {
+                e2.printStackTrace(System.err);
+            }
+        } catch (KeeperException e) {
+            e.printStackTrace(System.err);
+        } catch (InterruptedException e) {
+            e.printStackTrace(System.err);
+        }
+    }
+
+    @Override
+    public void process(WatchedEvent watchedEvent) {
+        ;;
     }
 }