Fixed issues with protected mode and persisten ephemeral

diff --git a/CHANGES.txt b/CHANGES.txt
index 17a5933..34655bc 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt

@@ -22,6 +22,9 @@
 
 * Issue 222: Counter and log messages reversed in RetryLoop.takeException().
 
+* Issue 201: Added PersistentEphemeralNode recipe. Thanks to user bbeck for the idea, initial design
+and most of the test cases. Details here:
+
 1.2.5 - November 27, 2012
 =========================
 * Depend on ZooKeeper 3.4.5

diff --git a/curator-framework/src/main/java/com/netflix/curator/framework/api/CreateModable.java b/curator-framework/src/main/java/com/netflix/curator/framework/api/CreateModable.java
index 02db6c9..d00a510 100644
--- a/curator-framework/src/main/java/com/netflix/curator/framework/api/CreateModable.java
+++ b/curator-framework/src/main/java/com/netflix/curator/framework/api/CreateModable.java

@@ -28,4 +28,13 @@
      * @return this
      */
     public T withMode(CreateMode mode);
+
+    /**
+     * If create builder is in {@link CreateBuilder#withProtection()} mode this returns
+     * the prefix that will be added to the node name. If not in protection mode the
+     * return result is undefined.
+     *
+     * @return prefix
+     */
+    public String getProtectedPrefix();
 }

diff --git a/curator-framework/src/main/java/com/netflix/curator/framework/imps/CreateBuilderImpl.java b/curator-framework/src/main/java/com/netflix/curator/framework/imps/CreateBuilderImpl.java
index 25905ea..0746938 100644
--- a/curator-framework/src/main/java/com/netflix/curator/framework/imps/CreateBuilderImpl.java
+++ b/curator-framework/src/main/java/com/netflix/curator/framework/imps/CreateBuilderImpl.java

@@ -98,6 +98,12 @@
                 transaction.add(Op.create(fixedPath, data, acling.getAclList(path), createMode), OperationType.CREATE, path);
                 return curatorTransaction;
             }
+
+            @Override
+            public String getProtectedPrefix()
+            {
+                return CreateBuilderImpl.this.getProtectedPrefix();
+            }
         };
     }
 
@@ -179,6 +185,12 @@
             {
                 return CreateBuilderImpl.this.forPath(path);
             }
+
+            @Override
+            public String getProtectedPrefix()
+            {
+                return CreateBuilderImpl.this.getProtectedPrefix();
+            }
         };
     }
 
@@ -315,6 +327,12 @@
             {
                 return CreateBuilderImpl.this.forPath(path);
             }
+
+            @Override
+            public String getProtectedPrefix()
+            {
+                return CreateBuilderImpl.this.getProtectedPrefix();
+            }
         };
     }
 
@@ -462,6 +480,12 @@
         );
     }
 
+    @Override
+    public String getProtectedPrefix()
+    {
+        return PROTECTED_PREFIX + protectedId + "-";
+    }
+
     private void backgroundCreateParentsThenNode(final OperationAndData<PathAndBytes> mainOperationAndData)
     {
         BackgroundOperation<PathAndBytes>     operation = new BackgroundOperation<PathAndBytes>()
@@ -536,6 +560,12 @@
             }
 
             @Override
+            public String getProtectedPrefix()
+            {
+                return CreateBuilderImpl.this.getProtectedPrefix();
+            }
+
+            @Override
             public String forPath(String path, byte[] data) throws Exception
             {
                 return CreateBuilderImpl.this.forPath(path, data);
@@ -642,7 +672,7 @@
                 }
             }
         );
-        
+
         trace.commit();
         return returnPath;
     }
@@ -707,9 +737,4 @@
         }
         return path;
     }
-
-    private String getProtectedPrefix() throws Exception
-    {
-        return PROTECTED_PREFIX + protectedId + "-";
-    }
 }

diff --git a/curator-recipes/src/main/java/com/netflix/curator/framework/recipes/nodes/PersistentEphemeralNode.java b/curator-recipes/src/main/java/com/netflix/curator/framework/recipes/nodes/PersistentEphemeralNode.java
index ff160ed..77ae765 100644
--- a/curator-recipes/src/main/java/com/netflix/curator/framework/recipes/nodes/PersistentEphemeralNode.java
+++ b/curator-recipes/src/main/java/com/netflix/curator/framework/recipes/nodes/PersistentEphemeralNode.java

@@ -24,7 +24,6 @@
 import com.netflix.curator.framework.api.BackgroundCallback;
 import com.netflix.curator.framework.api.CreateModable;
 import com.netflix.curator.framework.api.CuratorEvent;
-import com.netflix.curator.framework.api.PathAndBytesable;
 import com.netflix.curator.framework.state.ConnectionState;
 import com.netflix.curator.framework.state.ConnectionStateListener;
 import com.netflix.curator.utils.EnsurePath;
@@ -328,7 +327,7 @@
         try
         {
             String      existingPath = nodePath.get();
-            String      createPath = (existingPath != null) ? existingPath : basePath;
+            String      createPath = (existingPath != null) ? fixExistingPath(existingPath) : basePath;
             ensurePath.ensure(client.getZookeeperClient());
             createMethod.withMode(mode.getCreateMode(existingPath != null)).inBackground(backgroundCallback).forPath(createPath, data);
         }
@@ -338,6 +337,21 @@
         }
     }
 
+    private String fixExistingPath(String path)
+    {
+        if ( mode.isProtected() )
+        {
+            ZKPaths.PathAndNode pathAndNode = ZKPaths.getPathAndNode(path);
+            String              protectedPrefix = createMethod.getProtectedPrefix();
+            int                 index = pathAndNode.getNode().indexOf(protectedPrefix);
+            if ( index >= 0 )
+            {
+                return ZKPaths.makePath(pathAndNode.getPath(), pathAndNode.getNode().substring(index + protectedPrefix.length()));
+            }
+        }
+        return path;
+    }
+
     private void watchNode()
     {
         if ( !isActive() )