ZOOKEEPER-809. Improved REST Interface

git-svn-id: https://svn.apache.org/repos/asf/hadoop/zookeeper/trunk@986215 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -105,6 +105,8 @@ IMPROVEMENTS:
 
   ZOOKEEPER-765.  Add python example script (Travis and Andrei via mahadev)
 
+  ZOOKEEPER-809. Improved REST Interface (Andrei Savu via phunt)
+
 NEW FEATURES:
   ZOOKEEPER-729. Java client API to recursively delete a subtree.
   (Kay Kay via henry)

src/contrib/rest/README.txt

@@ -1,19 +1,24 @@
+
 ZooKeeper REST implementation using Jersey JAX-RS.
+--------------------------------------------------
 
-This is an implementation of version 1 of the ZooKeeper REST spec.
+This is an implementation of version 2 of the ZooKeeper REST spec.
 
 Note: This interface is currently experimental, may change at any time,
 etc... In general you should be using the Java/C client bindings to access
 the ZooKeeper server.
 
+This REST ZooKeeper gateway is useful because most of the languages
+have built-in support for working with HTTP based protocols.
+
 See SPEC.txt for details on the REST binding.
 
------------
 Quickstart:
+-----------
 
 1) start a zookeeper server on localhost port 2181
 
-2) run "ant runrestserver"
+2) run "ant run"
 
 3) use a REST client to access the data (see below for more details)
 
@@ -24,14 +29,14 @@ or use the provided src/python scripts
   zk_dump_tree.py
 
 
-----------
 Tests:
+----------
 
 1) the full testsuite can be run via "ant test" target
+2) the python client library also contains a test suite
 
-
-----------
 Examples Using CURL
+-------------------
 
 First review the spec SPEC.txt in this directory.
 
@@ -52,4 +57,16 @@ curl -T data.txt -w "\n%{http_code}\n" "http://localhost:9998/znodes/v1/cluster1
 
 #create a node
 curl -d "data1" -H'Content-Type: application/octet-stream' -w "\n%{http_code}\n" "http://localhost:9998/znodes/v1/?op=create&name=cluster2&dataformat=utf8"
+
 curl -d "data2" -H'Content-Type: application/octet-stream' -w "\n%{http_code}\n" "http://localhost:9998/znodes/v1/cluster2?op=create&name=leader&dataformat=utf8"
+
+#create a new session
+curl -d "" -H'Content-Type: application/octet-stream' -w "\n%{http_code}\n" "http://localhost:9998/sessions/v1/?op=create&expire=10"
+
+#session heartbeat
+curl -X "PUT" -H'Content-Type: application/octet-stream' -w "\n%{http_code}\n" "http://localhost:9998/sessions/v1/02dfdcc8-8667-4e53-a6f8-ca5c2b495a72"
+
+#delete a session
+curl -X "DELETE" -H'Content-Type: application/octet-stream' -w "\n%{http_code}\n" "http://localhost:9998/sessions/v1/02dfdcc8-8667-4e53-a6f8-ca5c2b495a72"
+
+

src/contrib/rest/SPEC.txt

@@ -13,11 +13,12 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-An HTTP gateway for ZooKeeper
+A REST HTTP gateway for ZooKeeper
+=================================
 
-Specification Version: 1
+Specification Version: 2
 
-ZooKeeper is a meant to enable distributed coordination and also store
+ZooKeeper is meant to enable distributed coordination and also store
 system configuration and other relatively small amounts of information
 that must be stored in a persistent and consistent manner. The
 information stored in ZooKeeper is meant to be highly available to a
@@ -41,14 +42,16 @@ would be a less cumbersome interface than the ZooKeeper library.
 This document describes a gatway for using HTTP to interact with a
 ZooKeeper repository.
 
-
 Binding ZooKeeper to HTTP
+-------------------------
 
 Encoding
+--------
 
 UTF-8 unless otherwise noted
 
 Paths
+-----
 
 A ZooKeeper paths are mapped to IRIs and URIs as follows. ZK paths
 are converted to IRIs by simply percent-encoding any characters in the
@@ -67,6 +70,7 @@ designers, but both recommend that application designers use NFC as a
 best practice.
 
 Root
+----
 
 The following examples assume that the ZooKeeper znode heirarchy is
 bound to the root of the HTTP servers namespace. This may not be the
@@ -77,10 +81,11 @@ example the URL for accessing /a/b/c may be:
 
 This is perfectly valid. Users of the REST service should be aware of
 this fact and code their clients to support any root (in this case
-"/zookeeper/znodes/v1" on the server localhost).
+"/zookeeper" on the server localhost).
 
 
 Basics: GET, PUT, HEAD, and DELETE
+----------------------------------
 
 HTTP's GET, PUT, HEAD, and DELETE operations map naturally to
 ZooKeeper's "get," "set," "exists," and "delete" operations.
@@ -96,6 +101,7 @@ cycles. Set/delete requests may include an optional parameter
 
 
 Getting ZooKeeper children
+--------------------------
 
 We overload the GET method to return the children of a ZooKeeper. In
 particular, the GET method takes an optional parameter "view" which
@@ -103,7 +109,7 @@ could be set to one of type values, either "data" or "children". The
 default is "data". Thus, to get the children of a znode named
 "/a/b/c", then the GET request should start:
 
-  GET /a/b/c?view=children HTTP/1.1
+  GET /znodes/v1/a/b/c?view=children HTTP/1.1
 
 If the requested view is "data", then the data of a znode is returned
 as described in the previous section. If the requested view is
@@ -112,13 +118,45 @@ document, or in a JSON object. (The default is JSON, but this can be
 controlled changed by setting the Accept header.)
 
 
+Creating a ZooKeeper session
+----------------------------
+
+In order to be able to create ephemeral nodes you first need to start
+a new session.
+
+  POST /sessions/v1?op=create&expire=<SECONDS> HTTP/1.1
+
+If the session creation is successful, then a 201 code will be returned.
+
+A session is just an UUID that you can pass around as a parameter and
+the REST server will foward your request on the attached persistent 
+connection.
+
+Keeping a session alive
+-----------------------
+
+To keep a session alive you must send hearbeat requests:
+
+  PUT /sessions/v1/<SESSION-UUID> HTTP/1.1
+
+Closing a ZooKeeper session
+---------------------------
+
+You can close a connection by sending a DELETE request.
+
+  DELETE /sessions/v1/<SESSION-UUID> HTTP/1.1
+
+If you don't close a session it will automatically expire after
+the amount of time you specified on creation. 
+
 Creating a ZooKeeper znode
+--------------------------
 
 We use the POST method to create a ZooKeeper znode. For example, to
 create a znode named "c" under a parent named "/a/b", then the POST
 request should start:
 
-  POST /a/b?op=create&name=c HTTP/1.1
+  POST /znodes/v1/a/b?op=create&name=c HTTP/1.1
 
 If the creation is successful, then a 201 code will be returned. If
 it fails, then a number of different codes might be returned
@@ -138,12 +176,17 @@ be provided explicitly if desired.)
 
 On success the actual path of the created znode will be returned.
 
+If you want to create an ephemeral node you need to specify an
+additional "ephemeral=true" parameter. (Note that "ephemeral" is an optional
+parameter, that defaults to "false")
+
 (Note: ZooKeeper also allows the client to set ACLs for the
 newly-created znode. This feature is not currently supported by the
 HTTP gateway to ZooKeeper.)
 
 
 Content types and negotiation
+-----------------------------
 
 ZooKeeper REST gateway implementations may support three content-types
 for request and response messages:
@@ -192,6 +235,10 @@ PATH
   uri is the full URI of the znode as seen by the REST server, does not
   include any query parameters (i.e. it's the path to the REST resource)
 
+SESSION
+  id : string UUID
+  uri : string
+
 CHILD
   PATH
   child_uri_template: string
@@ -225,6 +272,7 @@ STAT
 
 
 Error Codes
+-----------
 
 The ZooKeeper gateway uses HTTP response codes as follows:
 
@@ -249,6 +297,7 @@ server, the resulting Web Server might behave differently, e.g., it
 might do redirection, check other headers, etc.
 
 Error Messages
+--------------
 
 Error messages are returned to the caller, format is dependent on the
 format requested in the call. 
@@ -272,6 +321,7 @@ format requested in the call.
 
 
 Binding ZooKeeper to an HTTP server
+-----------------------------------
 
 It might be sage to assume that everyone is happy to run an Apache
 server, and thus write a "mod_zookeeper" for Apache that works only

src/contrib/rest/build.xml

@@ -34,6 +34,20 @@
     <property name="test.junit.haltonfailure" value="no" />
     <property name="test.junit.maxmem" value="512m" />
 
+    <!-- ====================================================== -->
+    <!-- Macro definitions                                      -->
+    <!-- ====================================================== -->
+    <macrodef name="macro_tar" description="Worker Macro for tar">
+      <attribute name="param.destfile"/>
+      <element name="param.listofitems"/>
+      <sequential>
+        <tar compression="gzip" longfile="gnu"
+             destfile="@{param.destfile}">
+          <param.listofitems/>
+        </tar>
+      </sequential>
+    </macrodef>
+
   <target name="setjarname">
     <property name="jarname"
               value="${build.dir}/zookeeper-${version}-${name}.jar"/>
@@ -120,16 +134,14 @@
     </jar>
   </target>
 
-  <target name="runrestserver" depends="jar">
+  <target name="run" depends="jar">
     <echo message="contrib: ${name}"/>
     <java classname="org.apache.zookeeper.server.jersey.RestMain" fork="true">
-      <arg value="http://localhost:9998/" />
-      <arg value="localhost:2181" />
       <classpath>
         <pathelement path="${jarname}" />
         <fileset dir="${build.dir}/lib" includes="*.jar"/>
         <fileset dir="${zk.root}/build" includes="zookeeper-*.jar"/>
-        <pathelement path="${zk.root}/conf" />
+        <pathelement path="${zk.root}/src/contrib/${name}/conf" />
         <fileset dir="${zk.root}/src/java/lib">
           <include name="**/*.jar" />
         </fileset>
@@ -137,5 +149,19 @@
     </java>
   </target>
 
+  <target name="tar" depends="clean, jar">
+    <echo message="building tar.gz: ${name}" />
+    <macro_tar param.destfile="${build.dir}/zookeeper-${version}-${name}.tar.gz">
+      <param.listofitems>
+        <tarfileset dir="${build.dir}/lib" prefix="lib" includes="**/*.jar" />
+        <tarfileset file="${build.dir}/zookeeper-*-rest.jar" />
+        <tarfileset dir="${zk.root}/build" includes="zookeeper-*.jar" prefix="lib" />
+        <tarfileset dir="${zk.root}/src/contrib/${name}/conf" prefix="conf" />
+        <tarfileset dir="${zk.root}/src/java/lib" prefix="lib" includes="**/*.jar" />
+        <tarfileset file="${zk.root}/src/contrib/${name}/rest.sh" />
+      </param.listofitems>
+    </macro_tar>
+  </target>
+
 </project>
 

src/contrib/rest/conf/keys/README

@@ -0,0 +1,8 @@
+
+In order to generate .jks (java keystore files) you need to use keytool.
+
+The password for the existing .jks is "123456" (without quotes).
+
+Some tutorials:
+ - http://www.mobilefish.com/tutorials/java/java_quickguide_keytool.html
+

src/contrib/rest/conf/rest.properties

@@ -0,0 +1,70 @@
+#
+# 
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+# 
+#   http://www.apache.org/licenses/LICENSE-2.0
+# 
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+# 
+#
+
+#
+# ZooKeeper REST Gateway Configuration file
+#
+
+rest.port = 9998
+
+#
+# Endpoint definition
+#
+
+# plain configuration <context-path>;<host-port>
+rest.endpoint.1 = /;localhost:2181,localhost:2182
+
+# ... or chrooted to /zookeeper
+# rest.endpoint.1 = /;localhost:2181,localhost:2182/zookeeper
+
+# HTTP Basic authentication for this endpoint
+# rest.endpoint.1.http.auth = root:root1
+
+# create -e /a data digest:'demo:ojnHEyje6F33LLzGVzg+yatf4Fc=':cdrwa
+# any session on this endpoint will use authentication
+# rest.endpoint.1.zk.digest = demo:test
+
+# you can easily generate the ACL using Python:
+# import sha; sha.sha('demo:test').digest().encode('base64').strip()
+
+#
+# ... you can define as many endpoints as you wish
+#
+
+# rest.endpoint.2 = /restricted;localhost:2181
+# rest.endpoint.2.http.auth = admin:pass
+
+# rest.endpoint.3 = /cluster1;localhost:2181,localhost:2182
+# ** you should configure one end-point for each ZooKeeper cluster
+# etc.
+
+# Global HTTP Basic Authentication 
+# You should also enable HTTPS-only access
+# The authentication credentials are sent as plain text
+
+# rest.http.auth = guest:guest1
+
+# Uncomment the lines bellow to allow https-only access
+
+# rest.ssl = true
+# rest.ssl.jks = keys/rest.jks
+# rest.ssl.jks.pass = 123456
+ 

src/contrib/rest/rest.sh

@@ -0,0 +1,90 @@
+#!/bin/sh
+
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+#
+# If this scripted is run out of /usr/bin or some other system bin directory
+# it should be linked to and not copied. Things like java jar files are found
+# relative to the canonical path of this script.
+#
+
+# Only follow symlinks if readlink supports it
+if readlink -f "$0" > /dev/null 2>&1
+then
+  ZKREST=`readlink -f "$0"`
+else
+  ZKREST="$0"
+fi
+ZKREST_HOME=`dirname "$ZKREST"`
+
+if $cygwin
+then
+    # cygwin has a "kill" in the shell itself, gets confused
+    KILL=/bin/kill
+else
+    KILL=kill
+fi
+
+if [ -z $ZKREST_PIDFILE ]
+    then ZKREST_PIDFILE=$ZKREST_HOME/server.pid
+fi
+
+ZKREST_MAIN=org.apache.zookeeper.server.jersey.RestMain
+
+ZKREST_CONF=$ZKREST_HOME/conf
+ZKREST_LOG=$ZKREST_HOME/zkrest.log
+
+CLASSPATH="$ZKREST_CONF:$CLASSPATH"
+
+for i in "$ZKREST_HOME"/lib/*.jar
+do
+    CLASSPATH="$i:$CLASSPATH"
+done
+
+for i in "$ZKREST_HOME"/zookeeper-*.jar
+do
+    CLASSPATH="$i:$CLASSPATH"
+done
+
+case $1 in
+start)
+    echo  "Starting ZooKeeper REST Gateway ... "
+    java  -cp "$CLASSPATH" $JVMFLAGS $ZKREST_MAIN >$ZKREST_LOG 2>&1 &
+    /bin/echo -n $! > "$ZKREST_PIDFILE"
+    echo STARTED
+    ;;
+stop)
+    echo "Stopping ZooKeeper REST Gateway ... "
+    if [ ! -f "$ZKREST_PIDFILE" ]
+    then
+    echo "error: could not find file $ZKREST_PIDFILE"
+    exit 1
+    else
+    $KILL -9 $(cat "$ZKREST_PIDFILE")
+    rm "$ZKREST_PIDFILE"
+    echo STOPPED
+    fi
+    ;;
+restart)
+    shift
+    "$0" stop ${@}
+    sleep 3
+    "$0" start ${@}
+    ;;
+*)
+    echo "Usage: $0 {start|stop|restart}" >&2
+
+esac

src/contrib/rest/src/java/org/apache/zookeeper/server/jersey/RestMain.java

@@ -18,61 +18,133 @@
 
 package org.apache.zookeeper.server.jersey;
 
+import java.io.File;
 import java.io.IOException;
-import java.util.HashMap;
-import java.util.Map;
+import java.net.URISyntaxException;
+import java.net.URL;
 
-import com.sun.grizzly.http.SelectorThread;
-import com.sun.jersey.api.container.grizzly.GrizzlyWebContainerFactory;
+import org.apache.log4j.Logger;
+import org.apache.zookeeper.server.jersey.cfg.Credentials;
+import org.apache.zookeeper.server.jersey.cfg.Endpoint;
+import org.apache.zookeeper.server.jersey.cfg.RestCfg;
+import org.apache.zookeeper.server.jersey.filters.HTTPBasicAuth;
+
+import com.sun.grizzly.SSLConfig;
+import com.sun.grizzly.http.embed.GrizzlyWebServer;
+import com.sun.grizzly.http.servlet.ServletAdapter;
+import com.sun.jersey.spi.container.servlet.ServletContainer;
 
 /**
  * Demonstration of how to run the REST service using Grizzly
  */
 public class RestMain {
-    private String baseUri;
-
-    public RestMain(String baseUri) {
-        this.baseUri = baseUri;
-    }
-
-    public SelectorThread execute() throws IOException {
-        final Map<String, String> initParams = new HashMap<String, String>();
-
-        initParams.put("com.sun.jersey.config.property.packages",
-                       "org.apache.zookeeper.server.jersey.resources");
-
-        System.out.println("Starting grizzly...");
-        SelectorThread threadSelector =
-            GrizzlyWebContainerFactory.create(baseUri, initParams);
-
-        return threadSelector;
-    }
-
-    /**
-     * The entry point for starting the server
-     *
-     * @param args requires 2 arguments; the base uri of the service (e.g.
-     *        http://localhost:9998/) and the zookeeper host:port string
-     */
-    public static void main(String[] args) throws Exception {
-        final String baseUri = args[0];
-        final String zkHostPort = args[1];
-
-        ZooKeeperService.mapUriBase(baseUri, zkHostPort);
-
-        RestMain main = new RestMain(baseUri);
-        SelectorThread sel = main.execute();
-
-        System.out.println(String.format(
-                "Jersey app started with WADL available at %sapplication.wadl\n” +"
-                + "Try out %szookeeper\nHit enter to stop it...",
-                baseUri, baseUri));
-
-        System.in.read();
-        sel.stopEndpoint();
 
-        ZooKeeperService.close(baseUri);
-        System.exit(0);
-    }
+   private static Logger LOG = Logger.getLogger(RestMain.class);
+
+   private GrizzlyWebServer gws;
+   private RestCfg cfg;
+
+   public RestMain(RestCfg cfg) {
+       this.cfg = cfg;
+   }
+
+   public void start() throws IOException {
+       System.out.println("Starting grizzly ...");
+
+       boolean useSSL = cfg.useSSL();
+       gws = new GrizzlyWebServer(cfg.getPort(), "/tmp/23cxv45345/2131xc2/", useSSL);
+       // BUG: Grizzly needs a doc root if you are going to register multiple adapters
+
+       for (Endpoint e : cfg.getEndpoints()) {
+           ZooKeeperService.mapContext(e.getContext(), e);
+           gws.addGrizzlyAdapter(createJerseyAdapter(e), new String[] { e
+                   .getContext() });
+       }
+       
+       if (useSSL) {
+           System.out.println("Starting SSL ...");
+           String jks = cfg.getJKS("keys/rest.jks");
+           String jksPassword = cfg.getJKSPassword();
+
+           SSLConfig sslConfig = new SSLConfig();
+           URL resource = getClass().getClassLoader().getResource(jks);
+           if (resource == null) {
+               LOG.error("Unable to find the keystore file: " + jks);
+               System.exit(2);
+           }
+           try {
+               sslConfig.setKeyStoreFile(new File(resource.toURI())
+                       .getAbsolutePath());
+           } catch (URISyntaxException e1) {
+               LOG.error("Unable to load keystore: " + jks, e1);
+               System.exit(2);
+           }
+           sslConfig.setKeyStorePass(jksPassword);
+           gws.setSSLConfig(sslConfig);
+       }
+
+       gws.start();
+   }
+
+   public void stop() {
+       gws.stop();
+       ZooKeeperService.closeAll();
+   }
+
+   private ServletAdapter createJerseyAdapter(Endpoint e) {
+       ServletAdapter jersey = new ServletAdapter();
+
+       jersey.setServletInstance(new ServletContainer());
+       jersey.addInitParameter("com.sun.jersey.config.property.packages",
+               "org.apache.zookeeper.server.jersey.resources");
+       jersey.setContextPath(e.getContext());
+
+       Credentials c = Credentials.join(e.getCredentials(), cfg
+               .getCredentials());
+       if (!c.isEmpty()) {
+           jersey.addFilter(new HTTPBasicAuth(c), e.getContext()
+                   + "-basic-auth", null);
+       }
+
+       return jersey;
+   }
+
+   /**
+    * The entry point for starting the server
+    * 
+    */
+   public static void main(String[] args) throws Exception {
+       RestCfg cfg = new RestCfg("rest.properties");
+
+       final RestMain main = new RestMain(cfg);
+       main.start();
+
+       Runtime.getRuntime().addShutdownHook(new Thread() {
+           @Override
+           public void run() {
+               main.stop();
+               System.out.println("Got exit request. Bye.");
+           }
+       });
+
+       printEndpoints(cfg);
+       System.out.println("Server started.");
+   }
+
+   private static void printEndpoints(RestCfg cfg) {
+       int port = cfg.getPort();
+
+       for (Endpoint e : cfg.getEndpoints()) {
+
+           String context = e.getContext();
+           if (context.charAt(context.length() - 1) != '/') {
+               context += "/";
+           }
+
+           System.out.println(String.format(
+                   "Started %s - WADL: http://localhost:%d%sapplication.wadl",
+                   context, port, context));
+       }
+   }
 
 }

src/contrib/rest/src/java/org/apache/zookeeper/server/jersey/ZooKeeperService.java

@@ -20,91 +20,222 @@ package org.apache.zookeeper.server.jersey;
 
 import java.io.IOException;
 import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+import java.util.Timer;
+import java.util.TimerTask;
+import java.util.TreeSet;
 
+import org.apache.log4j.Logger;
 import org.apache.zookeeper.WatchedEvent;
 import org.apache.zookeeper.Watcher;
 import org.apache.zookeeper.ZooKeeper;
 import org.apache.zookeeper.Watcher.Event.KeeperState;
+import org.apache.zookeeper.server.jersey.cfg.Endpoint;
 
 /**
- * Singleton which provides JAX-RS resources access to the ZooKeeper
- * client. There's a single session for each base uri (so usually just one).
+ * Singleton which provides JAX-RS resources access to the ZooKeeper client.
+ * There's a single session for each base uri (so usually just one).
  */
 public class ZooKeeperService {
-    /** Map base uri to ZooKeeper host:port parameters */
-    private static HashMap<String, String> uriMap
-        = new HashMap<String, String>();
-
-    /** Map base uri to ZooKeeper session */
-    private static HashMap<String, ZooKeeper> zkMap =
-        new HashMap<String, ZooKeeper>();
-
-    /** Track the status of the ZooKeeper session */
-    private static class MyWatcher implements Watcher {
-        volatile boolean connected;
-        final String uriBase;
-
-        /** Separate watcher for each base uri */
-        public MyWatcher(String uriBase) {
-            this.uriBase = uriBase;
-        }
-
-        /** Track state - in particular watch for expiration. if
-         * it happens for re-creation of the ZK client session
-         */
-        synchronized public void process(WatchedEvent event) {
-            if (event.getState() == KeeperState.SyncConnected) {
-                connected = true;
-            } else if (event.getState() == KeeperState.Expired) {
-                connected = false;
-                close(uriBase);
-            } else {
-                connected = false;
-            }
-        }
-    }
-
-    /**
-     * Specify ZooKeeper host:port for a particular base uri. The host:port
-     * string is passed to the ZK client, so this can be formatted with
-     * more than a single host:port pair.
-     */
-    synchronized public static void mapUriBase(String uriBase, String hostport)
-    {
-        uriMap.put(uriBase, hostport);
-    }
-
-    /**
-     * Close the ZooKeeper session and remove it from the internal maps
-     */
-    synchronized public static void close(String uriBase) {
-        ZooKeeper zk = zkMap.remove(uriBase);
-        if (zk == null) {
-            return;
-        }
-        try {
-            zk.close();
-        } catch (InterruptedException e) {
-            // FIXME
-            e.printStackTrace();
-        }
-    }
-
-    /**
-     * Return a ZooKeeper client which may or may not be connected, but
-     * it will not be expired. This method can be called multiple times,
-     * the same object will be returned except in the case where the
-     * session expires (at which point a new session will be returned)
-     */
-    synchronized public static ZooKeeper getClient(String baseUri)
-        throws IOException
-    {
-        ZooKeeper zk = zkMap.get(baseUri);
-        if (zk == null) {
-            String hostPort = uriMap.get(baseUri);
-            zk = new ZooKeeper(hostPort, 30000, new MyWatcher(baseUri));
-            zkMap.put(baseUri, zk);
-        }
-        return zk;
-    }
+
+   private static Logger LOG = Logger.getLogger(ZooKeeperService.class);
+
+   /** Map base uri to ZooKeeper host:port parameters */
+   private static Map<String, Endpoint> contextMap = new HashMap<String, Endpoint>();
+
+   /** Map base uri to ZooKeeper session */
+   private static Map<String, ZooKeeper> zkMap = new HashMap<String, ZooKeeper>();
+
+   /** Session timers */
+   private static Map<String, SessionTimerTask> zkSessionTimers = new HashMap<String, SessionTimerTask>();
+   private static Timer timer = new Timer();
+
+   /** Track the status of the ZooKeeper session */
+   private static class MyWatcher implements Watcher {
+       final String contextPath;
+
+       /** Separate watcher for each base uri */
+       public MyWatcher(String contextPath) {
+           this.contextPath = contextPath;
+       }
+
+       /**
+        * Track state - in particular watch for expiration. if it happens for
+        * re-creation of the ZK client session
+        */
+       synchronized public void process(WatchedEvent event) {
+           if (event.getState() == KeeperState.Expired) {
+               close(contextPath);
+           }
+       }
+   }
+
+   /** ZooKeeper session timer */
+   private static class SessionTimerTask extends TimerTask {
+
+       private int delay;
+       private String contextPath, session;
+       private Timer timer;
+
+       public SessionTimerTask(int delayInSeconds, String session,
+               String contextPath, Timer timer) {
+           delay = delayInSeconds * 1000; // convert to milliseconds
+           this.contextPath = contextPath;
+           this.session = session;
+           this.timer = timer;
+           reset();
+       }
+
+       public SessionTimerTask(SessionTimerTask t) {
+           this(t.delay / 1000, t.session, t.contextPath, t.timer);
+       }
+
+       @Override
+       public void run() {
+           if (LOG.isInfoEnabled()) {
+               LOG.info(String.format("Session '%s' expired after "
+                       + "'%d' milliseconds.", session, delay));
+           }
+           ZooKeeperService.close(contextPath, session);
+       }
+
+       public void reset() {
+           timer.schedule(this, delay);
+       }
+
+   }
+
+   /**
+    * Specify ZooKeeper host:port for a particular context path. The host:port
+    * string is passed to the ZK client, so this can be formatted with more
+    * than a single host:port pair.
+    */
+   synchronized public static void mapContext(String contextPath, Endpoint e) {
+       contextMap.put(contextPath, e);
+   }
+
+   /**
+    * Reset timer for a session
+    */
+   synchronized public static void resetTimer(String contextPath,
+           String session) {
+       if (session != null) {
+           String uri = concat(contextPath, session);
+
+           SessionTimerTask t = zkSessionTimers.remove(uri);
+           t.cancel();
+
+           zkSessionTimers.put(uri, new SessionTimerTask(t));
+       }
+   }
+
+   /**
+    * Close the ZooKeeper session and remove it from the internal maps
+    */
+   public static void close(String contextPath) {
+       close(contextPath, null);
+   }
+
+   /**
+    * Close the ZooKeeper session and remove it
+    */
+   synchronized public static void close(String contextPath, String session) {
+       String uri = concat(contextPath, session);
+
+       TimerTask t = zkSessionTimers.remove(uri);
+       if (t != null) {
+           t.cancel();
+       }
+
+       ZooKeeper zk = zkMap.remove(uri);
+       if (zk == null) {
+           return;
+       }
+       try {
+           zk.close();
+       } catch (InterruptedException e) {
+           LOG.error("Interrupted while closing ZooKeeper connection.", e);
+       }
+   }
+
+   /**
+    * Close all the ZooKeeper sessions and remove them from the internal maps
+    */
+   synchronized public static void closeAll() {
+       Set<String> sessions = new TreeSet<String>(zkMap.keySet());
+       for (String key : sessions) {
+           close(key);
+       }
+   }
+
+   /**
+    * Is there an active connection for this session?
+    */
+   synchronized public static boolean isConnected(String contextPath,
+           String session) {
+       return zkMap.containsKey(concat(contextPath, session));
+   }
+
+   /**
+    * Return a ZooKeeper client not tied to a specific session.
+    */
+   public static ZooKeeper getClient(String contextPath) throws IOException {
+       return getClient(contextPath, null);
+   }
+
+   /**
+    * Return a ZooKeeper client for a session with a default expire time
+    * 
+    * @throws IOException
+    */
+   public static ZooKeeper getClient(String contextPath, String session)
+           throws IOException {
+       return getClient(contextPath, session, 5);
+   }
+
+   /**
+    * Return a ZooKeeper client which may or may not be connected, but it will
+    * not be expired. This method can be called multiple times, the same object
+    * will be returned except in the case where the session expires (at which
+    * point a new session will be returned)
+    */
+   synchronized public static ZooKeeper getClient(String contextPath,
+           String session, int expireTime) throws IOException {
+       final String connectionId = concat(contextPath, session);
+
+       ZooKeeper zk = zkMap.get(connectionId);
+       if (zk == null) {
+
+           if (LOG.isInfoEnabled()) {
+               LOG.info(String.format("creating new "
+                       + "connection for : '%s'", connectionId));
+           }
+           Endpoint e = contextMap.get(contextPath);
+           zk = new ZooKeeper(e.getHostPort(), 30000, new MyWatcher(
+                   connectionId));
+           
+           for (Map.Entry<String, String> p : e.getZooKeeperAuthInfo().entrySet()) {
+               zk.addAuthInfo("digest", String.format("%s:%s", p.getKey(),
+                       p.getValue()).getBytes());
+           }
+           
+           zkMap.put(connectionId, zk);
+
+           // a session should automatically expire after an amount of time
+           if (session != null) {
+               zkSessionTimers.put(connectionId, new SessionTimerTask(
+                       expireTime, session, contextPath, timer));
+           }
+       }
+       return zk;
+   }
+
+   private static String concat(String contextPath, String session) {
+       if (session != null) {
+           return String.format("%s@%s", contextPath, session);
+       }
+       return contextPath;
+   }
+
 }

src/contrib/rest/src/java/org/apache/zookeeper/server/jersey/cfg/Credentials.java

@@ -0,0 +1,47 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.zookeeper.server.jersey.cfg;
+
+import java.util.HashMap;
+
+public class Credentials extends HashMap<String, String> {
+
+   public static Credentials join(Credentials a, Credentials b) {
+       Credentials result = new Credentials();
+       result.putAll(a);
+       result.putAll(b);
+       return result;
+   }
+   
+   public Credentials() {
+       super();
+   }
+   
+   public Credentials(String credentials) {
+       super();
+       
+       if (!credentials.trim().equals("")) {
+           String[] parts = credentials.split(",");
+           for(String p : parts) {
+               String[] userPass = p.split(":");
+               put(userPass[0], userPass[1]);
+           }
+       }
+   }
+}

src/contrib/rest/src/java/org/apache/zookeeper/server/jersey/cfg/Endpoint.java

@@ -0,0 +1,72 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.zookeeper.server.jersey.cfg;
+
+public class Endpoint {
+
+   private String context;
+   private HostPortSet hostPort;
+   private Credentials credentials;
+   private Credentials zookeeperAuth;
+
+   public Endpoint(String context, String hostPortList) {
+       this.context = context;
+       this.hostPort = new HostPortSet(hostPortList);
+   }
+
+   public String getContext() {
+       return context;
+   }
+
+   public String getHostPort() {
+       return hostPort.toString();
+   }
+
+   public Credentials getCredentials() {
+       return credentials;
+   }
+   
+   public void setCredentials(String c) {
+       this.credentials = new Credentials(c);
+   }
+   
+   public void setZooKeeperAuthInfo(String digest) {
+       zookeeperAuth = new Credentials(digest);
+   }
+   
+   public final Credentials getZooKeeperAuthInfo() {
+       return zookeeperAuth;
+   }
+
+   @Override
+   public boolean equals(Object o) {
+       Endpoint e = (Endpoint) o;
+       return context.equals(e.context);
+   }
+
+   @Override
+   public int hashCode() {
+       return context.hashCode();
+   }
+
+   @Override
+   public String toString() {
+       return String.format("<Endpoint %s %s>", context, hostPort.toString());
+   }
+}

src/contrib/rest/src/java/org/apache/zookeeper/server/jersey/cfg/HostPort.java

@@ -0,0 +1,51 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.zookeeper.server.jersey.cfg;
+
+public class HostPort {
+
+   private String host;
+   private int port;
+   
+   public HostPort(String hostPort) {
+       String[] parts = hostPort.split(":");
+       host = parts[0];
+       port = Integer.parseInt(parts[1]);
+   }
+
+   public String getHost() {
+       return host;
+   }
+
+   public int getPort() {
+       return port;
+   }
+
+   @Override
+   public boolean equals(Object o) {
+       HostPort p = (HostPort) o;
+       return host.equals(p.host) && port == p.port;
+   }
+   
+   @Override
+   public int hashCode() {
+       return String.format("%s:%d", host, port).hashCode();
+   }
+   
+}

src/contrib/rest/src/java/org/apache/zookeeper/server/jersey/cfg/HostPortSet.java

@@ -0,0 +1,51 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.zookeeper.server.jersey.cfg;
+
+import java.util.HashSet;
+import java.util.Set;
+
+public class HostPortSet {
+
+   private Set<HostPort> hostPortSet = new HashSet<HostPort>();
+   private String original;
+   
+   public HostPortSet(String hostPortList) {
+       original = hostPortList;
+       
+       int chrootStart = hostPortList.indexOf('/');
+       String hostPortPairs;
+       if (chrootStart != -1) {
+           hostPortPairs = hostPortList.substring(0, chrootStart);
+       } else {
+           hostPortPairs = hostPortList;
+       }
+       
+       String[] parts = hostPortPairs.split(",");
+       for(String p : parts) {
+           hostPortSet.add(new HostPort(p));
+       }
+   }
+   
+   @Override
+   public String toString() {
+       return original;
+   }
+   
+}

src/contrib/rest/src/java/org/apache/zookeeper/server/jersey/cfg/RestCfg.java

@@ -0,0 +1,106 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.zookeeper.server.jersey.cfg;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.HashSet;
+import java.util.Properties;
+import java.util.Set;
+
+public class RestCfg {
+
+   private Properties cfg = new Properties();
+
+   private Set<Endpoint> endpoints = new HashSet<Endpoint>();
+   private Credentials credentials = new Credentials();
+
+   public RestCfg(String resource) throws IOException {
+       this(RestCfg.class.getClassLoader().getResourceAsStream(resource));
+   }
+
+   public RestCfg(InputStream io) throws IOException {
+       cfg.load(io);
+       extractEndpoints();
+       extractCredentials();
+   }
+
+   private void extractCredentials() {
+       if (cfg.containsKey("rest.http.auth")) {
+           credentials = new Credentials(cfg.getProperty("rest.http.auth", ""));
+       }
+   }
+
+   private void extractEndpoints() {
+       int count = 1;
+       while (true) {
+           String e = cfg.getProperty(
+                   String.format("rest.endpoint.%d", count), null);
+           if (e == null) {
+               break;
+           }
+
+           String[] parts = e.split(";");
+           if (parts.length != 2) {
+               count++;
+               continue;
+           }
+           Endpoint point = new Endpoint(parts[0], parts[1]);
+           
+           String c = cfg.getProperty(String.format(
+                   "rest.endpoint.%d.http.auth", count), "");
+           point.setCredentials(c);
+           
+           String digest = cfg.getProperty(String.format(
+                   "rest.endpoint.%d.zk.digest", count), "");
+           point.setZooKeeperAuthInfo(digest);
+
+           endpoints.add(point);
+           count++;
+       }
+   }
+
+   public int getPort() {
+       return Integer.parseInt(cfg.getProperty("rest.port", "9998"));
+   }
+
+   public boolean useSSL() {
+       return Boolean.valueOf(cfg.getProperty("rest.ssl", "false"));
+   }
+
+   public final Set<Endpoint> getEndpoints() {
+       return endpoints;
+   }
+
+   public final Credentials getCredentials() {
+       return credentials;
+   }
+
+   public String getJKS() {
+       return cfg.getProperty("rest.ssl.jks");
+   }
+
+   public String getJKS(String def) {
+       return cfg.getProperty("rest.ssl.jks", def);
+   }
+
+   public String getJKSPassword() {
+       return cfg.getProperty("rest.ssl.jks.pass");
+   }
+}

src/contrib/rest/src/java/org/apache/zookeeper/server/jersey/filters/HTTPBasicAuth.java

@@ -0,0 +1,87 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.zookeeper.server.jersey.filters;
+
+import java.io.IOException;
+
+import javax.servlet.Filter;
+import javax.servlet.FilterChain;
+import javax.servlet.FilterConfig;
+import javax.servlet.ServletException;
+import javax.servlet.ServletRequest;
+import javax.servlet.ServletResponse;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+
+import org.apache.zookeeper.server.jersey.cfg.Credentials;
+
+import com.sun.jersey.core.util.Base64;
+
+public class HTTPBasicAuth implements Filter {
+
+    private Credentials credentials;
+
+    public HTTPBasicAuth(Credentials c) {
+       credentials = c;
+    }
+
+    @Override
+    public void doFilter(ServletRequest req0, ServletResponse resp0,
+            FilterChain chain) throws IOException, ServletException {
+
+        HttpServletRequest request = (HttpServletRequest) req0;
+        HttpServletResponse response = (HttpServletResponse) resp0;
+
+        String authorization = request.getHeader("Authorization");
+        if (authorization != null) {
+            String c[] = parseAuthorization(authorization);
+            if (c != null && credentials.containsKey(c[0])
+                    && credentials.get(c[0]).equals(c[1])) {
+                chain.doFilter(request, response);
+                return;
+            }
+        }
+
+        response.setHeader("WWW-Authenticate", "Basic realm=\"Restricted\"");
+        response.sendError(401);
+    }
+
+    private String[] parseAuthorization(String authorization) {
+        String parts[] = authorization.split(" ");
+        if (parts.length == 2 && parts[0].equalsIgnoreCase("Basic")) {
+            String userPass = Base64.base64Decode(parts[1]);
+
+            int p = userPass.indexOf(":");
+            if (p != -1) {
+                return new String[] { userPass.substring(0, p),
+                        userPass.substring(p + 1) };
+            }
+        }
+        return null;
+    }
+
+    @Override
+    public void init(FilterConfig arg0) throws ServletException {
+    }
+
+    @Override
+    public void destroy() {
+    }
+
+}

src/contrib/rest/src/java/org/apache/zookeeper/server/jersey/jaxb/ZSession.java

@@ -0,0 +1,55 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.zookeeper.server.jersey.jaxb;
+
+import javax.xml.bind.annotation.XmlRootElement;
+
+@XmlRootElement(name="session")
+public class ZSession {
+    public String id;
+    public String uri;
+    
+    public ZSession() {
+        // needed by jersey
+    }
+    
+    public ZSession(String id, String uri) {
+        this.id = id;
+        this.uri = uri;
+    }
+    
+    @Override
+    public int hashCode() {
+        return id.hashCode();
+    }
+    
+    @Override
+    public boolean equals(Object obj) {
+        if(!(obj instanceof ZSession)) {
+            return false;
+        }
+        ZSession s = (ZSession) obj;
+        return id.equals(s.id);
+    }
+    
+    @Override
+    public String toString() {
+        return "ZSession(" + id +")";   
+    }
+}

src/contrib/rest/src/java/org/apache/zookeeper/server/jersey/resources/SessionsResource.java

@@ -0,0 +1,134 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.zookeeper.server.jersey.resources;
+
+import java.io.IOException;
+import java.net.URI;
+import java.util.UUID;
+
+import javax.servlet.http.HttpServletRequest;
+import javax.ws.rs.Consumes;
+import javax.ws.rs.DELETE;
+import javax.ws.rs.DefaultValue;
+import javax.ws.rs.POST;
+import javax.ws.rs.PUT;
+import javax.ws.rs.Path;
+import javax.ws.rs.PathParam;
+import javax.ws.rs.Produces;
+import javax.ws.rs.QueryParam;
+import javax.ws.rs.WebApplicationException;
+import javax.ws.rs.core.Context;
+import javax.ws.rs.core.MediaType;
+import javax.ws.rs.core.Response;
+import javax.ws.rs.core.UriInfo;
+
+import org.apache.log4j.Logger;
+import org.apache.zookeeper.server.jersey.ZooKeeperService;
+import org.apache.zookeeper.server.jersey.jaxb.ZError;
+import org.apache.zookeeper.server.jersey.jaxb.ZSession;
+
+import com.sun.jersey.api.json.JSONWithPadding;
+
+@Path("sessions/v1/{session: .*}")
+public class SessionsResource {
+
+    private static Logger LOG = Logger.getLogger(SessionsResource.class);
+
+    private String contextPath;
+
+    public SessionsResource(@Context HttpServletRequest request) {
+        contextPath = request.getContextPath();
+        if (contextPath.equals("")) {
+            contextPath = "/";
+        }
+    }
+
+    @PUT
+    @Produces( { MediaType.APPLICATION_JSON, "application/javascript",
+            MediaType.APPLICATION_XML })
+    @Consumes(MediaType.APPLICATION_OCTET_STREAM)
+    public Response keepAliveSession(@PathParam("session") String session,
+            @Context UriInfo ui, byte[] data) {
+
+        if (!ZooKeeperService.isConnected(contextPath, session)) {
+            throwNotFound(session, ui);
+        }
+
+        ZooKeeperService.resetTimer(contextPath, session);
+        return Response.status(Response.Status.OK).build();
+    }
+
+    @POST
+    @Produces( { MediaType.APPLICATION_JSON, "application/javascript",
+            MediaType.APPLICATION_XML })
+    public Response createSession(@QueryParam("op") String op,
+            @DefaultValue("5") @QueryParam("expire") String expire,
+            @Context UriInfo ui) {
+        if (!op.equals("create")) {
+            throw new WebApplicationException(Response.status(
+                    Response.Status.BAD_REQUEST).entity(
+                    new ZError(ui.getRequestUri().toString(), "")).build());
+        }
+
+        int expireInSeconds;
+        try {
+            expireInSeconds = Integer.parseInt(expire);
+        } catch (NumberFormatException e) {
+            throw new WebApplicationException(Response.status(
+                    Response.Status.BAD_REQUEST).build());
+        }
+
+        String uuid = UUID.randomUUID().toString();
+        while (ZooKeeperService.isConnected(contextPath, uuid)) {
+            uuid = UUID.randomUUID().toString();
+        }
+
+        // establish the connection to the ZooKeeper cluster
+        try {
+            ZooKeeperService.getClient(contextPath, uuid, expireInSeconds);
+        } catch (IOException e) {
+            LOG.error("Failed while trying to create a new session", e);
+
+            throw new WebApplicationException(Response.status(
+                    Response.Status.INTERNAL_SERVER_ERROR).build());
+        }
+
+        URI uri = ui.getAbsolutePathBuilder().path(uuid).build();
+        return Response.created(uri).entity(
+                new JSONWithPadding(new ZSession(uuid, uri.toString())))
+                .build();
+    }
+
+    @DELETE
+    @Produces( { MediaType.APPLICATION_JSON, "application/javascript",
+            MediaType.APPLICATION_XML, MediaType.APPLICATION_OCTET_STREAM })
+    public void deleteSession(@PathParam("session") String session,
+            @Context UriInfo ui) {
+        ZooKeeperService.close(contextPath, session);
+    }
+
+    private static void throwNotFound(String session, UriInfo ui)
+            throws WebApplicationException {
+        throw new WebApplicationException(Response.status(
+                Response.Status.NOT_FOUND).entity(
+                new ZError(ui.getRequestUri().toString(), session
+                        + " not found")).build());
+    }
+
+}

src/contrib/rest/src/java/org/apache/zookeeper/server/jersey/resources/ZNodeResource.java

@@ -23,6 +23,7 @@ import java.net.URI;
 import java.util.ArrayList;
 import java.util.List;
 
+import javax.servlet.http.HttpServletRequest;
 import javax.ws.rs.Consumes;
 import javax.ws.rs.DELETE;
 import javax.ws.rs.DefaultValue;
@@ -62,8 +63,23 @@ import com.sun.jersey.api.json.JSONWithPadding;
 public class ZNodeResource {
     private final ZooKeeper zk;
 
-    public ZNodeResource(@Context UriInfo ui) throws IOException {
-        zk = ZooKeeperService.getClient(ui.getBaseUri().toString());
+    public ZNodeResource(@DefaultValue("") @QueryParam("session") String session,
+            @Context UriInfo ui,
+            @Context HttpServletRequest request
+            )
+            throws IOException {
+
+        String contextPath = request.getContextPath();
+        if (contextPath.equals("")) {
+            contextPath = "/";
+        }
+        if (session.equals("")) {
+            session = null;
+        } else if (!ZooKeeperService.isConnected(contextPath, session)) {
+            throw new WebApplicationException(Response.status(
+                    Response.Status.UNAUTHORIZED).build());
+        }
+        zk = ZooKeeperService.getClient(contextPath, session);
     }
 
     private void ensurePathNotNull(String path) {
@@ -73,12 +89,10 @@ public class ZNodeResource {
     }
 
     @HEAD
-    @Produces({MediaType.APPLICATION_JSON, "application/javascript",
-        MediaType.APPLICATION_XML})
+    @Produces( { MediaType.APPLICATION_JSON, "application/javascript",
+            MediaType.APPLICATION_XML })
     public Response existsZNode(@PathParam("path") String path,
-            @Context UriInfo ui)
-        throws InterruptedException, KeeperException
-    {
+            @Context UriInfo ui) throws InterruptedException, KeeperException {
         Stat stat = zk.exists(path, false);
         if (stat == null) {
             throwNotFound(path, ui);
@@ -87,11 +101,9 @@ public class ZNodeResource {
     }
 
     @HEAD
-    @Produces({MediaType.APPLICATION_OCTET_STREAM})
+    @Produces( { MediaType.APPLICATION_OCTET_STREAM })
     public Response existsZNodeAsOctet(@PathParam("path") String path,
-            @Context UriInfo ui)
-        throws InterruptedException, KeeperException
-    {
+            @Context UriInfo ui) throws InterruptedException, KeeperException {
         Stat stat = zk.exists(path, false);
         if (stat == null) {
             throwNotFound(path, ui);
@@ -101,40 +113,37 @@ public class ZNodeResource {
 
     /*
      * getZNodeList and getZNodeListJSON are bogus - but necessary.
-     * Unfortunately Jersey 1.0.3 is unable to render both xml and json
-     * properly in the case where a object contains a list/array. It's
-     * impossible to get it to render properly for both. As a result we
-     * need to split into two jaxb classes.
+     * Unfortunately Jersey 1.0.3 is unable to render both xml and json properly
+     * in the case where a object contains a list/array. It's impossible to get
+     * it to render properly for both. As a result we need to split into two
+     * jaxb classes.
      */
 
     @GET
-    @Produces({MediaType.APPLICATION_JSON, "application/javascript"})
-    public Response getZNodeListJSON(@PathParam("path") String path,
+    @Produces( { MediaType.APPLICATION_JSON, "application/javascript" })
+    public Response getZNodeListJSON(
+            @PathParam("path") String path,
             @QueryParam("callback") String callback,
             @DefaultValue("data") @QueryParam("view") String view,
             @DefaultValue("base64") @QueryParam("dataformat") String dataformat,
-            @Context UriInfo ui)
-        throws InterruptedException, KeeperException
-    {
-            return getZNodeList(true, path, callback, view, dataformat, ui);
+            @Context UriInfo ui) throws InterruptedException, KeeperException {
+        return getZNodeList(true, path, callback, view, dataformat, ui);
     }
 
     @GET
     @Produces(MediaType.APPLICATION_XML)
-    public Response getZNodeList(@PathParam("path") String path,
+    public Response getZNodeList(
+            @PathParam("path") String path,
             @QueryParam("callback") String callback,
             @DefaultValue("data") @QueryParam("view") String view,
             @DefaultValue("base64") @QueryParam("dataformat") String dataformat,
-            @Context UriInfo ui)
-        throws InterruptedException, KeeperException
-    {
+            @Context UriInfo ui) throws InterruptedException, KeeperException {
         return getZNodeList(false, path, callback, view, dataformat, ui);
     }
 
     private Response getZNodeList(boolean json, String path, String callback,
             String view, String dataformat, UriInfo ui)
-        throws InterruptedException, KeeperException
-    {
+            throws InterruptedException, KeeperException {
         ensurePathNotNull(path);
 
         if (view.equals("children")) {
@@ -150,8 +159,9 @@ public class ZNodeResource {
             }
             childTemplate += "{child}";
             if (json) {
-                child = new ZChildrenJSON(path, ui.getAbsolutePath().toString(),
-                        childTemplate, children);
+                child = new ZChildrenJSON(path,
+                        ui.getAbsolutePath().toString(), childTemplate,
+                        children);
             } else {
                 child = new ZChildren(path, ui.getAbsolutePath().toString(),
                         childTemplate, children);
@@ -167,7 +177,7 @@ public class ZNodeResource {
             if (data == null) {
                 data64 = null;
                 dataUtf8 = null;
-            } else if (!dataformat.equals("utf8")){
+            } else if (!dataformat.equals("utf8")) {
                 data64 = data;
                 dataUtf8 = null;
             } else {
@@ -175,12 +185,11 @@ public class ZNodeResource {
                 dataUtf8 = new String(data);
             }
             ZStat zstat = new ZStat(path, ui.getAbsolutePath().toString(),
-                    data64, dataUtf8, stat.getCzxid(),
-                    stat.getMzxid(), stat.getCtime(), stat.getMtime(),
-                    stat.getVersion(), stat.getCversion(),
-                    stat.getAversion(), stat.getEphemeralOwner(),
-                    stat.getDataLength(), stat.getNumChildren(),
-                    stat.getPzxid());
+                    data64, dataUtf8, stat.getCzxid(), stat.getMzxid(), stat
+                            .getCtime(), stat.getMtime(), stat.getVersion(),
+                    stat.getCversion(), stat.getAversion(), stat
+                            .getEphemeralOwner(), stat.getDataLength(), stat
+                            .getNumChildren(), stat.getPzxid());
 
             return Response.status(Response.Status.OK).entity(
                     new JSONWithPadding(zstat, callback)).build();
@@ -190,8 +199,7 @@ public class ZNodeResource {
     @GET
     @Produces(MediaType.APPLICATION_OCTET_STREAM)
     public Response getZNodeListAsOctet(@PathParam("path") String path)
-        throws InterruptedException, KeeperException
-    {
+            throws InterruptedException, KeeperException {
         ensurePathNotNull(path);
 
         Stat stat = new Stat();
@@ -205,18 +213,17 @@ public class ZNodeResource {
     }
 
     @PUT
-    @Produces({MediaType.APPLICATION_JSON, "application/javascript",
-        MediaType.APPLICATION_XML})
+    @Produces( { MediaType.APPLICATION_JSON, "application/javascript",
+            MediaType.APPLICATION_XML })
     @Consumes(MediaType.APPLICATION_OCTET_STREAM)
-    public Response setZNode(@PathParam("path") String path,
+    public Response setZNode(
+            @PathParam("path") String path,
             @QueryParam("callback") String callback,
             @DefaultValue("-1") @QueryParam("version") String versionParam,
             @DefaultValue("base64") @QueryParam("dataformat") String dataformat,
             @DefaultValue("false") @QueryParam("null") String setNull,
-            @Context UriInfo ui,
-            byte[] data)
-        throws InterruptedException, KeeperException
-    {
+            @Context UriInfo ui, byte[] data) throws InterruptedException,
+            KeeperException {
         ensurePathNotNull(path);
 
         int version;
@@ -225,8 +232,8 @@ public class ZNodeResource {
         } catch (NumberFormatException e) {
             throw new WebApplicationException(Response.status(
                     Response.Status.BAD_REQUEST).entity(
-                    new ZError(ui.getRequestUri().toString(),
-                            path + " bad version " + versionParam)).build());
+                    new ZError(ui.getRequestUri().toString(), path
+                            + " bad version " + versionParam)).build());
         }
 
         if (setNull.equals("true")) {
@@ -235,13 +242,12 @@ public class ZNodeResource {
 
         Stat stat = zk.setData(path, data, version);
 
-        ZStat zstat = new ZStat(path, ui.getAbsolutePath().toString(),
-                null, null, stat.getCzxid(),
-                stat.getMzxid(), stat.getCtime(), stat.getMtime(),
-                stat.getVersion(), stat.getCversion(),
-                stat.getAversion(), stat.getEphemeralOwner(),
-                stat.getDataLength(), stat.getNumChildren(),
-                stat.getPzxid());
+        ZStat zstat = new ZStat(path, ui.getAbsolutePath().toString(), null,
+                null, stat.getCzxid(), stat.getMzxid(), stat.getCtime(), stat
+                        .getMtime(), stat.getVersion(), stat.getCversion(),
+                stat.getAversion(), stat.getEphemeralOwner(), stat
+                        .getDataLength(), stat.getNumChildren(), stat
+                        .getPzxid());
 
         return Response.status(Response.Status.OK).entity(
                 new JSONWithPadding(zstat, callback)).build();
@@ -253,10 +259,8 @@ public class ZNodeResource {
     public void setZNodeAsOctet(@PathParam("path") String path,
             @DefaultValue("-1") @QueryParam("version") String versionParam,
             @DefaultValue("false") @QueryParam("null") String setNull,
-            @Context UriInfo ui,
-            byte[] data)
-        throws InterruptedException, KeeperException
-    {
+            @Context UriInfo ui, byte[] data) throws InterruptedException,
+            KeeperException {
         ensurePathNotNull(path);
 
         int version;
@@ -265,8 +269,8 @@ public class ZNodeResource {
         } catch (NumberFormatException e) {
             throw new WebApplicationException(Response.status(
                     Response.Status.BAD_REQUEST).entity(
-                    new ZError(ui.getRequestUri().toString(),
-                            path + " bad version " + versionParam)).build());
+                    new ZError(ui.getRequestUri().toString(), path
+                            + " bad version " + versionParam)).build());
         }
 
         if (setNull.equals("true")) {
@@ -277,20 +281,20 @@ public class ZNodeResource {
     }
 
     @POST
-    @Produces({MediaType.APPLICATION_JSON, "application/javascript",
-        MediaType.APPLICATION_XML})
+    @Produces( { MediaType.APPLICATION_JSON, "application/javascript",
+            MediaType.APPLICATION_XML })
     @Consumes(MediaType.APPLICATION_OCTET_STREAM)
-    public Response createZNode(@PathParam("path") String path,
+    public Response createZNode(
+            @PathParam("path") String path,
             @QueryParam("callback") String callback,
             @DefaultValue("create") @QueryParam("op") String op,
             @QueryParam("name") String name,
             @DefaultValue("base64") @QueryParam("dataformat") String dataformat,
             @DefaultValue("false") @QueryParam("null") String setNull,
             @DefaultValue("false") @QueryParam("sequence") String sequence,
-            @Context UriInfo ui,
-            byte[] data)
-        throws InterruptedException, KeeperException
-    {
+            @DefaultValue("false") @QueryParam("ephemeral") String ephemeral,
+            @Context UriInfo ui, byte[] data) throws InterruptedException,
+            KeeperException {
         ensurePathNotNull(path);
 
         if (path.equals("/")) {
@@ -302,8 +306,8 @@ public class ZNodeResource {
         if (!op.equals("create")) {
             throw new WebApplicationException(Response.status(
                     Response.Status.BAD_REQUEST).entity(
-                    new ZError(ui.getRequestUri().toString(),
-                            path + " bad operaton " + op)).build());
+                    new ZError(ui.getRequestUri().toString(), path
+                            + " bad operaton " + op)).build());
         }
 
         if (setNull.equals("true")) {
@@ -312,19 +316,24 @@ public class ZNodeResource {
 
         CreateMode createMode;
         if (sequence.equals("true")) {
-            createMode = CreateMode.PERSISTENT_SEQUENTIAL;
-        } else {
+            if (ephemeral.equals("false")) {
+                createMode = CreateMode.PERSISTENT_SEQUENTIAL;
+            } else {
+                createMode = CreateMode.EPHEMERAL_SEQUENTIAL;
+            }
+        } else if (ephemeral.equals("false")) {
             createMode = CreateMode.PERSISTENT;
+        } else {
+            createMode = CreateMode.EPHEMERAL;
         }
 
-        String newPath = zk.create(path, data, Ids.OPEN_ACL_UNSAFE,
-                createMode);
+        String newPath = zk.create(path, data, Ids.OPEN_ACL_UNSAFE, createMode);
 
         URI uri = ui.getAbsolutePathBuilder().path(newPath).build();
 
         return Response.created(uri).entity(
-                new JSONWithPadding(new ZPath(newPath,
-                        ui.getAbsolutePath().toString()))).build();
+                new JSONWithPadding(new ZPath(newPath, ui.getAbsolutePath()
+                        .toString()))).build();
     }
 
     @POST
@@ -335,10 +344,8 @@ public class ZNodeResource {
             @QueryParam("name") String name,
             @DefaultValue("false") @QueryParam("null") String setNull,
             @DefaultValue("false") @QueryParam("sequence") String sequence,
-            @Context UriInfo ui,
-            byte[] data)
-        throws InterruptedException, KeeperException
-    {
+            @Context UriInfo ui, byte[] data) throws InterruptedException,
+            KeeperException {
         ensurePathNotNull(path);
 
         if (path.equals("/")) {
@@ -350,8 +357,8 @@ public class ZNodeResource {
         if (!op.equals("create")) {
             throw new WebApplicationException(Response.status(
                     Response.Status.BAD_REQUEST).entity(
-                    new ZError(ui.getRequestUri().toString(),
-                            path + " bad operaton " + op)).build());
+                    new ZError(ui.getRequestUri().toString(), path
+                            + " bad operaton " + op)).build());
         }
 
         if (setNull.equals("true")) {
@@ -365,23 +372,20 @@ public class ZNodeResource {
             createMode = CreateMode.PERSISTENT;
         }
 
-        String newPath = zk.create(path, data, Ids.OPEN_ACL_UNSAFE,
-                createMode);
+        String newPath = zk.create(path, data, Ids.OPEN_ACL_UNSAFE, createMode);
 
         URI uri = ui.getAbsolutePathBuilder().path(newPath).build();
 
-        return Response.created(uri).entity(new ZPath(newPath,
-                ui.getAbsolutePath().toString())).build();
+        return Response.created(uri).entity(
+                new ZPath(newPath, ui.getAbsolutePath().toString())).build();
     }
 
     @DELETE
-    @Produces({MediaType.APPLICATION_JSON, "application/javascript",
-        MediaType.APPLICATION_XML, MediaType.APPLICATION_OCTET_STREAM})
+    @Produces( { MediaType.APPLICATION_JSON, "application/javascript",
+            MediaType.APPLICATION_XML, MediaType.APPLICATION_OCTET_STREAM })
     public void deleteZNode(@PathParam("path") String path,
             @DefaultValue("-1") @QueryParam("version") String versionParam,
-            @Context UriInfo ui)
-        throws InterruptedException, KeeperException
-    {
+            @Context UriInfo ui) throws InterruptedException, KeeperException {
         ensurePathNotNull(path);
 
         int version;
@@ -390,20 +394,19 @@ public class ZNodeResource {
         } catch (NumberFormatException e) {
             throw new WebApplicationException(Response.status(
                     Response.Status.BAD_REQUEST).entity(
-                    new ZError(ui.getRequestUri().toString(),
-                            path + " bad version " + versionParam)).build());
+                    new ZError(ui.getRequestUri().toString(), path
+                            + " bad version " + versionParam)).build());
         }
 
         zk.delete(path, version);
     }
 
     private static void throwNotFound(String path, UriInfo ui)
-        throws WebApplicationException
-    {
+            throws WebApplicationException {
         throw new WebApplicationException(Response.status(
                 Response.Status.NOT_FOUND).entity(
-                new ZError(ui.getRequestUri().toString(),
-                        path + " not found")).build());
+                new ZError(ui.getRequestUri().toString(), path + " not found"))
+                .build());
     }
 
 }

src/contrib/rest/src/python/README.txt

@@ -1,5 +1,8 @@
 Some basic python scripts which use the REST interface:
 
+zkrest.py -- basic REST ZooKeeper client
+demo_master_election.py -- shows how to implement master election
+demo_queue.py -- basic queue
 zk_dump_tree.py -- dumps the nodes & data of a znode hierarchy
 
 Generally these scripts require:

src/contrib/rest/src/python/demo_master_election.py

@@ -0,0 +1,90 @@
+#! /usr/bin/env python
+
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import sys
+import threading
+import time
+
+from zkrest import ZooKeeper
+
+class Agent(threading.Thread):
+    """ A basic agent that wants to become a master and exit """
+
+    root = '/election'
+
+    def __init__(self, id):
+        super(Agent, self).__init__()
+        self.zk = ZooKeeper()
+        self.id = id
+
+    def run(self):
+        print 'Starting #%s' % self.id
+        with self.zk.session(expire=5):
+
+            # signal agent presence
+            r = self.zk.create("%s/agent-" % self.root, 
+                sequence=True, ephemeral=True)
+            self.me = r['path']
+
+            while True:
+                children = sorted([el['path'] \
+                    for el in self.zk.get_children(self.root)])
+                master, previous = children[0], None
+                try:
+                    index = children.index(self.me)
+                    if index != 0:
+                        previous = children[index-1]
+                except ValueError:
+                    break
+
+                if previous is None:
+                    self.do_master_work()
+                    # and don't forget to send heartbeat messages
+                    break
+                else:
+                    # do slave work in another thread
+                    pass
+               
+                # wait for the previous agent or current master to exit / finish
+                while self.zk.exists(previous) or self.zk.exists(master):
+                    time.sleep(0.5)
+                    self.zk.heartbeat()
+
+                # TODO signal the slave thread to exit and wait for it
+                # and rerun the election loop
+
+    def do_master_work(self):
+        print "#%s: I'm the master: %s" % (self.id, self.me) 
+            
+def main():
+    zk = ZooKeeper()
+
+    # create the root node used for master election
+    if not zk.exists('/election'):
+        zk.create('/election')
+
+    print 'Starting 10 agents ...'
+    agents = [Agent(id) for id in range(0,15)]
+
+    map(Agent.start, agents)
+    map(Agent.join, agents)
+
+    zk.delete('/election')    
+
+if __name__ == '__main__':
+    sys.exit(main())

src/contrib/rest/src/python/demo_queue.py

@@ -0,0 +1,99 @@
+#! /usr/bin/env python
+
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+
+# This is a simple message queue built on top of ZooKeeper. In order
+# to be used in production it needs better error handling but it's 
+# still useful as a proof-of-concept. 
+
+# Why use ZooKeeper as a queue? Highly available by design and has
+# great performance.
+
+import sys
+import threading
+import time
+
+from zkrest import ZooKeeper
+
+class Queue(object):
+    def __init__(self, root, zk):
+        self.root = root
+
+        self.zk = zk
+
+    def put(self, data):
+        self.zk.create("%s/el-" % self.root, str(data), sequence=True, ephemeral=True)
+
+        # creating ephemeral nodes for easy cleanup
+        # in a real world scenario you should create
+        # normal sequential znodes
+
+    def fetch(self):
+        """ Pull an element from the queue
+
+        This function is not blocking if the queue is empty, it will
+        just return None.
+        """
+        children = sorted(self.zk.get_children(self.root), \
+            lambda a, b: cmp(a['path'], b['path']))
+
+        if not children:
+            return None
+
+        try:
+            first = children[0]
+            self.zk.delete(first['path'], version=first['version'])
+            if 'data64' not in first:
+                return ''
+            else:
+                return first['data64'].decode('base64')
+
+        except (ZooKeeper.WrongVersion, ZooKeeper.NotFound):
+            # someone changed the znode between the get and delete
+            # this should not happen
+            # in practice you should retry the fetch
+            raise
+        
+
+def main():
+    zk = ZooKeeper()
+    zk.start_session(expire=60)
+
+    if not zk.exists('/queue'):
+        zk.create('/queue')
+    q = Queue('/queue', zk)
+
+    print 'Pushing to queue 1 ... 5'
+    map(q.put, [1,2,3,4,5])
+
+    print 'Extracting ...'
+    while True:
+        el = q.fetch()
+        if el is None:
+            break
+        print el    
+
+    zk.close_session()
+    zk.delete('/queue')
+
+    print 'Done.'
+   
+
+if __name__ == '__main__':
+    sys.exit(main())
+

src/contrib/rest/src/python/test.py

@@ -0,0 +1,163 @@
+#! /usr/bin/env python
+
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import time
+import unittest
+
+from zkrest import ZooKeeper
+
+class ZooKeeperREST_TestCase(unittest.TestCase):
+    
+    BASE_URI = 'http://localhost:9998'
+
+    def setUp(self):
+        self.zk = ZooKeeper(self.BASE_URI)
+
+    def tearDown(self):
+        try:
+            self.zk.delete('/test')
+        except ZooKeeper.NotFound:
+            pass
+
+    def test_get_root_node(self):
+        assert self.zk.get('/') is not None
+
+    def test_get_node_not_found(self):
+        self.assertRaises(ZooKeeper.NotFound, \
+            self.zk.get, '/dummy-node')
+
+    def test_exists_node(self):
+        assert self.zk.exists('/zookeeper') is True
+
+    def test_get_children(self):
+        assert any([child['path'] == '/zookeeper/quota' \
+            for child in self.zk.get_children('/zookeeper')])
+            
+    def test_create_znode(self):
+        try:
+            self.zk.create('/test')
+        except ZooKeeper.ZNodeExists:
+            pass # it's ok if already exists
+        assert self.zk.exists('/test') is True
+
+    def test_create_hierarchy(self):
+        try:
+            self.zk.delete(['/a/b', '/a'])
+        except ZooKeeper.NotFound:
+            pass
+
+        self.zk.create('/a')
+        self.zk.create('/a/b')
+
+        self.zk.delete(['/a/b', '/a'])
+
+    def test_create_with_data(self):
+        self.zk.create('/test', 'some-data')
+
+        zn = self.zk.get('/test')
+        self.assertEqual(zn.get('data64', None), \
+            'some-data'.encode('base64').strip())
+
+    def test_delete_znode(self):
+        self.zk.create('/test')
+
+        self.zk.delete('/test')
+        assert not self.zk.exists('/test')
+
+    def test_delete_older_version(self):
+        self.zk.create('/test')
+
+        zn = self.zk.get('/test')
+        # do one more modification in order to increase the version number
+        self.zk.set('/test', 'dummy-data')
+
+        self.assertRaises(ZooKeeper.WrongVersion, \
+            self.zk.delete, '/test', version=zn['version'])
+
+    def test_delete_raise_not_found(self):
+        self.zk.create('/test')
+
+        zn = self.zk.get('/test')
+        self.zk.delete('/test')
+ 
+        self.assertRaises(ZooKeeper.NotFound, \
+            self.zk.delete, '/test', version=zn['version'])
+
+    def test_set(self):
+        self.zk.create('/test')
+
+        self.zk.set('/test', 'dummy')
+
+        self.assertEqual(self.zk.get('/test')['data64'], \
+            'dummy'.encode('base64').strip())
+
+    def test_set_with_older_version(self):
+        if not self.zk.exists('/test'):
+            self.zk.create('/test', 'random-data')
+
+        zn = self.zk.get('/test')
+        self.zk.set('/test', 'new-data')
+        self.assertRaises(ZooKeeper.WrongVersion, self.zk.set, \
+            '/test', 'older-version', version=zn['version'])
+
+    def test_set_null(self):
+        if not self.zk.exists('/test'):
+            self.zk.create('/test', 'random-data')
+        self.zk.set('/test', 'data')
+        assert 'data64' in self.zk.get('/test')
+
+        self.zk.set('/test', null=True)
+        assert 'data64' not in self.zk.get('/test')
+
+    def test_create_ephemeral_node(self):
+        with self.zk.session():
+            if self.zk.exists('/ephemeral-test'):
+                self.zk.delete('/ephemeral-test')
+
+            self.zk.create('/ephemeral-test', ephemeral=True)
+            zn = self.zk.get('/ephemeral-test')
+
+            assert zn['ephemeralOwner'] != 0
+
+    def test_create_session(self):
+        with self.zk.session() as sid:
+            self.assertEqual(len(sid), 36) # UUID
+
+    def test_session_invalidation(self):
+        self.zk.start_session(expire=1)
+        self.zk.create('/ephemeral-test', ephemeral=True)
+
+        # keep the session alive by sending heartbeat requests
+        for _ in range(1,2):
+            self.zk.heartbeat()
+            time.sleep(0.9)
+
+        time.sleep(2) # wait for the session to expire
+        self.assertRaises(ZooKeeper.InvalidSession, \
+            self.zk.create, '/ephemeral-test', ephemeral=True)
+
+    def test_presence_signaling(self):
+        with self.zk.session(expire=1):
+            self.zk.create('/i-am-online', ephemeral=True)
+            assert self.zk.exists('/i-am-online')
+        assert not self.zk.exists('/i-am-online')
+
+
+if __name__ == '__main__':
+    unittest.main()
+

src/contrib/rest/src/python/zkrest.py

@@ -0,0 +1,218 @@
+
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import urllib2
+import urllib
+import simplejson
+
+from contextlib import contextmanager
+
+class RequestWithMethod(urllib2.Request):
+    """ Request class that know how to set the method name """
+    def __init__(self, *args, **kwargs):
+        urllib2.Request.__init__(self, *args, **kwargs)
+        self._method = None
+
+    def get_method(self):
+        return self._method or \
+            urllib2.Request.get_method(self)
+
+    def set_method(self, method):
+        self._method = method
+
+class ZooKeeper(object):
+
+    class Error(Exception): pass
+
+    class NotFound(Error): pass
+
+    class ZNodeExists(Error): pass
+
+    class InvalidSession(Error): pass
+
+    class WrongVersion(Error): pass
+
+    def __init__(self, uri = 'http://localhost:9998'):
+        self._base = uri
+        self._session = None
+
+    def start_session(self, expire=5, id=None):
+        """ Create a session and return the ID """
+        if id is None:
+            url = "%s/sessions/v1/?op=create&expire=%d" % (self._base, expire)
+            self._session = self._do_post(url)['id']
+        else:
+            self._session = id
+        return self._session
+
+    def close_session(self):
+        """ Close the session on the server """
+        if self._session is not None:
+            url = '%s/sessions/v1/%s' % (self._base, self._session)
+            self._do_delete(url)
+            self._session = None
+
+    def heartbeat(self):
+        """ Send a heartbeat request. This is needed in order to keep a session alive """
+        if self._session is not None:
+            url = '%s/sessions/v1/%s' % (self._base, self._session)
+            self._do_put(url, '')
+
+    @contextmanager
+    def session(self, *args, **kwargs):
+        """ Session handling using a context manager """
+        yield self.start_session(*args, **kwargs)
+        self.close_session()
+
+    def get(self, path):
+        """ Get a node """
+        url = "%s/znodes/v1%s" % (self._base, path)
+        return self._do_get(url)
+
+    def get_children(self, path):
+        """ Get all the children for a given path. This function creates a generator """
+        url = "%s/znodes/v1%s?view=children" % (self._base, path)
+        resp = self._do_get(url)
+        for child in resp.get('children', []):
+            try:
+                yield self._do_get(resp['child_uri_template']\
+                    .replace('{child}', urllib2.quote(child)))
+            except ZooKeeper.NotFound:
+                continue
+
+    def create(self, path, data=None, sequence=False, ephemeral=False):
+        """ Create a new node. By default this call creates a persistent znode.
+
+        You can also create an ephemeral or a sequential znode.
+        """
+        ri = path.rindex('/')
+        head, name = path[:ri+1], path[ri+1:]
+        if head != '/': head = head[:-1]
+
+        flags = {
+            'null': 'true' if data is None else 'false',
+            'ephemeral': 'true' if ephemeral else 'false',
+            'sequence': 'true' if sequence else 'false'
+        }
+        if ephemeral:
+            if self._session:
+                flags['session'] = self._session
+            else:
+                raise ZooKeeper.Error, 'You need a session '\
+                    'to create an ephemeral node'
+        flags = urllib.urlencode(flags)
+
+        url = "%s/znodes/v1%s?op=create&name=%s&%s" % \
+            (self._base, head, name, flags)
+
+        return self._do_post(url, data)
+
+    def set(self, path, data=None, version=-1, null=False):
+        """ Set the value of node """
+        url = "%s/znodes/v1%s?%s" % (self._base, path, \
+            urllib.urlencode({
+                'version': version,
+                'null': 'true' if null else 'false'
+        }))
+        return self._do_put(url, data)
+
+    def delete(self, path, version=-1):
+        """ Delete a znode """
+        if type(path) is list:
+            map(lambda el: self.delete(el, version), path)
+            return
+
+        url = '%s/znodes/v1%s?%s' % (self._base, path, \
+            urllib.urlencode({
+                'version':version
+        }))
+        try:
+            return self._do_delete(url)
+        except urllib2.HTTPError, e:
+            if e.code == 412:
+                raise ZooKeeper.WrongVersion(path)
+            elif e.code == 404:
+                raise ZooKeeper.NotFound(path)
+            raise
+
+    def exists(self, path):
+        """ Do a znode exists """
+        try:
+            self.get(path)
+            return True
+        except ZooKeeper.NotFound:
+            return False
+
+    def _do_get(self, uri):
+        """ Send a GET request and convert errors to exceptions """
+        try:
+            req = urllib2.urlopen(uri)
+            resp = simplejson.load(req)
+
+            if 'Error' in resp:
+               raise ZooKeeper.Error(resp['Error'])
+
+            return resp
+        except urllib2.HTTPError, e:
+            if e.code == 404:
+                raise ZooKeeper.NotFound(uri)
+            raise
+
+    def _do_post(self, uri, data=None):
+        """ Send a POST request and convert errors to exceptions """
+        try:
+            req = urllib2.Request(uri, {})
+            req.add_header('Content-Type', 'application/octet-stream')
+            if data is not None:
+                req.add_data(data)
+
+            resp = simplejson.load(urllib2.urlopen(req))
+            if 'Error' in resp:
+                raise ZooKeeper.Error(resp['Error'])
+            return resp
+
+        except urllib2.HTTPError, e:
+            if e.code == 201:
+                return True
+            elif e.code == 409:
+                raise ZooKeeper.ZNodeExists(uri)
+            elif e.code == 401:
+                raise ZooKeeper.InvalidSession(uri)
+            raise
+
+    def _do_delete(self, uri):
+        """ Send a DELETE request """
+        req = RequestWithMethod(uri)
+        req.set_method('DELETE')
+        req.add_header('Content-Type', 'application/octet-stream')
+        return urllib2.urlopen(req).read()
+
+    def _do_put(self, uri, data):
+        """ Send a PUT request """
+        try:
+            req = RequestWithMethod(uri)
+            req.set_method('PUT')
+            req.add_header('Content-Type', 'application/octet-stream')
+            if data is not None:
+                req.add_data(data)
+
+            return urllib2.urlopen(req).read()
+        except urllib2.HTTPError, e:
+            if e.code == 412: # precondition failed
+                raise ZooKeeper.WrongVersion(uri)
+            raise
+

src/contrib/rest/src/test/org/apache/zookeeper/server/jersey/Base.java

@@ -18,6 +18,8 @@
 
 package org.apache.zookeeper.server.jersey;
 
+import java.io.ByteArrayInputStream;
+
 import junit.framework.TestCase;
 
 import org.apache.log4j.Logger;
@@ -25,65 +27,67 @@ import org.apache.zookeeper.CreateMode;
 import org.apache.zookeeper.ZooKeeper;
 import org.apache.zookeeper.ZooDefs.Ids;
 import org.apache.zookeeper.server.jersey.SetTest.MyWatcher;
+import org.apache.zookeeper.server.jersey.cfg.RestCfg;
 import org.junit.After;
 import org.junit.Before;
 
-import com.sun.grizzly.http.SelectorThread;
 import com.sun.jersey.api.client.Client;
 import com.sun.jersey.api.client.WebResource;
 
-
 /**
  * Test stand-alone server.
- *
+ * 
  */
 public class Base extends TestCase {
-    protected static final Logger LOG = Logger.getLogger(Base.class);
-
-    protected static final String BASEURI = "http://localhost:10104/";
-    protected static final String ZKHOSTPORT = "localhost:22182";
-    protected Client c;
-    protected WebResource r;
-
-    protected ZooKeeper zk;
+   protected static final Logger LOG = Logger.getLogger(Base.class);
 
-    private SelectorThread threadSelector;
+   protected static final String CONTEXT_PATH = "/zk";
+   protected static final int GRIZZLY_PORT = 10104;
+   protected static final String BASEURI = String.format(
+           "http://localhost:%d%s", GRIZZLY_PORT, CONTEXT_PATH);
+   protected static final String ZKHOSTPORT = "localhost:22182";
+   protected Client client;
+   protected WebResource znodesr, sessionsr;
 
-    @Before
-    public void setUp() throws Exception {
-        super.setUp();
+   protected ZooKeeper zk;
 
-        ZooKeeperService.mapUriBase(BASEURI, ZKHOSTPORT);
+   private RestMain rest;
 
-        RestMain main = new RestMain(BASEURI);
-        threadSelector = main.execute();
+   @Before
+   public void setUp() throws Exception {
+       super.setUp();
 
-        zk = new ZooKeeper(ZKHOSTPORT, 30000, new MyWatcher());
+       RestCfg cfg = new RestCfg(new ByteArrayInputStream(String.format(
+               "rest.port=%s\n" + 
+               "rest.endpoint.1=%s;%s\n",
+               GRIZZLY_PORT, CONTEXT_PATH, ZKHOSTPORT).getBytes()));
 
-        c = Client.create();
-        r = c.resource(BASEURI);
-        r = r.path("znodes/v1");
-    }
+       rest = new RestMain(cfg);
+       rest.start();
 
-    @After
-    public void tearDown() throws Exception {
-        super.tearDown();
+       zk = new ZooKeeper(ZKHOSTPORT, 30000, new MyWatcher());
 
-        c.destroy();
+       client = Client.create();
+       znodesr = client.resource(BASEURI).path("znodes/v1");
+       sessionsr = client.resource(BASEURI).path("sessions/v1/");
+   }
 
-        zk.close();
-        ZooKeeperService.close(BASEURI);
+   @After
+   public void tearDown() throws Exception {
+       super.tearDown();
 
-        threadSelector.stopEndpoint();
-    }
+       client.destroy();
+       zk.close();
+       rest.stop();
+   }
 
-    protected static String createBaseZNode() throws Exception {
-        ZooKeeper zk = new ZooKeeper(ZKHOSTPORT, 30000, new MyWatcher());
+   protected static String createBaseZNode() throws Exception {
+       ZooKeeper zk = new ZooKeeper(ZKHOSTPORT, 30000, new MyWatcher());
 
-        String baseZnode = zk.create("/test-", null, Ids.OPEN_ACL_UNSAFE,
-                CreateMode.PERSISTENT_SEQUENTIAL);
-        zk.close();
+       String baseZnode = zk.create("/test-", null, Ids.OPEN_ACL_UNSAFE,
+               CreateMode.PERSISTENT_SEQUENTIAL);
+       zk.close();
 
-        return baseZnode;
-    }
+       return baseZnode;
+   }
 }

src/contrib/rest/src/test/org/apache/zookeeper/server/jersey/CreateTest.java

@@ -116,7 +116,7 @@ public class CreateTest extends Base {
     public void testCreate() throws Exception {
         LOG.info("STARTING " + getName());
 
-        WebResource wr = r.path(path).queryParam("dataformat", encoding)
+        WebResource wr = znodesr.path(path).queryParam("dataformat", encoding)
             .queryParam("name", name);
         if (data == null) {
             wr = wr.queryParam("null", "true");
@@ -142,10 +142,10 @@ public class CreateTest extends Base {
         ZPath zpath = cr.getEntity(ZPath.class);
         if (sequence) {
             assertTrue(zpath.path.startsWith(expectedPath.path));
-            assertTrue(zpath.uri.startsWith(r.path(path).toString()));
+            assertTrue(zpath.uri.startsWith(znodesr.path(path).toString()));
         } else {
             assertEquals(expectedPath, zpath);
-            assertEquals(r.path(path).toString(), zpath.uri);
+            assertEquals(znodesr.path(path).toString(), zpath.uri);
         }
 
         // use out-of-band method to verify

src/contrib/rest/src/test/org/apache/zookeeper/server/jersey/DeleteTest.java

@@ -75,7 +75,7 @@ public class DeleteTest extends Base {
                     CreateMode.PERSISTENT_SEQUENTIAL);
         }
 
-        ClientResponse cr = r.path(zpath).accept(type).type(type)
+        ClientResponse cr = znodesr.path(zpath).accept(type).type(type)
             .delete(ClientResponse.class);
         assertEquals(expectedStatus, cr.getClientResponseStatus());
 

src/contrib/rest/src/test/org/apache/zookeeper/server/jersey/ExistsTest.java

@@ -59,7 +59,7 @@ public class ExistsTest extends Base {
     }
 
     private void verify(String type) {
-        ClientResponse cr = r.path(path).accept(type).type(type).head();
+        ClientResponse cr = znodesr.path(path).accept(type).type(type).head();
         if (type.equals(MediaType.APPLICATION_OCTET_STREAM)
                 && expectedStatus == ClientResponse.Status.OK) {
             assertEquals(ClientResponse.Status.NO_CONTENT,

src/contrib/rest/src/test/org/apache/zookeeper/server/jersey/GetChildrenTest.java

@@ -107,7 +107,7 @@ public class GetChildrenTest extends Base {
             }
         }
 
-        ClientResponse cr = r.path(path).queryParam("view", "children")
+        ClientResponse cr = znodesr.path(path).queryParam("view", "children")
             .accept(accept).get(ClientResponse.class);
         assertEquals(expectedStatus, cr.getClientResponseStatus());
 
@@ -120,16 +120,16 @@ public class GetChildrenTest extends Base {
             Collections.sort(expectedChildren);
             Collections.sort(zchildren.children);
             assertEquals(expectedChildren, zchildren.children);
-            assertEquals(r.path(path).toString(), zchildren.uri);
-            assertEquals(r.path(path).toString() + "/{child}",
+            assertEquals(znodesr.path(path).toString(), zchildren.uri);
+            assertEquals(znodesr.path(path).toString() + "/{child}",
                     zchildren.child_uri_template);
         } else if (accept.equals(MediaType.APPLICATION_XML)) {
             ZChildren zchildren = cr.getEntity(ZChildren.class);
             Collections.sort(expectedChildren);
             Collections.sort(zchildren.children);
             assertEquals(expectedChildren, zchildren.children);
-            assertEquals(r.path(path).toString(), zchildren.uri);
-            assertEquals(r.path(path).toString() + "/{child}",
+            assertEquals(znodesr.path(path).toString(), zchildren.uri);
+            assertEquals(znodesr.path(path).toString() + "/{child}",
                     zchildren.child_uri_template);
         } else {
             fail("unknown accept type");

src/contrib/rest/src/test/org/apache/zookeeper/server/jersey/GetTest.java

@@ -107,7 +107,7 @@ public class GetTest extends Base {
             }
         }
 
-        ClientResponse cr = r.path(path).queryParam("dataformat", encoding)
+        ClientResponse cr = znodesr.path(path).queryParam("dataformat", encoding)
             .accept(accept).get(ClientResponse.class);
         assertEquals(expectedStatus, cr.getClientResponseStatus());
 
@@ -117,6 +117,6 @@ public class GetTest extends Base {
 
         ZStat zstat = cr.getEntity(ZStat.class);
         assertEquals(expectedStat, zstat);
-        assertEquals(r.path(path).toString(), zstat.uri);
+        assertEquals(znodesr.path(path).toString(), zstat.uri);
     }
 }

src/contrib/rest/src/test/org/apache/zookeeper/server/jersey/RootTest.java

@@ -19,19 +19,13 @@
 package org.apache.zookeeper.server.jersey;
 
 import java.util.Arrays;
-import java.util.Collection;
 
 import javax.ws.rs.core.MediaType;
 
 import org.apache.log4j.Logger;
-import org.apache.zookeeper.WatchedEvent;
-import org.apache.zookeeper.Watcher;
 import org.apache.zookeeper.data.Stat;
 import org.apache.zookeeper.server.jersey.jaxb.ZPath;
 import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.junit.runners.Parameterized;
-import org.junit.runners.Parameterized.Parameters;
 
 import com.sun.jersey.api.client.ClientResponse;
 import com.sun.jersey.api.client.WebResource;
@@ -53,7 +47,7 @@ public class RootTest extends Base {
         String name = "roottest-create";
         byte[] data = "foo".getBytes();
 
-        WebResource wr = r.path(path).queryParam("dataformat", "utf8")
+        WebResource wr = znodesr.path(path).queryParam("dataformat", "utf8")
             .queryParam("name", name);
         Builder builder = wr.accept(MediaType.APPLICATION_JSON);
 
@@ -63,7 +57,7 @@ public class RootTest extends Base {
 
         ZPath zpath = cr.getEntity(ZPath.class);
         assertEquals(new ZPath(path + name), zpath);
-        assertEquals(r.path(path).toString(), zpath.uri);
+        assertEquals(znodesr.path(path).toString(), zpath.uri);
 
         // use out-of-band method to verify
         byte[] rdata = zk.getData(zpath.path, false, new Stat());

src/contrib/rest/src/test/org/apache/zookeeper/server/jersey/SessionTest.java

@@ -0,0 +1,133 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.zookeeper.server.jersey;
+
+import java.io.IOException;
+
+import javax.ws.rs.core.MediaType;
+
+import org.apache.log4j.Logger;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.data.Stat;
+import org.apache.zookeeper.server.jersey.jaxb.ZSession;
+import org.codehaus.jettison.json.JSONException;
+import org.junit.Test;
+
+import com.sun.jersey.api.client.Client;
+import com.sun.jersey.api.client.ClientResponse;
+import com.sun.jersey.api.client.WebResource;
+import com.sun.jersey.api.client.WebResource.Builder;
+
+public class SessionTest extends Base {
+    protected static final Logger LOG = Logger.getLogger(SessionTest.class);
+
+    private ZSession createSession() {
+        return createSession("30");
+    }
+
+    private ZSession createSession(String expire) {
+        WebResource wr = sessionsr.queryParam("op", "create")
+            .queryParam("expire", expire);
+        Builder b = wr.accept(MediaType.APPLICATION_JSON);
+
+        ClientResponse cr = b.post(ClientResponse.class, null);
+        assertEquals(ClientResponse.Status.CREATED, cr
+                .getClientResponseStatus());
+
+        return cr.getEntity(ZSession.class);
+    }
+
+    @Test
+    public void testCreateNewSession() throws JSONException {
+        ZSession session = createSession();
+        assertEquals(session.id.length(), 36);
+
+        // use out-of-band method to verify
+        assertTrue(ZooKeeperService.isConnected(CONTEXT_PATH, session.id));
+    }
+
+    @Test
+    public void testSessionExpires() throws InterruptedException {
+        ZSession session = createSession("1");
+
+        // use out-of-band method to verify
+        assertTrue(ZooKeeperService.isConnected(CONTEXT_PATH, session.id));
+
+        // wait for the session to be closed
+        Thread.sleep(1500);
+        assertFalse(ZooKeeperService.isConnected(CONTEXT_PATH, session.id));
+    }
+
+    @Test
+    public void testDeleteSession() {
+        ZSession session = createSession("30");
+
+        WebResource wr = sessionsr.path(session.id);
+        Builder b = wr.accept(MediaType.APPLICATION_JSON);
+
+        assertTrue(ZooKeeperService.isConnected(CONTEXT_PATH, session.id));
+        ClientResponse cr = b.delete(ClientResponse.class, null);
+        assertEquals(ClientResponse.Status.NO_CONTENT, 
+                cr.getClientResponseStatus());
+
+        assertFalse(ZooKeeperService.isConnected(CONTEXT_PATH, session.id));
+    }
+    
+    @Test
+    public void testSendHeartbeat() throws InterruptedException {
+        ZSession session = createSession("2");
+        
+        Thread.sleep(1000);
+        WebResource wr = sessionsr.path(session.id);
+        Builder b = wr.accept(MediaType.APPLICATION_JSON);
+        
+        ClientResponse cr = b.put(ClientResponse.class, null);
+        assertEquals(ClientResponse.Status.OK, cr.getClientResponseStatus());
+        
+        Thread.sleep(1500);
+        assertTrue(ZooKeeperService.isConnected(CONTEXT_PATH, session.id));
+        
+        Thread.sleep(1000);
+        assertFalse(ZooKeeperService.isConnected(CONTEXT_PATH, session.id));
+    }
+    
+    @Test
+    public void testCreateEphemeralZNode() 
+    throws KeeperException, InterruptedException, IOException {
+        ZSession session = createSession("30");
+        
+        WebResource wr = znodesr.path("/")
+            .queryParam("op", "create")
+            .queryParam("name", "ephemeral-test")
+            .queryParam("ephemeral", "true")
+            .queryParam("session", session.id)
+            .queryParam("null", "true");
+        
+        Builder b = wr.accept(MediaType.APPLICATION_JSON);
+        ClientResponse cr = b.post(ClientResponse.class);
+        assertEquals(ClientResponse.Status.CREATED, cr.getClientResponseStatus());
+        
+        Stat stat = new Stat();
+        zk.getData("/ephemeral-test", false, stat);
+        
+        ZooKeeper sessionZK = ZooKeeperService.getClient(CONTEXT_PATH, session.id);
+        assertEquals(stat.getEphemeralOwner(), sessionZK.getSessionId());
+    }
+}

src/contrib/rest/src/test/org/apache/zookeeper/server/jersey/SetTest.java

@@ -112,7 +112,7 @@ public class SetTest extends Base {
                     CreateMode.PERSISTENT);
         }
 
-        WebResource wr = r.path(path).queryParam("dataformat", encoding);
+        WebResource wr = znodesr.path(path).queryParam("dataformat", encoding);
         if (data == null) {
             wr = wr.queryParam("null", "true");
         }

src/contrib/rest/src/test/org/apache/zookeeper/server/jersey/WadlTest.java

@@ -34,7 +34,7 @@ public class WadlTest extends Base {
 
     @Test
     public void testApplicationWadl() {
-        WebResource r = c.resource(BASEURI);
+        WebResource r = client.resource(BASEURI);
         String serviceWadl = r.path("application.wadl").
                 accept(MediaTypes.WADL).get(String.class);
         assertTrue("Something wrong. Returned wadl length not > 0.",