Merge lp:~hazmat/txzookeeper/session-event-handling into lp:txzookeeper

[Kapil Thangavelu]

Session event and connection loss handling for txzookeeper.

  - Session tests using a ZK cluster as a test layer/test resource, cluster state
    is reset between tests.
  - Zookeeper Client session tests for server rotation across multi-node cluster.
  - Client server rotation and session/watch migration tests.
  - Session event handling, sent to user defined callback, else ignored by default.
  - Connection loss handling, sent to user defined callback, else raised at the
    API call point. The callback result if any is returned as the error to be
    returned to the API.

[Kapil Thangavelu (hazmat)]

Please note that this branch has a pre-requisite branch also in review.

[Gustavo Niemeyer (niemeyer)]

This looks very good. A few comments, and +1 with them considered.

[1]

+ zookeeper.NOTWATCHING_EVENT: "notwatching",
+ zookeeper.SESSION_EVENT: "session",
+ zookeeper.CREATED_EVENT: 'created',

Quotes are inconsistent here.

Also, "not-watching" would be a bit more friendly to the eye.

[2]

+def debug_symbols(sequence):

Left over?

[3]

error = error_class(error_msg)
+
+ if ConnectionException.is_connection_exception(error):

This idiom feels a little weird. A Foo is necessarily a Foo,
so we never have Foo.is_foo. Unless you have some background
for the choice that I miss, I suggest just moving that to a
"is_connection_exception()" top-level function.

[4]

+ # The result of the connection error handler is returned
+ # to the api.

That's a nice design, despite our conversations around the problem of
connection level erroring.

I'm a bit concerned about the session events, though. A session-termination
event may be the only event the deferred ever sees, so if we divert it to
a separate callback, the logic pending on the deferred will be dead forever.

I'm not sure about what we should do here. Have you thought about this?

If the answer is to just be so for now, we should at least add a note to
the proper location in the code mentioning that these deferreds are dead.

[5]

+ d = defer.maybeDeferred(
+ self._connection_error_callback,
+ self, error)

For consistency, either both callbacks should take "self", or neither of them
should.

[6]

+ # XXX/Debug
+ #print
+ #print "Session/Conn Event", ClientEvent(type, state, path)
+

Some left overs.

[7]

+ self._check_result(result, d)
+ if not d.called:
+ d.callback(True)

This is checking if the deferred is being called synchronously, which seems
dangerous as a pattern. Theorethically, the deferred may not have fired by
the time the function returns.

I suggest using the interface defined (considering the suggestions from
the pre-requisite branch):

  if not self._failed(result, d):
      d.callback(True)
  return d

review: Approve

[Jim Baker (jimbaker)]

+1, we certainly need this branch, and it looks quite solid.

In addition to Gustavo's comments:

[1]

Some minor grammar suggestions:

+ specified in comma separated fashion, if they are,

specified in comma separated fashion. If they are,

+ # If its a session event pass it to the session callback, else

If it's a session event, ...

+ nodes and watches)/ The connection's client id, is also useful
+ when introspecting the server logs for specific client
+ activity.

nodes and watches. The connection's client id is also...

+ Its used for all connection level events and session events.

It's used...

+ # Send session events to the callback, this in addition to any
+ # duplicate session events that will be sent for extant watches.

callback, in addition

+ """Set a callback to recieve session events.

receive

+ twisted integration using deferreds is not capable of recieving

receiving

+ to be invoked with them instead. The callback recieves a single

receives

+ call is made, by setting a connection level error handler,
+ applications can centralize their handling of connection loss,

Something like

By setting...

+ This is a global setting, across connections.

setting across all connections

+ A client session can be reattached to in a separate connection,
+ if the a session is expired, using the zookeeper connection will
+ raise a SessionExpiredException.

+ # On occassion we get a session expired event while connecting.

occasion

+ # Closing one connection, will close the session

Closing one connection will close the session.

+ A client connected to multiple servers, will transparently
+ migrate amongst them, as individual servers can no longer be
+ reached. A client's session will be maintined.

servers will transparently...

...will be maintained.

+ # get a zookeeper connectionloss exception on occassion.

occasion

+ We can specify a callback for connection errors, that
+ can perform recovery for a disconnected client, restablishing

This seems incomplete; also there should not be a comma before "that"

+ """On connection failure, An application can choose to use a
+ new connection with to reconnect to a different member of the
+ zookeeper cluster, reacquiring the extant session.

an application... with which to reconnect...

Maybe this concludes with "This enables reacquiring the existing
session." (Extant doesn't seem like the right word choice here.)

+ A large obvious caveat to using a new client instance rather
+ than reconnecting the existing client, is that even though the

instance, rather than...

+ # Ephemeral is destroyed when the session closed.

the session is closed.

[2]

+ http://bit.ly/mQrOMY
+ http://bit.ly/irKpfn

Why not just expand these links out? Alternatively use a readable URL
(the underlying nabble URL is modestly readable, I don't know how
permalink either one is, however).

[3]

+ assert callable(callback)

Shouldn't this be throwing a TypeError instead? Also, there is no
correponding assertion (or verification) on
set_connection_error_callback. Lastly, there is no test on this path.

[4]

+ def xtest_session_expired_event(self):

Do you want to include this test?

review: Approve

[Kapil Thangavelu (hazmat)]

Excerpts from Gustavo Niemeyer's message of Mon Jun 20 18:28:28 UTC 2011:
> Review: Approve
> This looks very good. A few comments, and +1 with them considered.
>

1-3 addressed.

>
> [4]
>
> + # The result of the connection error handler is returned
> + # to the api.
>
> That's a nice design, despite our conversations around the problem of
> connection level erroring.
>
> I'm a bit concerned about the session events, though. A session-termination
> event may be the only event the deferred ever sees, so if we divert it to
> a separate callback, the logic pending on the deferred will be dead forever.
>
> I'm not sure about what we should do here. Have you thought about this?
>
> If the answer is to just be so for now, we should at least add a note to
> the proper location in the code mentioning that these deferreds are dead.
>

Any deferreds attached to an unconnected client are dead. A session 'expired' event
is effectively a connection loss, the extant deferreds are dead. The expired
event is the only fatal session event. There is some testing around this (extant
watches on dead clients).

Alternatively, we could route session exceptions to the extant watcher, but then
we're back to the app code having to add guards to each watch for connection loss,
or lots of unhandled exceptions in deferred stacks. The connection exception
watcher will still fire as its always registered on the connection level watcher,
ie. the one that monitors the initial connection setup.

From a cleanup perspective this might be preferrable, but the numbers of extant
watches are small enough its not clear its really win, and most of that cleanup
will just result in ignorable log errors without connection loss handling on all
watches. The error handling itself and the error handling for a watchis basically
just ignore or propgate the exception to fully clear out the deferred chain with
the connection level error callback handling the actual reconnect scenario.

[5-6] Addressed

> [7]
>
> + self._check_result(result, d)
> + if not d.called:
> + d.callback(True)
>
> This is checking if the deferred is being called synchronously, which seems
> dangerous as a pattern. Theorethically, the deferred may not have fired by
> the time the function returns.
>
>
> I suggest using the interface defined (considering the suggestions from
> the pre-requisite branch):
>
> if not self._failed(result, d):
> d.callback(True)
> return d
>

In this case the txzk api is synchronous (close) and we're checking just checking
if its failed. I've updated the construct to not check the deferred.called status.

[Kapil Thangavelu (hazmat)]

Excerpts from Jim Baker's message of Tue Jun 21 00:18:33 UTC 2011:
> Review: Approve
> +1, we certainly need this branch, and it looks quite solid.
>
> In addition to Gustavo's comments:
>
> [1]
>
Added most of the grammatical changes.
>
> [2]
>
> + http://bit.ly/mQrOMY
> + http://bit.ly/irKpfn
>
> Why not just expand these links out? Alternatively use a readable URL
> (the underlying nabble URL is modestly readable, I don't know how
> permalink either one is, however).
>

The links are rather ugly by themselves and violate 80 col lines.

>
> [3]
>
> + assert callable(callback)
>
> Shouldn't this be throwing a TypeError instead? Also, there is no
> correponding assertion (or verification) on
> set_connection_error_callback. Lastly, there is no test on this path.

Well the tests definitely exercised it by way of using the functionality of the callback, but i've gone ahead and added the typeerror and invalid callback tests.

>
>
> [4]
>
> + def xtest_session_expired_event(self):
>
> Do you want to include this test?
>

nice catch, i did but it was before the implementation of the session work, i've moved the test to test_session

thanks for the review,

cheers,

Kapil

[Kapil Thangavelu (hazmat)]

gustavo, even though this branch is approved, i'm going to wait on a final comment/agreement on [4] before merging.

[Gustavo Niemeyer (niemeyer)]

Deep conversation about error handling in Ensemble, in the context of item [4] above.

We've agreed to merge this as-is, since clearly there's a lot of work to be made for the desired end result.

Jul 05 17:30:51 <niemeyer> Let me understand this.. are you introducing this logic so that the application we'll use this to crash the agent?
Jul 05 17:30:59 <niemeyer> s/we'll/will
Jul 05 17:31:14 <niemeyer> If that's the case, as we've already agreed, that's ok for now
Jul 05 17:31:15 <hazmat_> that's one approach
Jul 05 17:31:36 <hazmat_> the other is letting the agent stop/start
Jul 05 17:31:47 <niemeyer> I just want to understand what we're doing, though, and what would be the proper way
Jul 05 17:31:53 <hazmat_> i know.. there's dead stuff on the reactor if we start/stop
Jul 05 17:31:58 <niemeyer> Not just that..
Jul 05 17:32:10 <niemeyer> The dead stuff is not my major worry
Jul 05 17:32:21 <niemeyer> The real problem I see is that we simply can't trust any code anymore
Jul 05 17:32:33 <niemeyer> Imagine.. "Oh, I'll loop until file foo is created."
Jul 05 17:32:34 <niemeyer> NOT!
Jul 05 17:32:43 <niemeyer> The loop never dies..
Jul 05 17:32:55 <niemeyer> *anything* we do would be entirely unreliable
Jul 05 17:33:01 <niemeyer> if it's touching a watch
Jul 05 17:33:13 <niemeyer> It's like piecemeal crashing..
Jul 05 17:33:36 <hazmat_> right error handling is application specific
Jul 05 17:34:23 <hazmat_> with the existing watch protocols, its not an issue, except for the case of a hook currently executing (which needs cleanup even in the case of a process shutdown)
Jul 05 17:35:11 <niemeyer> I'm pretty sure I can find cases where it is an issue..
Jul 05 17:35:21 <niemeyer> relation-get ip
Jul 05 17:35:35 <niemeyer> relation-get goes into the agent..
Jul 05 17:35:45 <niemeyer> Never returns again..
Jul 05 17:35:52 <niemeyer> Dead processes
Jul 05 17:36:05 <hazmat_> that's the case i just mentioned
Jul 05 17:36:22 <niemeyer> I see
Jul 05 17:36:34 <niemeyer> I read "watch currently executed" when you said hook
Jul 05 17:36:56 <niemeyer> Anyway.. that's the kind of issue
Jul 05 17:37:07 <niemeyer> There are likely more problematic cases already
Jul 05 17:38:58 <hazmat_> yup.. the interesting case to me, would be something that where we our 'interactively' waiting on a watch, which we don't have, all the existing watch apis are effectively complete on their own (ie. request / response cycles).. if we want to consider arbitrary code.. sure we can construct problem scenarios.. but the existing usage i think is okay..
Jul 05 17:39:00 * hazmat_ brainstorms
Jul 05 17:39:24 <niemeyer> Well, it's already not ok as we have just both spotted
Jul 05 17:40:13 <hazmat_> well the hook scenario is very different then the dead logic waiting issue
Jul 05 17:40:19 <niemeyer> Why?
Jul 05 17:40:33 <hazmat_> it doesn't really care if the session terminates, it just needs an active connection to zk
Jul 05 17:41:06 <niemeyer> Hmmm
Jul 05 17:43:27 <hazmat_> pretty much all of the watch protocols/callbacks are based on examining the current state, and then effecting changes for their area of responsibility to match that current state.. if their restarted, they function t...

review: Approve

=== modified file 'txzookeeper/client.py'
--- txzookeeper/client.py	2011-06-17 23:52:34 +0000
+++ txzookeeper/client.py	2011-07-05 13:29:55 +0000
@@ -60,6 +60,24 @@
     zookeeper.SYSTEMERROR: zookeeper.SystemErrorException,
     zookeeper.UNIMPLEMENTED: zookeeper.UnimplementedException}
 
+# Mapping of connection state values to human strings.
+STATE_NAME_MAPPING = {
+    zookeeper.ASSOCIATING_STATE: "associating",
+    zookeeper.AUTH_FAILED_STATE: "auth-failed",
+    zookeeper.CONNECTED_STATE: "connected",
+    zookeeper.CONNECTING_STATE: "connecting",
+    zookeeper.EXPIRED_SESSION_STATE: "expired",
+}
+
+# Mapping of event type to human string.
+TYPE_NAME_MAPPING = {
+    zookeeper.NOTWATCHING_EVENT: "not-watching",
+    zookeeper.SESSION_EVENT: "session",
+    zookeeper.CREATED_EVENT: "created",
+    zookeeper.DELETED_EVENT: "deleted",
+    zookeeper.CHANGED_EVENT: "changed",
+    zookeeper.CHILD_EVENT: "child"}
+
 
 class NotConnectedException(zookeeper.ZooKeeperException):
     """
@@ -73,6 +91,29 @@
     Raised if an error occurs during the client's connection attempt.
     """
 
+    @property
+    def state_name(self):
+        return STATE_NAME_MAPPING[self.args[2]]
+
+    @property
+    def type_name(self):
+        return TYPE_NAME_MAPPING[self.args[1]]
+
+    def __str__(self):
+        return "<txzookeeper.ConnectionException type: %s state: %s>" % (
+            self.type_name, self.state_name)
+
+
+def is_connection_exception(e):
+    """
+    For connection errors in response to api calls, a utility method
+    to determine if the cause is a connection exception.
+    """
+    return isinstance(e,
+                      (zookeeper.ClosingException,
+                       zookeeper.ConnectionLossException,
+                       zookeeper.SessionExpiredException))
+
 
 class ConnectionTimeoutException(zookeeper.ZooKeeperException):
     """
@@ -87,94 +128,131 @@
     some event on the zookeeper client that the watch was requested on.
     """
 
-    type_name_map = {
-        -2: "notwatching",
-        -1: "session",
-        1: 'created',
-        2: 'deleted',
-        3: 'changed',
-        4: 'child'}
-
     @property
     def type_name(self):
-        return self.type_name_map[self.type]
+        return TYPE_NAME_MAPPING[self.type]
+
+    @property
+    def state_name(self):
+        return STATE_NAME_MAPPING[self.connection_state]
 
     def __repr__(self):
-        return  "<ClientEvent %s at %r>" % (self.type_name, self.path)
+        return  "<ClientEvent %s at %r state: %s>" % (
+            self.type_name, self.path, self.state_name)
 
 
 class ZookeeperClient(object):
     """Asynchronous twisted client for zookeeper."""
 
     def __init__(self, servers=None, session_timeout=None):
+        """
+        @param servers: A string specifying the servers and their
+                        ports to connect to. Multiple servers can be
+                        specified in comma separated fashion. if they are,
+                        then the client will automatically rotate
+                        among them if a server connection fails. Optionally
+                        a chroot can be specified. A full server spec looks
+                        like host:port/chroot_path
+
+        @param session_timeout: The client's zookeeper session timeout can be
+                       hinted. The actual value is negotiated between the
+                       client and server based on their respective
+                       configurations.
+        """
         self._servers = servers
         self._session_timeout = session_timeout
+        self._session_event_callback = None
+        self._connection_error_callback = None
         self.connected = False
         self.handle = None
 
-    def _check_connected(self):
+    def _check_connected(self, d):
         if not self.connected:
-            raise NotConnectedException("not connected")
-
-    def _check_result(self, result_code, callback=False, extra_codes=()):
+            d.errback(NotConnectedException("not connected"))
+            return d
+
+    def _check_result(self, result_code, deferred, extra_codes=()):
+        """Check an API call or result for errors.
+
+        :param result_code: The api result code.
+        :param deferred: The deferred returned the client api consumer.
+        :param extra_codes: Additional result codes accepted as valid/ok.
+
+        If the result code is an error, an appropriate Exception class
+        is constructed and the errback on the deferred is invoked with it.
+        """
         error = None
         if not result_code == zookeeper.OK and not result_code in extra_codes:
             error_msg = zookeeper.zerror(result_code)
             error_class = ERROR_MAPPING.get(
                 result_code, zookeeper.ZooKeeperException)
             error = error_class(error_msg)
-            if callback:
-                return error
-            raise error
+
+            if is_connection_exception(error):
+                # Mark the client as disconnected.
+                self.connected = False
+                # Route connection errors to a connection level error
+                # handler if specified.
+                if self._connection_error_callback:
+                    # The result of the connection error handler is returned
+                    # to the api.
+                    d = defer.maybeDeferred(
+                        self._connection_error_callback,
+                        self, error)
+                    d.chainDeferred(deferred)
+                    return True
+
+            deferred.errback(error)
+            return True
         return None
 
     def _get(self, path, watcher):
-        self._check_connected()
         d = defer.Deferred()
+        if self._check_connected(d):
+            return d
 
         def _cb_get(result_code, value, stat):
-            error = self._check_result(result_code, True)
-            if error:
-                return d.errback(error)
+            if self._check_result(result_code, d):
+                return
             d.callback((value, stat))
 
         callback = self._zk_thread_callback(_cb_get)
         watcher = self._wrap_watcher(watcher)
         result = zookeeper.aget(self.handle, path, watcher, callback)
-        self._check_result(result)
+        self._check_result(result, d)
         return d
 
     def _get_children(self, path, watcher):
-        self._check_connected()
         d = defer.Deferred()
+        if self._check_connected(d):
+            return d
 
         def _cb_get_children(result_code, children):
-            error = self._check_result(result_code, True)
-            if error:
-                return d.errback(error)
+            if self._check_result(result_code, d):
+                return
             d.callback(children)
 
         callback = self._zk_thread_callback(_cb_get_children)
         watcher = self._wrap_watcher(watcher)
         result = zookeeper.aget_children(self.handle, path, watcher, callback)
-        self._check_result(result)
+        self._check_result(result, d)
         return d
 
     def _exists(self, path, watcher):
-        self._check_connected()
         d = defer.Deferred()
+        if self._check_connected(d):
+            return d
 
         def _cb_exists(result_code, stat):
-            error = self._check_result(
-                result_code, True, extra_codes=(zookeeper.NONODE,))
-            if error:
-                return d.errback(error)
+            if self._check_result(
+                result_code, d, extra_codes=(zookeeper.NONODE,)):
+                return
             d.callback(stat)
 
         callback = self._zk_thread_callback(_cb_exists)
         watcher = self._wrap_watcher(watcher)
         result = zookeeper.aexists(self.handle, path, watcher, callback)
-        self._check_result(result)
+        self._check_result(result, d)
         return d
 
     def _wrap_watcher(self, watcher):
@@ -182,7 +260,22 @@
             return watcher
         if not callable(watcher):
             raise SyntaxError("invalid watcher")
-        return self._zk_thread_callback(watcher)
+        return self._zk_thread_callback(
+            partial(self._session_event_wrapper, watcher))
+
+    def _session_event_wrapper(self, watcher, event_type, conn_state, path):
+        """Watch wrapper that diverts session events to a connection callback.
+        """
+        # If it's a session event pass it to the session callback, else
+        # ignore it. Session events are sent repeatedly to watchers
+        # which we have modeled after deferred, which only accept a
+        # single return value.
+        if event_type == zookeeper.SESSION_EVENT:
+            if self._session_event_callback:
+                self._session_event_callback(
+                    self, ClientEvent(event_type, conn_state, path))
+        else:
+            return watcher(event_type, conn_state, path)
 
     def _zk_thread_callback(self, func):
         """
@@ -190,7 +283,6 @@
         any user defined callback so that they are called back in the main
         thread after, zookeeper calls the wrapper.
         """
-
         def wrapper(handle, *args):  # pragma: no cover
             reactor.callFromThread(func, *args)
         return wrapper
@@ -207,6 +299,7 @@
     def session_timeout(self):
         """
         What's the negotiated session timeout for this connection, in seconds.
+        If the client is not connected the value is None.
         """
         if self.connected:
             return zookeeper.recv_timeout(self.handle)
@@ -222,9 +315,14 @@
 
     @property
     def client_id(self):
-        """
-        The connection's client id, useful when introspecting the server logs
-        for specific client activity.
+        """Returns the client id that identifies the server side session.
+
+        A client id is a tuple represented by the session id and
+        session password. It can be used to manually connect to an
+        extant server session (which contains associated ephemeral
+        nodes and watches)/ The connection's client id is also useful
+        when introspecting the server logs for specific client
+        activity.
         """
         if self.handle is None:
             return None
@@ -239,34 +337,33 @@
         return bool(zookeeper.is_unrecoverable(self.handle))
 
     def add_auth(self, scheme, identity):
-        """
-        Adds an authentication identity to this connection. A connection
-        can use multiple authentication identities at the same time, all
-        are checked when verifying acls on a node.
+        """Adds an authentication identity to this connection.
+
+        A connection can use multiple authentication identities at the
+        same time, all are checked when verifying acls on a node.
 
         @param scheme: a string specifying a an authentication scheme
                        valid values include 'digest'.
-        @param identity: a string containingusername and password colon
-                      separated for example 'mary:apples'
+        @param identity: a string containing username and password colon
+                      separated, for example 'mary:apples'
         """
-        self._check_connected()
         d = defer.Deferred()
+        if self._check_connected(d):
+            return d
 
         def _cb_authenticated(result_code):
-            error = self._check_result(result_code, True)
-            if error:
-                return d.errback(error)
+            if self._check_result(result_code, d):
+                return
             d.callback(self)
 
         callback = self._zk_thread_callback(_cb_authenticated)
         result = zookeeper.add_auth(self.handle, scheme, identity, callback)
-        self._check_result(result)
+        self._check_result(result, d)
         return d
 
     def close(self, force=False):
         """
-        Close the underlying socket connection and zookeeper server side
-        session.
+        Close the underlying socket connection and server side session.
 
         @param force: boolean, require the connection to be closed now or
                       an exception be raised.
@@ -276,23 +373,31 @@
 
         result = zookeeper.close(self.handle)
         self.connected = False
-        self._check_result(result)
-        return result
-
-    def connect(self, servers=None, timeout=10):
+        d = defer.Deferred()
+
+        if self._check_result(result, d):
+            return d
+        d.callback(True)
+        return d
+
+    def connect(self, servers=None, timeout=10, client_id=None):
         """
         Establish a connection to the given zookeeper server(s).
 
         @param servers: A string specifying the servers and their ports to
-                        connect to.
+                        connect to. Multiple servers can be specified in
+                        comma separated fashion.
         @param timeout: How many seconds to wait on a connection to the
                         zookeeper servers.
+
+        @param session_id:
         @returns A deferred that's fired when the connection is established.
         """
         d = defer.Deferred()
 
         if self.connected:
-            raise zookeeper.ZooKeeperException("Already Connected")
+            return defer.fail(
+                zookeeper.ZooKeeperException("Already Connected"))
 
         # Use a scheduled function to ensure a timeout.
         def _check_timeout():
@@ -311,31 +416,45 @@
         if servers is not None:
             self._servers = servers
 
-        self.handle = zookeeper.init(
-            self._servers, callback, self._session_timeout)
+        # Assemble client id if specified.
+        if client_id:
+            self.handle = zookeeper.init(
+                self._servers, callback, self._session_timeout, client_id)
+        else:
+            self.handle = zookeeper.init(
+                self._servers, callback, self._session_timeout)
 
         return d
 
     def _cb_connected(
         self, scheduled_timeout, connect_deferred, type, state, path):
+        """This callback is invoked through the lifecycle of the connection.
 
+        It's used for all connection level events and session events.
+        """
         # Cancel the timeout delayed task if it hasn't fired.
         if scheduled_timeout.active():
             scheduled_timeout.cancel()
 
+        # Update connected boolean
+        if state == zookeeper.CONNECTED_STATE:
+            self.connected = True
+        elif state != zookeeper.CONNECTING_STATE:
+            self.connected = False
+
         if connect_deferred.called:
             # If we timed out and then connected, then close the conn.
-            if state == zookeeper.CONNECTED_STATE:
-                self.connected = True
+            if state == zookeeper.CONNECTED_STATE and scheduled_timeout.called:
                 self.close()
-            # If the client is reused across multiple connect/close
-            # cycles, and a previous connect timed out, then a
-            # subsequent connect may trigger the previous connect's
-            # handler notifying of a CONNECTING_STATE, ignore.
+
+            # Send session events to the callback, in addition to any
+            # duplicate session events that will be sent for extant watches.
+            if self._session_event_callback:
+                self._session_event_callback(
+                    self, ClientEvent(type, state, path))
+
             return
         elif state == zookeeper.CONNECTED_STATE:
-            # Connection established.
-            self.connected = True
             connect_deferred.callback(self)
             return
 
@@ -351,19 +470,19 @@
         @params acls: A list of dictionaries specifying permissions.
         @params flags: Node creation flags (ephemeral, sequence, persistent)
         """
-        self._check_connected()
         d = defer.Deferred()
+        if self._check_connected(d):
+            return d
 
         def _cb_created(result_code, path):
-            error = self._check_result(result_code, True)
-            if error:
-                return d.errback(error)
+            if self._check_result(result_code, d):
+                return
             d.callback(path)
 
         callback = self._zk_thread_callback(_cb_created)
         result = zookeeper.acreate(
             self.handle, path, data, acls, flags, callback)
-        self._check_result(result)
+        self._check_result(result, d)
         return d
 
     def delete(self, path, version=-1):
@@ -376,18 +495,18 @@
         @param path: the path of the node to be deleted.
         @param version: the integer version of the node.
         """
-        self._check_connected()
         d = defer.Deferred()
+        if self._check_connected(d):
+            return d
 
         def _cb_delete(result_code):
-            error = self._check_result(result_code, True)
-            if error:
-                return d.errback(error)
+            if self._check_result(result_code, d):
+                return
             d.callback(result_code)
 
         callback = self._zk_thread_callback(_cb_delete)
         result = zookeeper.adelete(self.handle, path, version, callback)
-        self._check_result(result)
+        self._check_result(result, d)
         return d
 
     def exists(self, path):
@@ -395,8 +514,9 @@
         Check that the given node path exists. Returns a deferred that
         holds the node stat information if the node exists (created,
         modified, version, etc.), or ``None`` if it does not exist.
+
+        @param path: The path of the node whose existence will be checked.
         """
-
         return self._exists(path, None)
 
     def exists_and_watch(self, path):
@@ -406,20 +526,22 @@
         In addition to the deferred method result, this method returns
         a deferred that is called back when the node is modified or
         removed (once).
+
+        @param path: The path of the node whose existence will be checked.
         """
-
         d = defer.Deferred()
 
-        def callback(*args):
+        def watcher(*args):
             d.callback(ClientEvent(*args))
-        return self._exists(path, callback), d
+        return self._exists(path, watcher), d
 
     def get(self, path):
         """
         Get the node's data for the given node path. Returns a
         deferred that holds the content of the node.
+
+        @param path: The path of the node whose content will be retrieved.
         """
-
         return self._get(path, None)
 
     def get_and_watch(self, path):
@@ -429,17 +551,20 @@
         In addition to the deferred method result, this method returns
         a deferred that is called back when the node is modified or
         removed (once).
+
+        @param path: The path of the node whose content will be retrieved.
         """
-
         d = defer.Deferred()
 
-        def callback(*args):
+        def watcher(*args):
             d.callback(ClientEvent(*args))
-        return self._get(path, callback), d
+        return self._get(path, watcher), d
 
     def get_children(self, path):
         """
         Get the ids of all children directly under the given path.
+
+        @param path: The path of the node whose children will be retrieved.
         """
         return self._get_children(path, None)
 
@@ -450,13 +575,14 @@
         In addition to the deferred method result, this method returns
         a deferred that is called back when a change happens on the
         provided path (once).
+
+        @param path: The path of the node whose children will be retrieved.
         """
-
         d = defer.Deferred()
 
-        def callback(*args):
+        def watcher(*args):
             d.callback(ClientEvent(*args))
-        return self._get_children(path, callback), d
+        return self._get_children(path, watcher), d
 
     def get_acl(self, path):
         """
@@ -464,19 +590,21 @@
 
         Each acl is a dictionary containing keys/values for scheme, id,
         and perms.
+
+        @param path: The path of the node whose acl will be retrieved.
         """
-        self._check_connected()
         d = defer.Deferred()
+        if self._check_connected(d):
+            return d
 
         def _cb_get_acl(result_code, acls, stat):
-            error = self._check_result(result_code, True)
-            if error:
-                return d.errback(error)
+            if self._check_result(result_code, d):
+                return
             d.callback((acls, stat))
 
         callback = self._zk_thread_callback(_cb_get_acl)
         result = zookeeper.aget_acl(self.handle, path, callback)
-        self._check_result(result)
+        self._check_result(result, d)
         return d
 
     def set_acl(self, path, acls, version=-1):
@@ -502,19 +630,19 @@
                         doesn't match the version on the server, then a
                         BadVersionException is raised.
         """
-        self._check_connected()
         d = defer.Deferred()
+        if self._check_connected(d):
+            return d
 
         def _cb_set_acl(result_code):
-            error = self._check_result(result_code, True)
-            if error:
-                return d.errback(error)
+            if self._check_result(result_code, d):
+                return
             d.callback(result_code)
 
         callback = self._zk_thread_callback(_cb_set_acl)
         result = zookeeper.aset_acl(
             self.handle, path, version, acls, callback)
-        self._check_result(result)
+        self._check_result(result, d)
         return d
 
     def set(self, path, data="", version=-1):
@@ -528,18 +656,18 @@
         @param data: The data to store on the node.
         @param version: Integer version value
         """
-        self._check_connected()
         d = defer.Deferred()
+        if self._check_connected(d):
+            return d
 
         def _cb_set(result_code, node_stat):
-            error = self._check_result(result_code, True)
-            if error:
-                return d.errback(error)
+            if self._check_result(result_code, d):
+                return
             d.callback(node_stat)
 
         callback = self._zk_thread_callback(_cb_set)
         result = zookeeper.aset(self.handle, path, data, version, callback)
-        self._check_result(result)
+        self._check_result(result, d)
         return d
 
     def set_connection_watcher(self, watcher):
@@ -549,25 +677,72 @@
 
         @param: watcher function
         """
+        if not callable(watcher):
+            raise SyntaxError("Invalid Watcher %r" % (watcher))
         watcher = self._wrap_watcher(watcher)
         zookeeper.set_watcher(self.handle, watcher)
 
+    def set_session_callback(self, callback):
+        """Set a callback to receive session events.
+
+        Session events are by default ignored. Interested applications
+        may choose to set a session event watcher on the connection
+        to receive session events. Session events are typically broadcast
+        by the libzookeeper library to all extant watchers, but the
+        twisted integration using deferreds is not capable of receiving
+        multiple values (session events and watch events), so this
+        client implementation instead provides for a user defined callback
+        to be invoked with them instead. The callback receives a single
+        parameter, the session event in the form of a ClientEvent instance.
+
+        Additional details on session events
+        ------------------------------------
+        http://bit.ly/mQrOMY
+        http://bit.ly/irKpfn
+        """
+        if not callable(callback):
+            raise TypeError("Invalid callback %r" % callback)
+        self._session_event_callback = callback
+
+    def set_connection_error_callback(self, callback):
+        """Set a callback to receive connection error exceptions.
+
+        By default the error will be raised when the client API
+        call is made. Setting a connection level error handler allows
+        applications to centralize their handling of connection loss,
+        instead of having to guard every zk interaction.
+
+        The callback receives two parameters, the client instance
+        and the exception.
+        """
+        if not callable(callback):
+            raise TypeError("Invalid callback %r" % callback)
+        self._connection_error_callback = callback
+
+    def set_determinstic_order(self, boolean):
+        """
+        The zookeeper client will by default randomize the server hosts
+        it will connect to unless this is set to True.
+
+        This is a global setting across connections.
+        """
+        zookeeper.deterministic_conn_order(bool(boolean))
+
     def sync(self, path="/"):
-        """
-        Flushes the zookeeper connection to the leader.
+        """Flushes the connected zookeeper server with the leader.
 
         @param path: The root path to flush, all child nodes are also flushed.
         """
-        self._check_connected()
         d = defer.Deferred()
+        if self._check_connected(d):
+            return d
 
         def _cb_sync(result_code, path):
-            error = self._check_result(result_code, True)
-            if error:
-                return d.errback(error)
+            if self._check_result(result_code, d):
+                return
             d.callback(path)
 
         callback = self._zk_thread_callback(_cb_sync)
         result = zookeeper.async(self.handle, path, callback)
-        self._check_result(result)
+        self._check_result(result, d)
         return d
 
=== modified file 'txzookeeper/queue.py'
--- txzookeeper/queue.py	2011-06-17 23:52:34 +0000
+++ txzookeeper/queue.py	2011-07-05 13:29:55 +0000
@@ -90,7 +90,7 @@
         def on_queue_items_changed(*args):
             """Event watcher on queue node child events."""
             if request.complete or not self._client.connected:
-                return # pragma: no cover
+                return  # pragma: no cover
 
             if request.processing_children:
                 # If deferred stack is currently processing a set of children
@@ -273,7 +273,7 @@
     """
 
     def _item_processed_callback(self, result_code, item_path):
-        return self._client.delete(item_path+"-processing")
+        return self._client.delete(item_path + "-processing")
 
     def _filter_children(self, children, suffix="-processing"):
         """
@@ -300,7 +300,7 @@
 
         def on_node_exists(stat, path):
             """Reserve the node for consumer processing."""
-            d = self._client.create(path+"-processing",
+            d = self._client.create(path + "-processing",
                                     flags=zookeeper.EPHEMERAL)
             d.addCallback(on_reservation_success, path)
             d.addErrback(on_reservation_failed)
@@ -315,7 +315,7 @@
 
         def on_get_node_failed(failure, path):
             """If we can't fetch the node, delete the processing node."""
-            d = self._client.delete(path+"-processing")
+            d = self._client.delete(path + "-processing")
 
             # propogate unexpected errors appropriately
             if not failure.check(zookeeper.NoNodeException):
@@ -372,7 +372,7 @@
 
     def __init__(self, path, client, acl=None, persistent=False):
         super(SerializedQueue, self).__init__(path, client, acl, persistent)
-        self._lock = Lock("%s/%s"%(self.path, "_lock"), client)
+        self._lock = Lock("%s/%s" % (self.path, "_lock"), client)
 
     def _item_processed_callback(self, result_code, item_path):
         return self._lock.release()
 
=== modified file 'txzookeeper/tests/__init__.py'
--- txzookeeper/tests/__init__.py	2011-06-17 23:52:34 +0000
+++ txzookeeper/tests/__init__.py	2011-07-05 13:29:55 +0000
@@ -36,8 +36,8 @@
 
     def tearDown(self):
         super(ZookeeperTestCase, self).tearDown()
-        zookeeper.set_log_stream(sys.stderr) # reset to default
-        zookeeper.set_debug_level(zookeeper.LOG_LEVEL_DEBUG)
+        #zookeeper.set_log_stream(sys.stderr)  # reset to default
+        #zookeeper.set_debug_level(zookeeper.LOG_LEVEL_DEBUG)
         self.log_file.close()
 
     def get_log(self):
 
=== added file 'txzookeeper/tests/common.py'
--- txzookeeper/tests/common.py	1970-01-01 00:00:00 +0000
+++ txzookeeper/tests/common.py	2011-07-05 13:29:55 +0000
@@ -0,0 +1,216 @@
+import os
+import os.path
+import shutil
+import subprocess
+import tempfile
+
+from itertools import chain
+from collections import namedtuple
+from glob import glob
+
+
+ServerInfo = namedtuple(
+    "ServerInfo", "server_id client_port election_port leader_port")
+
+
+class ManagedZooKeeper(object):
+    """Class to manage the running of a ZooKeeper instance for testing.
+
+    Note: no attempt is made to probe the ZooKeeper instance is
+    actually available, or that the selected port is free. In the
+    future, we may want to do that, especially when run in a
+    Hudson/Buildbot context, to ensure more test robustness."""
+
+    def __init__(self, software_path, server_info, peers=()):
+        """Define the ZooKeeper test instance.
+
+        @param install_path: The path to the install for ZK
+        @param port: The port to run the managed ZK instance
+        """
+        self.install_path = software_path
+        self.server_info = server_info
+        self.host = "127.0.0.1"
+        self.peers = peers
+        self.working_path = tempfile.mkdtemp()
+        self._running = False
+
+    def run(self):
+        """Run the ZooKeeper instance under a temporary directory.
+
+        Writes ZK log messages to zookeeper.log in the current directory.
+        """
+        config_path = os.path.join(self.working_path, "zoo.cfg")
+        log_path = os.path.join(self.working_path, "log")
+        log4j_path = os.path.join(self.working_path, "log4j.properties")
+        data_path = os.path.join(self.working_path, "data")
+
+        # various setup steps
+        if not os.path.exists(self.working_path):
+            os.mdir(self.working_path)
+        if not os.path.exists(log_path):
+            os.mkdir(log_path)
+        if not os.path.exists(data_path):
+            os.mkdir(data_path)
+
+        with open(config_path, "w") as config:
+            config.write("""
+tickTime=2000
+dataDir=%s
+clientPort=%s
+maxClientCnxns=0
+""" % (data_path, self.server_info.client_port))
+
+        # setup a replicated setup if peers are specified
+        if self.peers:
+            servers_cfg = []
+            for p in chain((self.server_info,), self.peers):
+                servers_cfg.append("server.%s=localhost:%s:%s" % (
+                    p.server_id, p.leader_port, p.election_port))
+
+            with open(config_path, "a") as config:
+                config.write("""
+initLimit=4
+syncLimit=2
+%s
+""" % ("\n".join(servers_cfg)))
+
+        # Write server ids into datadir
+        with open(os.path.join(data_path, "myid"), "w") as myid_file:
+            myid_file.write(str(self.server_info.server_id))
+
+        with open(log4j_path, "w") as log4j:
+            log4j.write("""
+# DEFAULT: console appender only
+log4j.rootLogger=INFO, ROLLINGFILE
+log4j.appender.ROLLINGFILE.layout=org.apache.log4j.PatternLayout
+log4j.appender.ROLLINGFILE.layout.ConversionPattern=%d{ISO8601} [myid:%X{myid}] - %-5p [%t:%C{1}@%L] - %m%n
+log4j.appender.ROLLINGFILE=org.apache.log4j.RollingFileAppender
+log4j.appender.ROLLINGFILE.Threshold=DEBUG
+log4j.appender.ROLLINGFILE.File=""" + (
+                            self.working_path + os.sep + "zookeeper.log\n"))
+
+        self.process = subprocess.Popen(
+            args=["java",
+                  "-cp", self.classpath,
+                  "-Dzookeeper.log.dir=%s" % log_path,
+                  "-Dzookeeper.root.logger=INFO,CONSOLE",
+                  "-Dlog4j.configuration=file:%s" % log4j_path,
+                  # "-Dlog4j.debug",
+                  "org.apache.zookeeper.server.quorum.QuorumPeerMain",
+                  config_path],
+            )
+        self._running = True
+
+    @property
+    def classpath(self):
+        """Get the classpath necessary to run ZooKeeper."""
+
+        # Two possibilities, as seen in zkEnv.sh:
+        # Check for a release - top-level zookeeper-*.jar?
+        jars = glob((os.path.join(
+            self.install_path, 'zookeeper-*.jar')))
+        if jars:
+            # Relase build (`ant package`)
+            jars.extend(glob(os.path.join(
+                self.install_path,
+                "lib/*.jar")))
+        else:
+            # Development build (plain `ant`)
+            jars = glob((os.path.join(
+                self.install_path, 'build/zookeeper-*.jar')))
+            jars.extend(glob(os.path.join(
+                self.install_path,
+                "build/lib/*.jar")))
+        return ":".join(jars)
+
+    @property
+    def address(self):
+        """Get the address of the ZooKeeper instance."""
+        return "%s:%s" % (self.host, self.client_port)
+
+    @property
+    def running(self):
+        return self._running
+
+    @property
+    def client_port(self):
+        return self.server_info.client_port
+
+    def reset(self):
+        """Stop the zookeeper instance, cleaning out its on disk-data."""
+        self.stop()
+        shutil.rmtree(os.path.join(self.working_path, "data"))
+        os.mkdir(os.path.join(self.working_path, "data"))
+        with open(os.path.join(self.working_path, "data", "myid"), "w") as fh:
+            fh.write(str(self.server_info.server_id))
+
+    def stop(self):
+        """Stop the Zookeeper instance, retaining on disk state."""
+        if not self._running:
+            return
+        self.process.terminate()
+        self.process.wait()
+        self._running = False
+
+    def destroy(self):
+        """Stop the ZooKeeper instance and destroy its on disk-state"""
+        # called by at exit handler, reimport to avoid cleanup race.
+        import shutil
+        self.stop()
+
+        shutil.rmtree(self.working_path)
+
+
+class ZookeeperCluster(object):
+
+    def __init__(self, install_path, size=3, port_offset=20000):
+        self._install_path = install_path
+        self._servers = []
+
+        # Calculate ports and peer group
+        port = port_offset
+        peers = []
+
+        for i in range(size):
+            port += i * 10
+            info = ServerInfo(i + 1, port, port + 1, port + 2)
+            peers.append(info)
+
+        # Instantiate Managed ZK Servers
+        for i in range(size):
+            server_peers = list(peers)
+            server_info = server_peers.pop(i)
+            self._servers.append(
+                ManagedZooKeeper(
+                    self._install_path, server_info, server_peers))
+
+    def __getitem__(self, k):
+        return self._servers[k]
+
+    def __iter__(self):
+        return iter(self._servers)
+
+    def start(self):
+        # Zookeeper client expresses a preference for either lower ports or
+        # lexographical ordering of hosts, to ensure that all servers have a
+        # chance to startup, start them in reverse order.
+        for server in reversed(list(self)):
+            server.run()
+        # Giving the servers a moment to start, decreases the overall time
+        # required for a client to successfully connect (2s vs. 4s without
+        # the sleep).
+        import time
+        time.sleep(2)
+
+    def stop(self):
+        for server in self:
+            server.stop()
+        self._servers = []
+
+    def terminate(self):
+        for server in self:
+            server.destroy()
+
+    def reset(self):
+        for server in self:
+            server.reset()
 
=== modified file 'txzookeeper/tests/mocker.py'
--- txzookeeper/tests/mocker.py	2011-06-08 21:14:20 +0000
+++ txzookeeper/tests/mocker.py	2011-07-05 13:29:55 +0000
@@ -1923,6 +1923,7 @@
     def run(self, path):
         return Mock(self.mocker, path)
 
+
 def mock_returner_recorder(mocker, event):
     """Events that lead to other events must return mock objects."""
     parent_path = event.path.parent_path
 
=== modified file 'txzookeeper/tests/test_client.py'
--- txzookeeper/tests/test_client.py	2011-06-17 23:52:34 +0000
+++ txzookeeper/tests/test_client.py	2011-07-05 13:29:55 +0000
@@ -23,19 +23,25 @@
 
 from twisted.internet.defer import Deferred
 from twisted.internet.base import DelayedCall
+from twisted.python.failure import Failure
 
 import zookeeper
 
+from mocker import ANY, MATCH
+from txzookeeper.tests import ZookeeperTestCase, utils
 from txzookeeper.client import (
     ZookeeperClient, ZOO_OPEN_ACL_UNSAFE, ConnectionTimeoutException,
-    ConnectionException, ClientEvent)
-
-from mocker import ANY
-from txzookeeper.tests import ZookeeperTestCase, utils
+    ConnectionException, NotConnectedException, ClientEvent)
 
 PUBLIC_ACL = ZOO_OPEN_ACL_UNSAFE
 
 
+def match_deferred(arg):
+    return isinstance(arg, Deferred)
+
+DEFERRED_MATCH = MATCH(match_deferred)
+
+
 class ClientTests(ZookeeperTestCase):
 
     def setUp(self):
@@ -47,9 +53,10 @@
         if self.client.connected:
             utils.deleteTree(handle=self.client.handle)
             self.client.close()
-        del self.client
+
         if self.client2 and self.client2.connected:
             self.client2.close()
+
         super(ClientTests, self).tearDown()
 
     def test_wb_connect_after_timeout(self):
@@ -113,8 +120,10 @@
         return d
 
     def test_client_event_repr(self):
-        event = ClientEvent(4, 'state', 'path')
-        self.assertEqual(repr(event), "<ClientEvent child at 'path'>")
+        event = ClientEvent(zookeeper.SESSION_EVENT,
+                            zookeeper.EXPIRED_SESSION_STATE, '')
+        self.assertEqual(repr(event),
+                         "<ClientEvent session at '' state: expired>")
 
     def test_client_event_attributes(self):
         event = ClientEvent(4, 'state', 'path')
@@ -123,6 +132,10 @@
         self.assertEqual(event.path, 'path')
         self.assertEqual(event, (4, 'state', 'path'))
 
+    def test_client_use_while_disconnected_returns_failure(self):
+        return self.assertFailure(
+            self.client.exists("/"), NotConnectedException)
+
     def test_create_ephemeral_node_and_close_connection(self):
         """
         The client can create transient nodes that are destroyed when the
@@ -307,14 +320,17 @@
             return self.client.create("/foobar-watched", "rabbit")
 
         def get_node(path):
-            return self.client.get_and_watch(path)
+            data, watch = self.client.get_and_watch(path)
+            return data.addCallback(lambda x: (watch,))
 
-        def new_connection((data, watch)):
+        def new_connection((watch,)):
             self.client2 = ZookeeperClient("127.0.0.1:2181")
-            return self.client2.connect(), watch
+            return self.client2.connect().addCallback(
+                lambda x, y=None, z=None: (x, watch))
 
         def trigger_watch((client, watch)):
             zookeeper.delete(self.client2.handle, "/foobar-watched")
+            self.client2.close()
             return watch
 
         def verify_watch(event):
@@ -385,11 +401,16 @@
         """
         d = self.client.connect()
 
+        def inject_error(result_code, d, extra_codes=None):
+            error = SyntaxError()
+            d.errback(error)
+            return error
+
         def check_exists(client):
             mock_client = self.mocker.patch(client)
             mock_client._check_result(
-                ANY, True, extra_codes=(zookeeper.NONODE,))
-            self.mocker.result(SyntaxError())
+                ANY, DEFERRED_MATCH, extra_codes=(zookeeper.NONODE,))
+            self.mocker.call(inject_error)
             self.mocker.replay()
             return client.exists("/zebra-moon")
 
@@ -655,20 +676,21 @@
         return d
 
     def test_get_children_with_error(self):
+        """If the result of an api call is an error, its propgated.
+        """
         d = self.client.connect()
 
         def get_children(client):
-            mock_client = self.mocker.patch(self.client)
-            mock_client._check_result(ANY, True)
-            self.mocker.result(SyntaxError())
-            self.mocker.replay()
+            # Get the children of a nonexistant node
             return client.get_children("/tower")
 
         def verify_failure(failure):
-            self.assertTrue(isinstance(failure.value, SyntaxError))
+            self.assertTrue(isinstance(failure, Failure))
+            self.assertTrue(
+                isinstance(failure.value, zookeeper.NoNodeException))
 
         d.addCallback(get_children)
-        d.addErrback(verify_failure)
+        d.addBoth(verify_failure)
         return d
 
     # seems to be a segfault on this one, must be running latest zk
@@ -825,7 +847,7 @@
         def verify_node_access(stat):
             self.assertEqual(stat['version'], 1)
             self.assertEqual(stat['dataLength'], 3)
-            self.assertTrue(failed) # we should have hit the errback
+            self.assertTrue(failed)  # we should have hit the errback
 
         d.addCallback(add_auth_one)
         d.addCallback(create_node)
@@ -905,10 +927,6 @@
         acl = dict(scheme="digest", id="a:b", perms=zookeeper.PERM_ALL)
 
         def set_acl(client):
-            mock_client = self.mocker.patch(client)
-            mock_client._check_result(ANY, True)
-            self.mocker.result(zookeeper.NoNodeException())
-            self.mocker.replay()
             return client.set_acl("/zebra-moon22", [acl])
 
         def verify_failure(failure):
@@ -946,23 +964,22 @@
         """
         d = self.client.connect()
 
-        def create_node(client):
-            return client.create("/moose")
+        def inject_error(result, d):
+            error = zookeeper.ZooKeeperException()
+            d.errback(error)
+            return error
 
         def get_acl(path):
-            mock_client = self.mocker.patch(self.client)
-            mock_client._check_result(ANY, True)
-            self.mocker.result(zookeeper.ZooKeeperException("foobar"))
-            self.mocker.replay()
-            return self.client.get_acl(path)
+            # Get the ACL of a nonexistant node
+            return self.client.get_acl("/moose")
 
         def verify_failure(failure):
+            self.assertTrue(isinstance(failure, Failure))
             self.assertTrue(
                 isinstance(failure.value, zookeeper.ZooKeeperException))
 
-        d.addCallback(create_node)
         d.addCallback(get_acl)
-        d.addErrback(verify_failure)
+        d.addBoth(verify_failure)
         return d
 
     def test_client_id(self):
@@ -1006,35 +1023,6 @@
         d.addCallback(verify_sync)
         return d
 
-    def test_sync_error(self):
-        """
-        On error the sync callback returns a an exception/failure.
-        """
-        d = self.client.connect()
-
-        def create_node(client):
-            return client.create("/abc")
-
-        def client_sync(path):
-            mock_client = self.mocker.patch(self.client)
-            mock_client._check_result(ANY, True)
-            self.mocker.result(zookeeper.ZooKeeperException("foobar"))
-            self.mocker.replay()
-            return self.client.sync(path)
-
-        def verify_failure(failure):
-            self.assertTrue(
-                isinstance(failure.value, zookeeper.ZooKeeperException))
-
-        def assert_failed(extra):
-            self.fail("Should have gone to errback")
-
-        d.addCallback(create_node)
-        d.addCallback(client_sync)
-        d.addCallback(assert_failed)
-        d.addErrback(verify_failure)
-        return d
-
     def test_property_servers(self):
         """
         The servers property of the client, shows which if any servers
@@ -1087,13 +1075,67 @@
             return client.set_connection_watcher(1)
 
         def verify_invalid(failure):
-            self.assertEqual(failure.value.args, ("invalid watcher",))
+            self.assertEqual(failure.value.args, ("Invalid Watcher 1",))
             self.assertTrue(isinstance(failure.value, SyntaxError))
 
         d.addCallback(set_invalid_watcher)
         d.addErrback(verify_invalid)
         return d
 
+    def xtest_session_expired_event(self):
+        """
+        A client session can be reattached to in a separate connection,
+        if the a session is expired, using the zookeeper connection will
+        raise a SessionExpiredException.
+        """
+        d = self.client.connect()
+
+        class StopTest(Exception):
+            pass
+
+        def new_connection_same_connection(client):
+            self.assertEqual(client.state, zookeeper.CONNECTED_STATE)
+            return ZookeeperClient("127.0.0.1:2181").connect(
+                client_id=client.client_id).addErrback(
+                guard_session_expired, client)
+
+        def guard_session_expired(failure, client):
+            # On occassion we get a session expired event while connecting.
+            failure.trap(ConnectionException)
+            self.assertEqual(failure.value.state_name, "expired")
+            # Stop the test from proceeding
+            raise StopTest()
+
+        def close_new_connection(client):
+            # Verify both connections are using same session
+            self.assertEqual(self.client.client_id, client.client_id)
+            self.assertEqual(client.state, zookeeper.CONNECTED_STATE)
+
+            # Closing one connection will close the session
+            client.close()
+
+            # Continued use of the other client will get a
+            # disconnect exception.
+            return self.client.exists("/")
+
+        def verify_original_closed(failure):
+            if not isinstance(failure, Failure):
+                self.fail("Test did not raise exception.")
+            failure.trap(
+                zookeeper.SessionExpiredException,
+                zookeeper.ConnectionLossException)
+
+            #print "client close"
+            self.client.close()
+            #print "creating new client for teardown"
+            return self.client.connect()
+
+        d.addCallback(new_connection_same_connection)
+        d.addCallback(close_new_connection)
+        d.addBoth(verify_original_closed)
+
+        return d
+
     def test_connect_with_server(self):
         """
         A client's servers can be specified in the connect method.
@@ -1164,14 +1206,14 @@
         All of the client apis (with the exception of connect) attempt
         to ensure the client is connected before executing an operation.
         """
-        self.assertRaises(
-            zookeeper.ZooKeeperException, self.client.get_children, "/abc")
-
-        self.assertRaises(
-            zookeeper.ZooKeeperException, self.client.create, "/abc")
-
-        self.assertRaises(
-            zookeeper.ZooKeeperException, self.client.set, "/abc", "123")
+        self.assertFailure(
+            self.client.get_children("/abc"), zookeeper.ZooKeeperException)
+
+        self.assertFailure(
+            self.client.create("/abc"), zookeeper.ZooKeeperException)
+
+        self.assertFailure(
+            self.client.set("/abc", "123"), zookeeper.ZooKeeperException)
 
     def test_connect_multiple_raises(self):
         """
@@ -1181,8 +1223,9 @@
         d = self.client.connect()
 
         def connect_again(client):
-            self.assertRaises(
-                zookeeper.ZooKeeperException, client.connect)
+            d = client.connect()
+            self.failUnlessFailure(d, zookeeper.ZooKeeperException)
+            return d
 
         d.addCallback(connect_again)
         return d
@@ -1199,18 +1242,18 @@
         d = self.client.connect()
 
         def verify_failure(client):
-            self.assertRaises(
-                zookeeper.ZooKeeperException, client.create, "/abc")
+            d = client.create("/abc")
+            self.failUnlessFailure(d, zookeeper.ZooKeeperException)
 
         d.addCallback(verify_failure)
         return d
 
     def test_connection_watcher(self):
         """
-        A connection watcher can be set that recieves notices on when the
-        connection state changes. Technically zookeeper would also use this as
-        a global watcher for node state changes, but zkpython doesn't expose
-        that api, as its mostly considered legacy.
+        A connection watcher can be set that receives notices on when
+        the connection state changes. Technically zookeeper would also
+        use this as a global watcher for node watches, but zkpython
+        doesn't expose that api, as its mostly considered legacy.
 
         its out of scope to simulate a connection level event within unit tests
         such as the server restarting.
@@ -1243,3 +1286,13 @@
         If the client is not connected, closing returns None.
         """
         self.assertEqual(self.client.close(), None)
+
+    def test_invalid_connection_error_callback(self):
+        self.assertRaises(TypeError,
+                          self.client.set_connection_error_callback,
+                          None)
+
+    def test_invalid_session_callback(self):
+        self.assertRaises(TypeError,
+                          self.client.set_session_callback,
+                          None)
 
=== modified file 'txzookeeper/tests/test_node.py'
--- txzookeeper/tests/test_node.py	2011-06-17 23:52:34 +0000
+++ txzookeeper/tests/test_node.py	2011-07-05 13:29:55 +0000
@@ -59,7 +59,7 @@
     def _make_digest_identity(self, credentials):
         user, password = credentials.split(":")
         digest = hashlib.new("sha1", credentials).digest()
-        return "%s:%s"%(user, base64.b64encode(digest))
+        return "%s:%s" % (user, base64.b64encode(digest))
 
     def test_node_name_and_path(self):
         """
 
=== modified file 'txzookeeper/tests/test_queue.py'
--- txzookeeper/tests/test_queue.py	2011-06-17 23:52:34 +0000
+++ txzookeeper/tests/test_queue.py	2011-07-05 13:29:55 +0000
@@ -175,7 +175,7 @@
         watch = Deferred()
         self.mocker.result((succeed(["entry-000000"]), watch))
 
-        item_path = "%s/%s"%(path, "entry-000000")
+        item_path = "%s/%s" % (path, "entry-000000")
         mock_client.get(item_path)
         self.mocker.result(fail(SyntaxError("x")))
         self.mocker.replay()
@@ -230,13 +230,13 @@
             producer_done = Deferred()
 
             def iteration(i):
-                if len(items) == (item_count-1):
+                if len(items) == (item_count - 1):
                     return producer_done.callback(None)
                 items.append(i)
                 queue.put(str(i))
 
             for i in range(item_count):
-                reactor.callLater(i*0.05, iteration, i)
+                reactor.callLater(i * 0.05, iteration, i)
             yield producer_done
             returnValue(items)
 
@@ -284,7 +284,7 @@
         def producer(start, offset):
             client = yield self.open_client()
             q = self.queue_factory(path, client)
-            for i in range(start, start+offset):
+            for i in range(start, start + offset):
                 yield q.put(str(i))
                 produce_results.append(str(i))
 
@@ -311,7 +311,7 @@
         yield DeferredList(
             [consumer(8), consumer(8), consumer(4)])
 
-        err = set(produce_results)-set(consume_results)
+        err = set(produce_results) - set(consume_results)
         self.assertFalse(err)
 
         self.assertEqual(len(consume_results), len(produce_results))
 
=== modified file 'txzookeeper/tests/test_security.py'
--- txzookeeper/tests/test_security.py	2011-06-17 23:52:34 +0000
+++ txzookeeper/tests/test_security.py	2011-07-05 13:29:55 +0000
@@ -73,9 +73,9 @@
     def connect_users(self, *users):
         clients = []
         for name in users:
-            ident_user = getattr(self, "ident_%s"%(name), None)
+            ident_user = getattr(self, "ident_%s" % (name), None)
             if ident_user is None:
-                raise AttributeError("Invalid User %s"%(name))
+                raise AttributeError("Invalid User %s" % (name))
             client = ZookeeperClient("127.0.0.1:2181", 3000)
             clients.append(client)
             yield self.open_and_authenticate(client, ident_user)
 
=== added file 'txzookeeper/tests/test_session.py'
--- txzookeeper/tests/test_session.py	1970-01-01 00:00:00 +0000
+++ txzookeeper/tests/test_session.py	2011-07-05 13:29:55 +0000
@@ -0,0 +1,245 @@
+
+import atexit
+import os
+
+import zookeeper
+
+from twisted.internet import reactor
+from twisted.internet.defer import inlineCallbacks, Deferred, returnValue
+
+from txzookeeper import ZookeeperClient
+from txzookeeper.client import NotConnectedException, ConnectionException
+
+from txzookeeper.tests.common import ZookeeperCluster
+from txzookeeper.tests import ZookeeperTestCase
+
+
+ZK_HOME = os.environ.get("ZOOKEEPER_PATH")
+assert ZK_HOME, "ZOOKEEPER_PATH environment variable must be defined"
+
+CLUSTER = ZookeeperCluster(ZK_HOME)
+atexit.register(lambda cluster: cluster.terminate(), CLUSTER)
+
+
+class ClientSessionTests(ZookeeperTestCase):
+
+    def setUp(self):
+        super(ClientSessionTests, self).setUp()
+        self.cluster.start()
+        self.client = None
+        self.client2 = None
+        zookeeper.deterministic_conn_order(True)
+        zookeeper.set_debug_level(0)
+
+    def sleep(self, delay):
+        """Non-blocking sleep."""
+        deferred = Deferred()
+        reactor.callLater(delay, deferred.callback, None)
+        return deferred
+
+    @property
+    def cluster(self):
+        return CLUSTER
+
+    def tearDown(self):
+        super(ClientSessionTests, self).tearDown()
+        self.cluster.reset()
+
+    @inlineCallbacks
+    def test_client_session_migration(self):
+        """A client will automatically rotate servers to ensure a connection.
+
+        A client connected to multiple servers, will transparently
+        migrate amongst them, as individual servers can no longer be
+        reached. A client's session will be maintined.
+        """
+        # Connect to the Zookeeper Cluster
+        servers = ",".join([s.address for s in self.cluster])
+        self.client = ZookeeperClient(servers)
+        yield self.client.connect()
+        yield self.client.create("/hello", flags=zookeeper.EPHEMERAL)
+
+        # Shutdown the server the client is connected to
+        self.cluster[0].stop()
+
+        # Wait for the shutdown and cycle, if we don't wait we'll
+        # get a zookeeper connectionloss exception on occassion.
+        yield self.sleep(0.1)
+
+        self.assertTrue(self.client.connected)
+        exists = yield self.client.exists("/hello")
+        self.assertTrue(exists)
+
+    @inlineCallbacks
+    def test_client_watch_migration(self):
+        """On server rotation, extant watches are still active.
+
+        A client connected to multiple servers, will transparently
+        migrate amongst them, as individual servers can no longer be
+        reached. Watch deferreds issued from the same client instance will
+        continue to function as the session is maintained.
+        """
+        session_events = []
+
+        def session_event_callback(connection, e):
+            session_events.append(e)
+
+        # Connect to the Zookeeper Cluster
+        servers = ",".join([s.address for s in self.cluster])
+        self.client = ZookeeperClient(servers)
+        self.client.set_session_callback(session_event_callback)
+        yield self.client.connect()
+
+        # Setup a watch
+        yield self.client.create("/hello")
+        exists_d, watch_d = self.client.exists_and_watch("/hello")
+        yield exists_d
+
+        # Shutdown the server the client is connected to
+        self.cluster[0].stop()
+
+        # Wait for the shutdown and cycle, if we don't wait we'll
+        # get occasionally get a  zookeeper connectionloss exception.
+        yield self.sleep(0.1)
+
+        # The session events that would have been ignored are sent
+        # to the session event callback.
+        self.assertTrue(session_events)
+        self.assertTrue(self.client.connected)
+
+        # If we delete the node, we'll see the watch fire.
+        yield self.client.delete("/hello")
+        event = yield watch_d
+        self.assertEqual(event.type_name, "deleted")
+        self.assertEqual(event.path, "/hello")
+
+    @inlineCallbacks
+    def test_connection_error_handler(self):
+        """A callback can be specified for connection errors.
+
+        We can specify a callback for connection errors, that
+        can perform recovery for a disconnected client, restablishing
+        """
+        @inlineCallbacks
+        def connection_error_handler(connection, error):
+            # On loss of the connection, reconnect the client w/ same session.
+            yield connection.connect(
+                self.cluster[1].address, client_id=connection.client_id)
+            returnValue(23)
+
+        self.client = ZookeeperClient(self.cluster[0].address)
+        self.client.set_connection_error_callback(connection_error_handler)
+        yield self.client.connect()
+
+        yield self.client.create("/hello")
+        exists_d, watch_d = self.client.exists_and_watch("/hello")
+        yield exists_d
+
+        # Shutdown the server the client is connected to
+        self.cluster[0].stop()
+        yield self.sleep(0.1)
+
+        # Results in connection loss exception, and invoking of error handler.
+        result = yield self.client.exists("/hello")
+
+        # The result of the error handler is returned to the api
+        self.assertEqual(result, 23)
+
+        exists = yield self.client.exists("/hello")
+        self.assertTrue(exists)
+
+    @inlineCallbacks
+    def test_client_session_expiration_event(self):
+        """A client which recieves a session expiration event.
+        """
+        session_events = []
+        events_received = Deferred()
+
+        def session_event_callback(connection, e):
+            session_events.append(e)
+            if len(session_events) == 4:
+                events_received.callback(True)
+
+        # Connect to a node in the cluster and establish a watch
+        self.client = ZookeeperClient(self.cluster[0].address)
+        self.client.set_session_callback(session_event_callback)
+        yield self.client.connect()
+
+        child_d, watch_d = self.client.get_children_and_watch("/")
+        yield child_d
+
+        # Connect a client to the same session on a different node.
+        self.client2 = ZookeeperClient(self.cluster[0].address)
+        yield self.client2.connect(client_id=self.client.client_id)
+
+        # Close the new client and wait for the event propogation
+        yield self.client2.close()
+
+        # It can take some time for this to propagate
+        yield events_received
+        self.assertEqual(len(session_events), 4)
+        self.assertEqual(session_events[-1].state_name, "expired")
+
+        # The connection is dead without reconnecting.
+        yield self.assertFailure(
+            self.client.exists("/"),
+            NotConnectedException, ConnectionException)
+
+        self.assertTrue(self.client.unrecoverable)
+
+        # If a reconnect attempt is made with a dead session id
+        #yield self.client.connect(client_id=self.client.client_id)
+
+    test_client_session_expiration_event.timeout = 10
+
+    @inlineCallbacks
+    def test_client_reconnect_session_on_different_server(self):
+        """On connection failure, An application can choose to use a
+        new connection with which to reconnect to a different member
+        of the zookeeper cluster, reacquiring the extant session.
+
+        A large obvious caveat to using a new client instance rather
+        than reconnecting the existing client, is that even though the
+        session has outstanding watches, the watch callbacks/deferreds
+        won't be active unless the client instance used to create them
+        is connected.
+        """
+        session_events = []
+
+        def session_event_callback(connection, e):
+            session_events.append(e)
+
+        # Connect to a node in the cluster and establish a watch
+        self.client = ZookeeperClient(self.cluster[2].address)
+        self.client.set_session_callback(session_event_callback)
+        yield self.client.connect()
+
+        yield self.client.create("/hello", flags=zookeeper.EPHEMERAL)
+        exists_d, watch_d = self.client.exists_and_watch("/hello")
+        yield exists_d
+
+        # Shutdown the server the client is connected to
+        self.cluster[2].stop()
+        yield self.sleep(0.1)
+
+        # Verify we got a session event regarding the down server
+        self.assertTrue(session_events)
+
+        # Open up a new connection to a different server with same session
+        self.client2 = ZookeeperClient(self.cluster[0].address)
+        yield self.client2.connect(client_id=self.client.client_id)
+
+        # Close the old disconnected client
+        self.client.close()
+
+        # Verify the ephemeral still exists
+        exists = yield self.client2.exists("/hello")
+        self.assertTrue(exists)
+
+        # Destroy the session and reconnect
+        self.client2.close()
+        yield self.client.connect(self.cluster[0].address)
+
+        # Ephemeral is destroyed when the session closed.
+        exists = yield self.client.exists("/hello")
+        self.assertFalse(exists)
 
=== modified file 'txzookeeper/tests/test_utils.py'
--- txzookeeper/tests/test_utils.py	2011-06-17 23:52:34 +0000
+++ txzookeeper/tests/test_utils.py	2011-07-05 13:29:55 +0000
@@ -34,7 +34,7 @@
     def update_function_increment(self, content, stat):
         if not content:
             return str(0)
-        return str(int(content)+1)
+        return str(int(content) + 1)
 
     def setUp(self):
         super(RetryChangeTest, self).setUp()
 
=== modified file 'txzookeeper/tests/utils.py'
--- txzookeeper/tests/utils.py	2011-06-17 23:52:34 +0000
+++ txzookeeper/tests/utils.py	2011-07-05 13:29:55 +0000
@@ -25,9 +25,9 @@
     Destroy all the nodes in zookeeper (typically under a chroot for testing)
     """
     for child in zookeeper.get_children(handle, path):
-        if child == "zookeeper": # skip the metadata node
+        if child == "zookeeper":  # skip the metadata node
             continue
-        child_path = "/"+("%s/%s"%(path, child)).strip("/")
+        child_path = "/" + ("%s/%s" % (path, child)).strip("/")
         try:
             deleteTree(child_path, handle)
             zookeeper.delete(handle, child_path, -1)