Merge lp:~rogpeppe/gozk/update-event-interface into lp:~juju/gozk/trunk

[Roger Peppe]

Updated event interface and factored-out server-starting interface.

Event interface
-----------

The various functions returning event channels are changed
to return channels that more accurately reflect the kinds
of notifications that can happen on those channels.
This makes programming with the wait channels less error-prone
(no need worry about the wrong kind of event arriving
on a wait channel) and more self-documenting.

The Event type is removed:
- the Type field is now redundant, as only one kind of event can arrive through any given channel.
- the Path field was always redundant, as it is always the same as the node that is being watched.
- the value of the State field is now sent to the session watcher.

When the server goes down, watch channels are simply closed.
I can't think of any scenario when a node watcher will do something
differently depending on how the server has failed, and if it wants
to, it can always interrogate whatever is watching the session notifications.

Likely points of controversy:

- the spelling of the new constants. I chose mixed case rather than
all-caps - YMMV.

- i'd like to think of a better spelling for event_CREATED etc,
now that those constants no longer need exporting.

- use of explicit zero rather than having a zero constant defined
for each type. I *think* that it's less confusing and simpler this
way than having NodeClosed (NodeDown), ChildClosed etc.
Can a session ever go down permanently?

Server-starting interface
-------------------

I realised that server.go has increased significantly in size and
that all the new code was logically independent of ZooKeeper itself.
This inspired me to factor out the code into a new package (currently
gozk/zk/service which will potentially enable similar server packages
at little cost. There are a number of other enhancements that can now
be made to this package without increasing the zk package's complexity.

I think the new packaging works well, but it's arguable whether zookeeper
clients should use the service package directly or whether zk should
return *service.Service itself.

One specific advantage of the former (in addition to reusability and
nicer modularity) is that zookeeper clients can query or set aspects
of the zk.Server; the advantage of the latter is that it adds a line of
code and requires clients to know about another package.

YMMV as always.

[lp:~rogpeppe/gozk/update-event-interface]

21. By Roger Peppe on 2011-09-29

Rewrite Event comment.

[Gustavo Niemeyer (niemeyer)]

2692 - event := Event{
2693 - Type: int(data.event_type),
2694 - Path: C.GoString(data.event_path),
2695 - State: int(data.connection_state),
2696 - }

The key problem with this proposal is the trashing of information coming
from the server. It's discarding not only why e.g. a state is being
trashed, but also details like the path which is related to an event.
This path isn't just being returned from the path that called the watch.
It's actually part of the wire protocol, and comes from the server side.

Trashing both the session state error information and the paths like that
is not a good idea.

review: Disapprove

[lp:~rogpeppe/gozk/update-event-interface]

22. By Roger Peppe on 2011-09-30

merged ChildStatus into NodeStatus.

23. By Roger Peppe on 2011-09-30

gofmt

24. By Roger Peppe on 2011-09-30

Fixed comments. Added LOG_NONE.

[Roger Peppe (rogpeppe)]

On 29 September 2011 18:56, Gustavo Niemeyer <email address hidden> wrote:
> Review: Disapprove
>
>
> 2692    -               event := Event{
> 2693    -                       Type:  int(data.event_type),
> 2694    -                       Path:  C.GoString(data.event_path),
> 2695    -                       State: int(data.connection_state),
> 2696    -               }
>
> The key problem with this proposal is the trashing of information coming
> from the server. It's discarding not only why e.g. a state is being
> trashed, but also details like the path which is related to an event.
> This path isn't just being returned from the path that called the watch.
> It's actually part of the wire protocol, and comes from the server side.
>
> Trashing both the session state error information and the paths like that
> is not a good idea.

there is no information in the paths. they are always the same as the path
of the request. if this ever changes, then i'd fully support adding the path,
but as it is Event.Path gives a misleading impression (for instance that Path
might name a child that has been added).

we are trashing no information from the server. if we are then we
should make sure that we panic immediately because the server is doing
something unexpected. [i've just added the requisite check].
the exact specification, that is, what callbacks may be made for a given
API call, is unclear because explanation seems to be omitted in
the ZooKeeper manual itself. i think we owe it to gozk's users to provide
a better defined interface, as you will need to know the details of
the interface
to use the API, and so people will rely on the undocumented behaviour
anyway.

PTAL. the latest version has only two kind of events - node changed events
and session status events. i really think this interface makes sense.

please reconsider.

[Gustavo Niemeyer (niemeyer)]

Let's have a single thread on this topic, please.

As I answered to your email:

>> The point isn't C or Go.. the issue is simply that there's real
>> information coming through the pipe being trashed in the suggested
>> interface.
>
> there really isn't.

The protocol provides three details to the watch. Two of them are
being dropped on the floor in your branch and not reaching the event
listener. If you disagree with this we wont' be able to agree on
anything else around that change.

Unmerged revisions

24. By Roger Peppe on 2011-09-30

Fixed comments. Added LOG_NONE.

23. By Roger Peppe on 2011-09-30

gofmt

22. By Roger Peppe on 2011-09-30

merged ChildStatus into NodeStatus.

21. By Roger Peppe on 2011-09-29

Rewrite Event comment.

20. By Roger Peppe on 2011-09-29

merged changes to rename branch.

=== modified file 'Makefile'
--- Makefile	2011-08-03 01:56:58 +0000
+++ Makefile	2011-09-30 13:35:26 +0000
@@ -2,10 +2,14 @@
 
 all: package
 
-TARG=gozk
+TARG=launchpad.net/gozk/zk
 
+GOFILES=\
+	server.go\
+	status.go\
+	
 CGOFILES=\
-	gozk.go\
+	zk.go\
 
 CGO_OFILES=\
 	helpers.o\
 
=== added directory 'example'
=== renamed file 'example.go' => 'example/example.go'
--- example.go	2011-08-03 01:47:25 +0000
+++ example/example.go	2011-09-30 13:35:26 +0000
@@ -1,30 +1,29 @@
 package main
 
 import (
-    "gozk"
+	"launchpad.net/gozk/zk"
 )
 
 func main() {
-    zk, session, err := gozk.Init("localhost:2181", 5000)
-    if err != nil {
-        println("Couldn't connect: " + err.String())
-        return
-    }
-
-    defer zk.Close()
-
-    // Wait for connection.
-    event := <-session
-    if event.State != gozk.STATE_CONNECTED {
-        println("Couldn't connect")
-        return
-    }
-
-    _, err = zk.Create("/counter", "0", 0, gozk.WorldACL(gozk.PERM_ALL))
-    if err != nil {
-        println(err.String())
-    } else {
-        println("Created!")
-    }
+	conn, session, err := zk.Dial("localhost:2181", 5e9)
+	if err != nil {
+		println("Couldn't connect: " + err.String())
+		return
+	}
+
+	defer conn.Close()
+
+	// Wait for connection.
+	event := <-session
+	if event != zk.SessionConnected {
+		println("Couldn't connect")
+		return
+	}
+
+	_, err = conn.Create("/counter", "0", 0, zk.WorldACL(zk.PERM_ALL))
+	if err != nil {
+		println(err.String())
+	} else {
+		println("Created!")
+	}
 }
-
 
=== modified file 'helpers.c'
--- helpers.c	2011-01-11 12:58:26 +0000
+++ helpers.c	2011-09-30 13:35:26 +0000
@@ -4,7 +4,6 @@
 #include <string.h>
 #include "helpers.h"
 
-
 static pthread_mutex_t watch_mutex = PTHREAD_MUTEX_INITIALIZER;
 static pthread_cond_t  watch_available = PTHREAD_COND_INITIALIZER;
 
@@ -38,8 +37,9 @@
     pthread_mutex_lock(&watch_mutex);
     {
         watch_data *data = malloc(sizeof(watch_data)); // XXX Check data.
+
+        data->event_type = event_type;
         data->connection_state = connection_state;
-        data->event_type = event_type;
         data->event_path = strdup(event_path); // XXX Check event_path.
         data->watch_context = watch_context;
         data->next = NULL;
 
=== added file 'reattach_test.go'
--- reattach_test.go	1970-01-01 00:00:00 +0000
+++ reattach_test.go	2011-09-30 13:35:26 +0000
@@ -0,0 +1,202 @@
+package zk_test
+
+import (
+	"bufio"
+	. "launchpad.net/gocheck"
+	"launchpad.net/gozk/zk/service"
+	"launchpad.net/gozk/zk"
+	"exec"
+	"flag"
+	"fmt"
+	"os"
+	"strings"
+	"testing"
+	"time"
+)
+
+var reattach = flag.Bool("zktest.reattach", false, "internal flag used for testing")
+var reattachRunDir = flag.String("zktest.rundir", "", "internal flag used for testing")
+var reattachAbnormalStop = flag.Bool("zktest.stop", false, "internal flag used for testing")
+
+// This is the reentrancy point for testing ZooKeeper servers
+// started by processes that are not direct children of the
+// testing process. This test always succeeds - the status
+// will be written to stdout and read by indirectServer.
+func TestStartNonChildServer(t *testing.T) {
+	if !*reattach {
+		// not re-entrant, so ignore this test.
+		return
+	}
+	err := startServer(*reattachRunDir, *reattachAbnormalStop)
+	if err != nil {
+		fmt.Printf("error:%v\n", err)
+		return
+	}
+	fmt.Printf("done\n")
+}
+
+func (s *S) startServer(c *C, abort bool) {
+	err := startServer(s.zkTestRoot, abort)
+	c.Assert(err, IsNil)
+}
+
+// startServerIndirect starts a ZooKeeper server that is not
+// a direct child of the current process. If abort is true,
+// the server will be started and then terminated abnormally.
+func (s *S) startServerIndirect(c *C, abort bool) {
+	if len(os.Args) == 0 {
+		c.Fatal("Cannot find self executable name")
+	}
+	cmd := exec.Command(
+		os.Args[0],
+		"-zktest.reattach",
+		"-zktest.rundir", s.zkTestRoot,
+		"-zktest.stop=", fmt.Sprint(abort),
+		"-test.run", "StartNonChildServer",
+	)
+	r, err := cmd.StdoutPipe()
+	c.Assert(err, IsNil)
+	defer r.Close()
+	if err := cmd.Start(); err != nil {
+		c.Fatalf("cannot start re-entrant gotest process: %v", err)
+	}
+	defer cmd.Wait()
+	bio := bufio.NewReader(r)
+	for {
+		line, err := bio.ReadSlice('\n')
+		if err != nil {
+			c.Fatalf("indirect server status line not found: %v", err)
+		}
+		s := string(line)
+		if strings.HasPrefix(s, "error:") {
+			c.Fatalf("indirect server error: %s", s[len("error:"):])
+		}
+		if s == "done\n" {
+			return
+		}
+	}
+	panic("not reached")
+}
+
+// startServer starts a ZooKeeper server, and terminates it abnormally
+// if abort is true.
+func startServer(runDir string, abort bool) os.Error {
+	s, err := zk.AttachServer(runDir)
+	if err != nil {
+		return fmt.Errorf("cannot attach to server at %q: %v", runDir, err)
+	}
+	srv := service.New(s)
+	if err := srv.Start(); err != nil {
+		return fmt.Errorf("cannot start server: %v", err)
+	}
+	if abort {
+		// Give it time to start up, then kill the server process abnormally,
+		// leaving the pid.txt file behind.
+		time.Sleep(0.5e9)
+		p, err := srv.Process()
+		if err != nil {
+			return fmt.Errorf("cannot get server process: %v", err)
+		}
+		defer p.Release()
+		if err := p.Kill(); err != nil {
+			return fmt.Errorf("cannot kill server process: %v", err)
+		}
+	}
+	return nil
+}
+
+func (s *S) checkCookie(c *C) {
+	conn, watch, err := zk.Dial(s.zkAddr, 5e9)
+	c.Assert(err, IsNil)
+
+	e, ok := <-watch
+	c.Assert(ok, Equals, true)
+	c.Assert(e.Ok(), Equals, true)
+
+	c.Assert(err, IsNil)
+	cookie, _, err := conn.Get("/testAttachCookie")
+	c.Assert(err, IsNil)
+	c.Assert(cookie, Equals, "testAttachCookie")
+	conn.Close()
+}
+
+// cases to test:
+// child server, stopped normally; reattach, start
+// non-direct child server, killed abnormally; reattach, start (->error), remove pid.txt; start
+// non-direct child server, still running; reattach, start (->error), stop, start
+// child server, still running; reattach, start (-> error)
+// child server, still running; reattach, stop, start.
+// non-direct child server, still running; reattach, stop, start.
+func (s *S) TestAttachServer(c *C) {
+	// Create a cookie so that we know we are reattaching to the same instance.
+	conn, _ := s.init(c)
+	_, err := conn.Create("/testAttachCookie", "testAttachCookie", 0, zk.WorldACL(zk.PERM_ALL))
+	c.Assert(err, IsNil)
+	s.checkCookie(c)
+	s.zkServer.Stop()
+	s.zkServer = nil
+
+	s.testAttachServer(c, (*S).startServer)
+	s.testAttachServer(c, (*S).startServerIndirect)
+	s.testAttachServerAbnormalTerminate(c, (*S).startServer)
+	s.testAttachServerAbnormalTerminate(c, (*S).startServerIndirect)
+
+	srv, err := zk.AttachServer(s.zkTestRoot)
+	c.Assert(err, IsNil)
+
+	s.zkServer = service.New(srv)
+	err = s.zkServer.Start()
+	c.Assert(err, IsNil)
+
+	conn, _ = s.init(c)
+	err = conn.Delete("/testAttachCookie", -1)
+	c.Assert(err, IsNil)
+}
+
+func (s *S) testAttachServer(c *C, start func(*S, *C, bool)) {
+	start(s, c, false)
+
+	s.checkCookie(c)
+
+	// try attaching to it while it is still running - it should fail.
+	a, err := zk.AttachServer(s.zkTestRoot)
+	c.Assert(err, IsNil)
+	srv := service.New(a)
+
+	err = srv.Start()
+	c.Assert(err, NotNil)
+
+	// stop it and then start it again - it should succeed.
+	err = srv.Stop()
+	c.Assert(err, IsNil)
+
+	err = srv.Start()
+	c.Assert(err, IsNil)
+
+	s.checkCookie(c)
+
+	err = srv.Stop()
+	c.Assert(err, IsNil)
+}
+
+func (s *S) testAttachServerAbnormalTerminate(c *C, start func(*S, *C, bool)) {
+	start(s, c, true)
+
+	// try attaching to it and starting - it should fail, because pid.txt
+	// won't have been removed.
+	a, err := zk.AttachServer(s.zkTestRoot)
+	c.Assert(err, IsNil)
+	srv := service.New(a)
+	err = srv.Start()
+	c.Assert(err, NotNil)
+
+	// stopping it should bring things back to normal.
+	err = srv.Stop()
+	c.Assert(err, IsNil)
+	err = srv.Start()
+	c.Assert(err, IsNil)
+
+	s.checkCookie(c)
+	err = srv.Stop()
+	c.Assert(err, IsNil)
+}
 
=== modified file 'retry_test.go'
--- retry_test.go	2011-08-19 01:51:37 +0000
+++ retry_test.go	2011-09-30 13:35:26 +0000
@@ -1,42 +1,41 @@
-package gozk_test
+package zk_test
 
 import (
 	. "launchpad.net/gocheck"
-	"gozk"
+	"launchpad.net/gozk/zk"
 	"os"
 )
 
 func (s *S) TestRetryChangeCreating(c *C) {
-	zk, _ := s.init(c)
+	conn, _ := s.init(c)
 
-	err := zk.RetryChange("/test", gozk.EPHEMERAL, gozk.WorldACL(gozk.PERM_ALL),
-		func(data string, stat gozk.Stat) (string, os.Error) {
+	err := conn.RetryChange("/test", zk.EPHEMERAL, zk.WorldACL(zk.PERM_ALL),
+		func(data string, stat *zk.Stat) (string, os.Error) {
 			c.Assert(data, Equals, "")
 			c.Assert(stat, IsNil)
 			return "new", nil
 		})
 	c.Assert(err, IsNil)
 
-	data, stat, err := zk.Get("/test")
+	data, stat, err := conn.Get("/test")
 	c.Assert(err, IsNil)
 	c.Assert(stat, NotNil)
 	c.Assert(stat.Version(), Equals, int32(0))
 	c.Assert(data, Equals, "new")
 
-	acl, _, err := zk.ACL("/test")
+	acl, _, err := conn.ACL("/test")
 	c.Assert(err, IsNil)
-	c.Assert(acl, Equals, gozk.WorldACL(gozk.PERM_ALL))
+	c.Assert(acl, Equals, zk.WorldACL(zk.PERM_ALL))
 }
 
 func (s *S) TestRetryChangeSetting(c *C) {
-	zk, _ := s.init(c)
+	conn, _ := s.init(c)
 
-	_, err := zk.Create("/test", "old", gozk.EPHEMERAL,
-		gozk.WorldACL(gozk.PERM_ALL))
+	_, err := conn.Create("/test", "old", zk.EPHEMERAL, zk.WorldACL(zk.PERM_ALL))
 	c.Assert(err, IsNil)
 
-	err = zk.RetryChange("/test", gozk.EPHEMERAL, []gozk.ACL{},
-		func(data string, stat gozk.Stat) (string, os.Error) {
+	err = conn.RetryChange("/test", zk.EPHEMERAL, []zk.ACL{},
+		func(data string, stat *zk.Stat) (string, os.Error) {
 			c.Assert(data, Equals, "old")
 			c.Assert(stat, NotNil)
 			c.Assert(stat.Version(), Equals, int32(0))
@@ -44,27 +43,26 @@
 		})
 	c.Assert(err, IsNil)
 
-	data, stat, err := zk.Get("/test")
+	data, stat, err := conn.Get("/test")
 	c.Assert(err, IsNil)
 	c.Assert(stat, NotNil)
 	c.Assert(stat.Version(), Equals, int32(1))
 	c.Assert(data, Equals, "brand new")
 
 	// ACL was unchanged by RetryChange().
-	acl, _, err := zk.ACL("/test")
+	acl, _, err := conn.ACL("/test")
 	c.Assert(err, IsNil)
-	c.Assert(acl, Equals, gozk.WorldACL(gozk.PERM_ALL))
+	c.Assert(acl, Equals, zk.WorldACL(zk.PERM_ALL))
 }
 
 func (s *S) TestRetryChangeUnchangedValueDoesNothing(c *C) {
-	zk, _ := s.init(c)
+	conn, _ := s.init(c)
 
-	_, err := zk.Create("/test", "old", gozk.EPHEMERAL,
-		gozk.WorldACL(gozk.PERM_ALL))
+	_, err := conn.Create("/test", "old", zk.EPHEMERAL, zk.WorldACL(zk.PERM_ALL))
 	c.Assert(err, IsNil)
 
-	err = zk.RetryChange("/test", gozk.EPHEMERAL, []gozk.ACL{},
-		func(data string, stat gozk.Stat) (string, os.Error) {
+	err = conn.RetryChange("/test", zk.EPHEMERAL, []zk.ACL{},
+		func(data string, stat *zk.Stat) (string, os.Error) {
 			c.Assert(data, Equals, "old")
 			c.Assert(stat, NotNil)
 			c.Assert(stat.Version(), Equals, int32(0))
@@ -72,7 +70,7 @@
 		})
 	c.Assert(err, IsNil)
 
-	data, stat, err := zk.Get("/test")
+	data, stat, err := conn.Get("/test")
 	c.Assert(err, IsNil)
 	c.Assert(stat, NotNil)
 	c.Assert(stat.Version(), Equals, int32(0)) // Unchanged!
@@ -80,14 +78,14 @@
 }
 
 func (s *S) TestRetryChangeConflictOnCreate(c *C) {
-	zk, _ := s.init(c)
+	conn, _ := s.init(c)
 
-	changeFunc := func(data string, stat gozk.Stat) (string, os.Error) {
+	changeFunc := func(data string, stat *zk.Stat) (string, os.Error) {
 		switch data {
 		case "":
 			c.Assert(stat, IsNil)
-			_, err := zk.Create("/test", "conflict", gozk.EPHEMERAL,
-				gozk.WorldACL(gozk.PERM_ALL))
+			_, err := conn.Create("/test", "conflict", zk.EPHEMERAL,
+				zk.WorldACL(zk.PERM_ALL))
 			c.Assert(err, IsNil)
 			return "<none> => conflict", nil
 		case "conflict":
@@ -100,11 +98,10 @@
 		return "can't happen", nil
 	}
 
-	err := zk.RetryChange("/test", gozk.EPHEMERAL, gozk.WorldACL(gozk.PERM_ALL),
-		changeFunc)
+	err := conn.RetryChange("/test", zk.EPHEMERAL, zk.WorldACL(zk.PERM_ALL), changeFunc)
 	c.Assert(err, IsNil)
 
-	data, stat, err := zk.Get("/test")
+	data, stat, err := conn.Get("/test")
 	c.Assert(err, IsNil)
 	c.Assert(data, Equals, "conflict => new")
 	c.Assert(stat, NotNil)
@@ -112,18 +109,17 @@
 }
 
 func (s *S) TestRetryChangeConflictOnSetDueToChange(c *C) {
-	zk, _ := s.init(c)
+	conn, _ := s.init(c)
 
-	_, err := zk.Create("/test", "old", gozk.EPHEMERAL,
-		gozk.WorldACL(gozk.PERM_ALL))
+	_, err := conn.Create("/test", "old", zk.EPHEMERAL, zk.WorldACL(zk.PERM_ALL))
 	c.Assert(err, IsNil)
 
-	changeFunc := func(data string, stat gozk.Stat) (string, os.Error) {
+	changeFunc := func(data string, stat *zk.Stat) (string, os.Error) {
 		switch data {
 		case "old":
 			c.Assert(stat, NotNil)
 			c.Assert(stat.Version(), Equals, int32(0))
-			_, err := zk.Set("/test", "conflict", 0)
+			_, err := conn.Set("/test", "conflict", 0)
 			c.Assert(err, IsNil)
 			return "old => new", nil
 		case "conflict":
@@ -136,10 +132,10 @@
 		return "can't happen", nil
 	}
 
-	err = zk.RetryChange("/test", gozk.EPHEMERAL, []gozk.ACL{}, changeFunc)
+	err = conn.RetryChange("/test", zk.EPHEMERAL, []zk.ACL{}, changeFunc)
 	c.Assert(err, IsNil)
 
-	data, stat, err := zk.Get("/test")
+	data, stat, err := conn.Get("/test")
 	c.Assert(err, IsNil)
 	c.Assert(data, Equals, "conflict => new")
 	c.Assert(stat, NotNil)
@@ -147,18 +143,17 @@
 }
 
 func (s *S) TestRetryChangeConflictOnSetDueToDelete(c *C) {
-	zk, _ := s.init(c)
+	conn, _ := s.init(c)
 
-	_, err := zk.Create("/test", "old", gozk.EPHEMERAL,
-		gozk.WorldACL(gozk.PERM_ALL))
+	_, err := conn.Create("/test", "old", zk.EPHEMERAL, zk.WorldACL(zk.PERM_ALL))
 	c.Assert(err, IsNil)
 
-	changeFunc := func(data string, stat gozk.Stat) (string, os.Error) {
+	changeFunc := func(data string, stat *zk.Stat) (string, os.Error) {
 		switch data {
 		case "old":
 			c.Assert(stat, NotNil)
 			c.Assert(stat.Version(), Equals, int32(0))
-			err := zk.Delete("/test", 0)
+			err := conn.Delete("/test", 0)
 			c.Assert(err, IsNil)
 			return "old => <deleted>", nil
 		case "":
@@ -170,55 +165,53 @@
 		return "can't happen", nil
 	}
 
-	err = zk.RetryChange("/test", gozk.EPHEMERAL, gozk.WorldACL(gozk.PERM_READ),
-		changeFunc)
+	err = conn.RetryChange("/test", zk.EPHEMERAL, zk.WorldACL(zk.PERM_READ), changeFunc)
 	c.Assert(err, IsNil)
 
-	data, stat, err := zk.Get("/test")
+	data, stat, err := conn.Get("/test")
 	c.Assert(err, IsNil)
 	c.Assert(data, Equals, "<deleted> => new")
 	c.Assert(stat, NotNil)
 	c.Assert(stat.Version(), Equals, int32(0))
 
 	// Should be the new ACL.
-	acl, _, err := zk.ACL("/test")
+	acl, _, err := conn.ACL("/test")
 	c.Assert(err, IsNil)
-	c.Assert(acl, Equals, gozk.WorldACL(gozk.PERM_READ))
+	c.Assert(acl, Equals, zk.WorldACL(zk.PERM_READ))
 }
 
 func (s *S) TestRetryChangeErrorInCallback(c *C) {
-	zk, _ := s.init(c)
+	conn, _ := s.init(c)
 
-	err := zk.RetryChange("/test", gozk.EPHEMERAL, gozk.WorldACL(gozk.PERM_ALL),
-		func(data string, stat gozk.Stat) (string, os.Error) {
+	err := conn.RetryChange("/test", zk.EPHEMERAL, zk.WorldACL(zk.PERM_ALL),
+		func(data string, stat *zk.Stat) (string, os.Error) {
 			return "don't use this", os.NewError("BOOM!")
 		})
 	c.Assert(err, NotNil)
-	c.Assert(err.Code(), Equals, gozk.ZSYSTEMERROR)
 	c.Assert(err.String(), Equals, "BOOM!")
 
-	stat, err := zk.Exists("/test")
+	stat, err := conn.Exists("/test")
 	c.Assert(err, IsNil)
 	c.Assert(stat, IsNil)
 }
 
 func (s *S) TestRetryChangeFailsReading(c *C) {
-	zk, _ := s.init(c)
+	conn, _ := s.init(c)
 
-	_, err := zk.Create("/test", "old", gozk.EPHEMERAL,
-		gozk.WorldACL(gozk.PERM_WRITE)) // Write only!
+	// Write only!
+	_, err := conn.Create("/test", "old", zk.EPHEMERAL, zk.WorldACL(zk.PERM_WRITE))
 	c.Assert(err, IsNil)
 
 	var called bool
-	err = zk.RetryChange("/test", gozk.EPHEMERAL, gozk.WorldACL(gozk.PERM_ALL),
-		func(data string, stat gozk.Stat) (string, os.Error) {
+	err = conn.RetryChange("/test", zk.EPHEMERAL, zk.WorldACL(zk.PERM_ALL),
+		func(data string, stat *zk.Stat) (string, os.Error) {
 			called = true
 			return "", nil
 		})
 	c.Assert(err, NotNil)
-	c.Assert(err.Code(), Equals, gozk.ZNOAUTH)
+	c.Assert(err, Equals, zk.ZNOAUTH)
 
-	stat, err := zk.Exists("/test")
+	stat, err := conn.Exists("/test")
 	c.Assert(err, IsNil)
 	c.Assert(stat, NotNil)
 	c.Assert(stat.Version(), Equals, int32(0))
@@ -227,22 +220,21 @@
 }
 
 func (s *S) TestRetryChangeFailsSetting(c *C) {
-	zk, _ := s.init(c)
+	conn, _ := s.init(c)
 
-	_, err := zk.Create("/test", "old", gozk.EPHEMERAL,
-		gozk.WorldACL(gozk.PERM_READ)) // Read only!
+	// Read only!
+	_, err := conn.Create("/test", "old", zk.EPHEMERAL, zk.WorldACL(zk.PERM_READ))
 	c.Assert(err, IsNil)
 
 	var called bool
-	err = zk.RetryChange("/test", gozk.EPHEMERAL, gozk.WorldACL(gozk.PERM_ALL),
-		func(data string, stat gozk.Stat) (string, os.Error) {
+	err = conn.RetryChange("/test", zk.EPHEMERAL, zk.WorldACL(zk.PERM_ALL),
+		func(data string, stat *zk.Stat) (string, os.Error) {
 			called = true
 			return "", nil
 		})
-	c.Assert(err, NotNil)
-	c.Assert(err.Code(), Equals, gozk.ZNOAUTH)
+	c.Assert(err, Equals, zk.ZNOAUTH)
 
-	stat, err := zk.Exists("/test")
+	stat, err := conn.Exists("/test")
 	c.Assert(err, IsNil)
 	c.Assert(stat, NotNil)
 	c.Assert(stat.Version(), Equals, int32(0))
@@ -251,23 +243,22 @@
 }
 
 func (s *S) TestRetryChangeFailsCreating(c *C) {
-	zk, _ := s.init(c)
+	conn, _ := s.init(c)
 
-	_, err := zk.Create("/test", "old", gozk.EPHEMERAL,
-		gozk.WorldACL(gozk.PERM_READ)) // Read only!
+	// Read only!
+	_, err := conn.Create("/test", "old", zk.EPHEMERAL, zk.WorldACL(zk.PERM_READ))
 	c.Assert(err, IsNil)
 
 	var called bool
-	err = zk.RetryChange("/test/sub", gozk.EPHEMERAL,
-		gozk.WorldACL(gozk.PERM_ALL),
-		func(data string, stat gozk.Stat) (string, os.Error) {
+	err = conn.RetryChange("/test/sub", zk.EPHEMERAL, zk.WorldACL(zk.PERM_ALL),
+		func(data string, stat *zk.Stat) (string, os.Error) {
 			called = true
 			return "", nil
 		})
 	c.Assert(err, NotNil)
-	c.Assert(err.Code(), Equals, gozk.ZNOAUTH)
+	c.Assert(err, Equals, zk.ZNOAUTH)
 
-	stat, err := zk.Exists("/test/sub")
+	stat, err := conn.Exists("/test/sub")
 	c.Assert(err, IsNil)
 	c.Assert(stat, IsNil)
 
 
=== added file 'server.go'
--- server.go	1970-01-01 00:00:00 +0000
+++ server.go	2011-09-30 13:35:26 +0000
@@ -0,0 +1,239 @@
+package zk
+
+import (
+	"bufio"
+	"bytes"
+	"fmt"
+	"io/ioutil"
+	"net"
+	"os"
+	"path/filepath"
+	"strings"
+)
+
+// Server represents a ZooKeeper server, its data and configuration files.
+type Server struct {
+	runDir     string
+	installDir string
+}
+
+func (srv *Server) Directory() string {
+	return srv.runDir
+}
+
+// CreateServer creates the directory runDir and sets up a ZooKeeper server
+// environment inside it. It is an error if runDir already exists.
+// The server will listen on the specified TCP port.
+// 
+// The ZooKeeper installation directory is specified by installDir.
+// If this is empty, a system default will be used.
+//
+// CreateServer does not start the server - the *Server that
+// is returned can be used with the service package, for example:
+// see launchpad.net/gozk/zk/service.
+func CreateServer(port int, runDir, installDir string) (*Server, os.Error) {
+	if err := os.Mkdir(runDir, 0777); err != nil {
+		return nil, err
+	}
+	srv := &Server{runDir: runDir, installDir: installDir}
+	if err := srv.writeLog4JConfig(); err != nil {
+		return nil, err
+	}
+	if err := srv.writeZooKeeperConfig(port); err != nil {
+		return nil, err
+	}
+	if err := srv.writeInstallDir(); err != nil {
+		return nil, err
+	}
+	return srv, nil
+}
+
+// AttachServer creates a new ZooKeeper Server instance
+// to operate inside an existing run directory, runDir.
+// The directory must have been created with CreateServer.
+func AttachServer(runDir string) (*Server, os.Error) {
+	srv := &Server{runDir: runDir}
+	if err := srv.readInstallDir(); err != nil {
+		return nil, fmt.Errorf("cannot read server install directory: %v", err)
+	}
+	return srv, nil
+}
+
+func (srv *Server) path(name string) string {
+	return filepath.Join(srv.runDir, name)
+}
+
+func (srv *Server) CheckAvailability() os.Error {
+	port, err := srv.NetworkPort()
+	if err != nil {
+		return fmt.Errorf("cannot get network port: %v", err)
+	}
+	l, err := net.Listen("tcp", fmt.Sprintf("localhost:%d", port))
+	if err != nil {
+		return fmt.Errorf("cannot listen on port %v: %v", port, err)
+	}
+	l.Close()
+	return nil
+}
+
+// ServerCommand returns the command used to start the
+// ZooKeeper server. It is provided for debugging and testing
+// purposes only.
+func (srv *Server) Command() ([]string, os.Error) {
+	cp, err := srv.classPath()
+	if err != nil {
+		return nil, fmt.Errorf("cannot get class path: %v", err)
+	}
+	return []string{
+		"java",
+		"-cp", strings.Join(cp, ":"),
+		"-Dzookeeper.root.logger=INFO,CONSOLE",
+		"-Dlog4j.configuration=file:" + srv.path("log4j.properties"),
+		"org.apache.zookeeper.server.quorum.QuorumPeerMain",
+		srv.path("zoo.cfg"),
+	}, nil
+}
+
+var log4jProperties = `
+log4j.rootLogger=INFO, CONSOLE
+log4j.appender.CONSOLE=org.apache.log4j.ConsoleAppender
+log4j.appender.CONSOLE.Threshold=INFO
+log4j.appender.CONSOLE.layout=org.apache.log4j.PatternLayout
+log4j.appender.CONSOLE.layout.ConversionPattern=%d{ISO8601} - %-5p [%t:%C{1}@%L] - %m%n
+`
+
+func (srv *Server) writeLog4JConfig() (err os.Error) {
+	return ioutil.WriteFile(srv.path("log4j.properties"), []byte(log4jProperties), 0666)
+}
+
+func (srv *Server) writeZooKeeperConfig(port int) (err os.Error) {
+	return ioutil.WriteFile(srv.path("zoo.cfg"), []byte(fmt.Sprintf(
+		"tickTime=2000\n"+
+			"dataDir=%s\n"+
+			"clientPort=%d\n"+
+			"maxClientCnxns=500\n",
+		srv.runDir, port)), 0666)
+}
+
+// NetworkPort returns the TCP port number that
+// the server is configured for.
+func (srv *Server) NetworkPort() (int, os.Error) {
+	f, err := os.Open(srv.path("zoo.cfg"))
+	if err != nil {
+		return 0, err
+	}
+	r := bufio.NewReader(f)
+	for {
+		line, err := r.ReadSlice('\n')
+		if err != nil {
+			return 0, fmt.Errorf("cannot get port from %q", srv.path("zoo.cfg"))
+		}
+		var port int
+		if n, _ := fmt.Sscanf(string(line), "clientPort=%d\n", &port); n == 1 {
+			return port, nil
+		}
+	}
+	panic("not reached")
+}
+
+func (srv *Server) writeInstallDir() os.Error {
+	return ioutil.WriteFile(srv.path("installdir.txt"), []byte(srv.installDir+"\n"), 0666)
+}
+
+func (srv *Server) readInstallDir() os.Error {
+	data, err := ioutil.ReadFile(srv.path("installdir.txt"))
+	if err != nil {
+		return err
+	}
+	if data[len(data)-1] == '\n' {
+		data = data[0 : len(data)-1]
+	}
+	srv.installDir = string(data)
+	return nil
+}
+
+func (srv *Server) classPath() ([]string, os.Error) {
+	dir := srv.installDir
+	if dir == "" {
+		return systemClassPath()
+	}
+	if err := checkDirectory(dir); err != nil {
+		return nil, err
+	}
+	// Two possibilities, as seen in zkEnv.sh:
+	// 1) locally built binaries (jars are in build directory)
+	// 2) release binaries
+	if build := filepath.Join(dir, "build"); checkDirectory(build) == nil {
+		dir = build
+	}
+	classPath, err := filepath.Glob(filepath.Join(dir, "zookeeper-*.jar"))
+	if err != nil {
+		panic(fmt.Errorf("glob for jar files: %v", err))
+	}
+	more, err := filepath.Glob(filepath.Join(dir, "lib/*.jar"))
+	if err != nil {
+		panic(fmt.Errorf("glob for lib jar files: %v", err))
+	}
+
+	classPath = append(classPath, more...)
+	if len(classPath) == 0 {
+		return nil, fmt.Errorf("zookeeper libraries not found in %q", dir)
+	}
+	return classPath, nil
+}
+
+const zookeeperEnviron = "/etc/zookeeper/conf/environment"
+
+func systemClassPath() ([]string, os.Error) {
+	f, err := os.Open(zookeeperEnviron)
+	if f == nil {
+		return nil, err
+	}
+	r := bufio.NewReader(f)
+	for {
+		line, err := r.ReadSlice('\n')
+		if err != nil {
+			break
+		}
+		if !bytes.HasPrefix(line, []byte("CLASSPATH=")) {
+			continue
+		}
+
+		// remove variable and newline
+		path := string(line[len("CLASSPATH=") : len(line)-1])
+
+		// trim white space
+		path = strings.Trim(path, " \t\r")
+
+		// strip quotes
+		if path[0] == '"' {
+			path = path[1 : len(path)-1]
+		}
+
+		// split on :
+		classPath := strings.Split(path, ":")
+
+		// split off $ZOOCFGDIR
+		if len(classPath) > 0 && classPath[0] == "$ZOOCFGDIR" {
+			classPath = classPath[1:]
+		}
+
+		if len(classPath) == 0 {
+			return nil, fmt.Errorf("empty class path in %q", zookeeperEnviron)
+		}
+		return classPath, nil
+	}
+	return nil, fmt.Errorf("no class path found in %q", zookeeperEnviron)
+}
+
+// checkDirectory returns an error if the given path
+// does not exist or is not a directory.
+func checkDirectory(path string) os.Error {
+	if info, err := os.Stat(path); err != nil || !info.IsDirectory() {
+		if err == nil {
+			err = &os.PathError{Op: "stat", Path: path, Error: os.NewError("is not a directory")}
+		}
+		return err
+	}
+	return nil
+}
 
=== added directory 'service'
=== added file 'service/Makefile'
--- service/Makefile	1970-01-01 00:00:00 +0000
+++ service/Makefile	2011-09-30 13:35:26 +0000
@@ -0,0 +1,20 @@
+include $(GOROOT)/src/Make.inc
+
+TARG=launchpad.net/gozk/zk/service
+
+GOFILES=\
+	service.go\
+
+GOFMT=gofmt
+BADFMT:=$(shell $(GOFMT) -l $(GOFILES) $(CGOFILES) $(wildcard *_test.go))
+
+gofmt: $(BADFMT)
+	@for F in $(BADFMT); do $(GOFMT) -w $$F && echo $$F; done
+
+ifneq ($(BADFMT),)
+ifneq ($(MAKECMDGOALS),gofmt)
+$(warning WARNING: make gofmt: $(BADFMT))
+endif
+endif
+
+include $(GOROOT)/src/Make.pkg
 
=== added file 'service/service.go'
--- service/service.go	1970-01-01 00:00:00 +0000
+++ service/service.go	2011-09-30 13:35:26 +0000
@@ -0,0 +1,160 @@
+// The service package provides support for long-running services.
+package service
+
+import (
+	"os"
+	"fmt"
+	"io/ioutil"
+	"strconv"
+	"path/filepath"
+	"exec"
+	"strings"
+	"time"
+)
+
+// Interface represents a single-process server.
+type Interface interface {
+	// Directory gives the path to the directory containing
+	// the server's configuration and data files. It is assumed that
+	// this exists and can be modified.
+	Directory() string
+
+	// CheckAvailability should return an error if resources
+	// required by the server are not available; for instance
+	// if the server requires a network port which is not free.
+	CheckAvailability() os.Error
+
+	// Command should return a command that can be
+	// used to run the server. The first element of the returned
+	// slice is the executable name; the rest of the elements are
+	// its arguments.
+	Command() ([]string, os.Error)
+}
+
+// Service represents a possibly running server process.
+type Service struct {
+	srv Interface
+}
+
+// New returns a new Service using the given
+// server interface.
+func New(srv Interface) *Service {
+	return &Service{srv}
+}
+
+// Process returns a reference to the identity
+// of the last running server in the Service's directory.
+// If the server was shut down, it returns (nil, nil).
+// The server process may not still be running
+// (for instance if it was terminated abnormally).
+// This function is provided for debugging and testing
+// purposes only.
+func (s *Service) Process() (*os.Process, os.Error) {
+	data, err := ioutil.ReadFile(s.path("pid.txt"))
+	if err != nil {
+		if err, ok := err.(*os.PathError); ok && err.Error == os.ENOENT {
+			return nil, nil
+		}
+		return nil, err
+	}
+	pid, err := strconv.Atoi(string(data))
+	if err != nil {
+		return nil, os.NewError("bad process id found in pid.txt")
+	}
+	return os.FindProcess(pid)
+}
+
+func (s *Service) path(name string) string {
+	return filepath.Join(s.srv.Directory(), name)
+}
+
+// Start starts the server. It returns an error if the server is already running.
+// It stores the process ID of the server in the file "pid.txt"
+// inside the server's directory. Stdout and stderr of the server
+// are similarly written to "log.txt".
+func (s *Service) Start() os.Error {
+	if err := s.srv.CheckAvailability(); err != nil {
+		return err
+	}
+	p, err := s.Process()
+	if p != nil || err != nil {
+		if p != nil {
+			p.Release()
+		}
+		return fmt.Errorf("server may already be running (remove %q to clear)", s.path("pid.txt"))
+	}
+
+	// create the pid file before starting the process so that if we get two
+	// programs trying to concurrently start a server on the same directory
+	// at the same time, only one should succeed.
+	pidf, err := os.OpenFile(s.path("pid.txt"), os.O_EXCL|os.O_CREATE|os.O_WRONLY, 0666)
+	if err != nil {
+		return fmt.Errorf("cannot create pid.txt: %v", err)
+	}
+	defer pidf.Close()
+
+	args, err := s.srv.Command()
+	if err != nil {
+		return fmt.Errorf("cannot determine command: %v", err)
+	}
+	cmd := exec.Command(args[0], args[1:]...)
+
+	logf, err := os.OpenFile(s.path("log.txt"), os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0666)
+	if err != nil {
+		return fmt.Errorf("cannot create log file: %v", err)
+	}
+	defer logf.Close()
+	cmd.Stdout = logf
+	cmd.Stderr = logf
+	if err := cmd.Start(); err != nil {
+		return fmt.Errorf("cannot start server: %v", err)
+	}
+	if _, err := fmt.Fprint(pidf, cmd.Process.Pid); err != nil {
+		return fmt.Errorf("cannot write pid file: %v", err)
+	}
+	return nil
+}
+
+// Stop kills the server. It is a no-op if it is already running.
+func (s *Service) Stop() os.Error {
+	p, err := s.Process()
+	if p == nil {
+		if err != nil {
+			return fmt.Errorf("cannot read process ID of server: %v", err)
+		}
+		return nil
+	}
+	defer p.Release()
+	if err := p.Kill(); err != nil && !strings.Contains(err.String(), "no such process") {
+		return fmt.Errorf("cannot kill server process: %v", err)
+	}
+	// ignore the error returned from Wait because there's little
+	// we can do about it - it either means that the process has just exited
+	// anyway or that we can't wait for it for some other reason,
+	// for example because it was originally started by some other process.
+	_, err = p.Wait(0)
+	if err != nil && strings.Contains(err.String(), "no child processes") {
+		// If we can't wait for the server, it's possible that it was running
+		// but not as a child of this process, so give it a little while
+		// to exit. TODO poll with kill(no signal)?
+		time.Sleep(0.5e9)
+	}
+
+	if err := os.Remove(s.path("pid.txt")); err != nil {
+		return fmt.Errorf("cannot remove server process ID file: %v", err)
+	}
+	return nil
+}
+
+// Destroy stops the server, and then removes its
+// directory and all its contents.
+// Warning: this will destroy all data associated with the server.
+func (s *Service) Destroy() os.Error {
+	if err := s.Stop(); err != nil {
+		return err
+	}
+	if err := os.RemoveAll(s.srv.Directory()); err != nil {
+		return err
+	}
+	return nil
+}
 
=== added file 'status.go'
--- status.go	1970-01-01 00:00:00 +0000
+++ status.go	2011-09-30 13:35:26 +0000
@@ -0,0 +1,116 @@
+package zk
+
+import (
+	"fmt"
+)
+
+// event types, as defined by libzookeeper
+const (
+	_ = iota
+	event_CREATED
+	event_DELETED
+	event_CHANGED
+	event_CHILD
+	event_SESSION     = -1
+	event_NOTWATCHING = -2
+)
+
+// The statusChan interface represents an event channel.
+// It translates from libzookeeper events to Go channel sends
+// of the appropriate type.
+type statusChan interface {
+	// close closes the wait channel.
+	close()
+
+	// send sends an event with the given type and connection
+	// state to the wait channel. It should do a non-blocking
+	// send and return false if the send would have blocked.
+	send(eventType, connectionState int, path string) bool
+}
+
+// SessionStatus represents the status of a ZooKeeper session.
+type SessionStatus int
+
+// Constants that represent the status of ZooKeeper
+// connection session.
+const (
+	_ SessionStatus = iota
+	SessionConnecting
+	SessionAssociating
+	SessionConnected
+	SessionExpired    SessionStatus = -112
+	SessionAuthFailed SessionStatus = -113
+)
+
+type sessionStatusChan chan SessionStatus
+
+func (c sessionStatusChan) close() {
+	close(c)
+}
+
+func (c sessionStatusChan) send(eventType, eventState int, path string) bool {
+	if eventType != event_SESSION {
+		panic(fmt.Errorf("unexpected event, type %v; status %v", eventType, eventState))
+	}
+	if path != "" {
+		panic(fmt.Errorf("non-empty path received on session event"))
+	}
+	select {
+	case c <- SessionStatus(eventState):
+	default:
+		return false
+	}
+	return true
+}
+
+// NodeStatus represents the status of a node after ExistsW,
+// GetW, or ChildrenW has been called on the node.
+type NodeStatus int
+
+// Constants that represent changes to a ZooKeeper node.
+const (
+	_                NodeStatus = iota
+	NodeDeleted                 // The node has been deleted.
+	NodeCreated                 // The node has been created.
+	NodeChanged                 // The node data has changed.
+	NodeChangedChild            // The number of children of the node has changed.
+)
+
+type nodeStatusChan struct {
+	c    chan NodeStatus
+	path string // stored so we can check that the server is behaving.
+}
+
+func newNodeStatusChan(path string) *nodeStatusChan {
+	return &nodeStatusChan{path: path, c: make(chan NodeStatus, 1)}
+}
+
+func (c *nodeStatusChan) close() {
+	close(c.c)
+}
+
+func (c *nodeStatusChan) send(eventType, eventState int, path string) bool {
+	if path != c.path {
+		panic(fmt.Errorf("unexpected path %q received on node change event (type %v, status %v)",
+			path, eventType, eventState))
+	}
+	var val NodeStatus
+	switch eventType {
+	case event_CREATED:
+		val = NodeCreated
+	case event_DELETED:
+		val = NodeDeleted
+	case event_CHANGED:
+		val = NodeChanged
+	case event_CHILD:
+		val = NodeChangedChild
+	default:
+		panic(fmt.Errorf("unexpected event, type %v; status %v", eventType, eventState))
+	}
+	select {
+	case c.c <- val:
+	default:
+		return false
+	}
+	return true
+}
 
=== modified file 'suite_test.go'
--- suite_test.go	2011-08-19 01:43:37 +0000
+++ suite_test.go	2011-09-30 13:35:26 +0000
@@ -1,94 +1,72 @@
-package gozk_test
+package zk_test
 
 import (
 	. "launchpad.net/gocheck"
-	"testing"
-	"io/ioutil"
-	"path"
 	"fmt"
+	"launchpad.net/gozk/zk/service"
+	"launchpad.net/gozk/zk"
 	"os"
-	"gozk"
+	"testing"
 	"time"
 )
 
 func TestAll(t *testing.T) {
-	TestingT(t)
+	if !*reattach {
+		TestingT(t)
+	}
 }
 
 var _ = Suite(&S{})
 
 type S struct {
-	zkRoot      string
-	zkTestRoot  string
-	zkTestPort  int
-	zkServerSh  string
-	zkServerOut *os.File
-	zkAddr      string
+	zkServer   *service.Service
+	zkTestRoot string
+	zkAddr     string
 
-	handles     []*gozk.ZooKeeper
-	events      []*gozk.Event
+	handles     []*zk.Conn
+	events      []zk.SessionStatus
 	liveWatches int
 	deadWatches chan bool
 }
 
-var logLevel = 0 //gozk.LOG_ERROR
-
-
-var testZooCfg = ("dataDir=%s\n" +
-	"clientPort=%d\n" +
-	"tickTime=2000\n" +
-	"initLimit=10\n" +
-	"syncLimit=5\n" +
-	"")
-
-var testLog4jPrp = ("log4j.rootLogger=INFO,CONSOLE\n" +
-	"log4j.appender.CONSOLE=org.apache.log4j.ConsoleAppender\n" +
-	"log4j.appender.CONSOLE.Threshold=DEBUG\n" +
-	"log4j.appender.CONSOLE.layout=org.apache.log4j.PatternLayout\n" +
-	"log4j.appender.CONSOLE.layout.ConversionPattern=%d{ISO8601} [myid:%X{myid}] - %-5p [%t:%C{1}@%L] - %m%n" +
-	"")
-
-func (s *S) init(c *C) (*gozk.ZooKeeper, chan gozk.Event) {
-	zk, watch, err := gozk.Init(s.zkAddr, 5e9)
+var logLevel = 0 // zk.LOG_ERROR
+
+func (s *S) init(c *C) (*zk.Conn, <-chan zk.SessionStatus) {
+	conn, watch, err := zk.Dial(s.zkAddr, 5e9)
 	c.Assert(err, IsNil)
 
-	s.handles = append(s.handles, zk)
-
-	event := <-watch
-
-	c.Assert(event.Type, Equals, gozk.EVENT_SESSION)
-	c.Assert(event.State, Equals, gozk.STATE_CONNECTED)
-
-	bufferedWatch := make(chan gozk.Event, 256)
-	bufferedWatch <- event
+	s.handles = append(s.handles, conn)
+
+	status := <-watch
+	c.Assert(status == zk.SessionConnected, Equals, true)
+
+	bufferedWatch := make(chan zk.SessionStatus, 256)
+	bufferedWatch <- status
 
 	s.liveWatches += 1
 	go func() {
-	loop:
 		for {
+			status, ok := <-watch
+			if !ok {
+				break
+			}
 			select {
-			case event, ok := <-watch:
-				if !ok {
-					close(bufferedWatch)
-					break loop
-				}
-				select {
-				case bufferedWatch <- event:
-				default:
-					panic("Too many events in buffered watch!")
-				}
+			case bufferedWatch <- status:
+			default:
+				panic("Too many events in buffered watch!")
 			}
 		}
+		close(bufferedWatch)
 		s.deadWatches <- true
 	}()
 
-	return zk, bufferedWatch
+	return conn, bufferedWatch
 }
 
 func (s *S) SetUpTest(c *C) {
-	c.Assert(gozk.CountPendingWatches(), Equals, 0,
+	c.Assert(zk.CountPendingWatches(), Equals, 0,
 		Bug("Test got a dirty watch state before running!"))
-	gozk.SetLogLevel(logLevel)
+	zk.SetLogLevel(logLevel)
 }
 
 func (s *S) TearDownTest(c *C) {
@@ -108,98 +86,38 @@
 	}
 
 	// Reset the list of handles.
-	s.handles = make([]*gozk.ZooKeeper, 0)
+	s.handles = make([]*zk.Conn, 0)
 
-	c.Assert(gozk.CountPendingWatches(), Equals, 0,
+	c.Assert(zk.CountPendingWatches(), Equals, 0,
 		Bug("Test left live watches behind!"))
 }
 
-// We use the suite set up and tear down to manage a custom zookeeper
+// We use the suite set up and tear down to manage a custom ZooKeeper
 //
 func (s *S) SetUpSuite(c *C) {
-
+	var err os.Error
 	s.deadWatches = make(chan bool)
 
-	var err os.Error
-
-	s.zkRoot = os.Getenv("ZKROOT")
-	if s.zkRoot == "" {
-		panic("You must define $ZKROOT ($ZKROOT/bin/zkServer.sh is needed)")
-	}
-
-	s.zkTestRoot = c.MkDir()
-	s.zkTestPort = 21812
-
-	println("ZooKeeper test server directory:", s.zkTestRoot)
-	println("ZooKeeper test server port:", s.zkTestPort)
-
-	s.zkAddr = fmt.Sprintf("localhost:%d", s.zkTestPort)
-
-	s.zkServerSh = path.Join(s.zkRoot, "bin/zkServer.sh")
-	s.zkServerOut, err = os.OpenFile(path.Join(s.zkTestRoot, "stdout.txt"),
-		os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0644)
-	if err != nil {
-		panic("Can't open stdout.txt file for server: " + err.String())
-	}
-
-	dataDir := path.Join(s.zkTestRoot, "data")
-	confDir := path.Join(s.zkTestRoot, "conf")
-
-	os.Mkdir(dataDir, 0755)
-	os.Mkdir(confDir, 0755)
-
-	err = os.Setenv("ZOOCFGDIR", confDir)
-	if err != nil {
-		panic("Can't set $ZOOCFGDIR: " + err.String())
-	}
-
-	zooCfg := []byte(fmt.Sprintf(testZooCfg, dataDir, s.zkTestPort))
-	err = ioutil.WriteFile(path.Join(confDir, "zoo.cfg"), zooCfg, 0644)
-	if err != nil {
-		panic("Can't write zoo.cfg: " + err.String())
-	}
-
-	log4jPrp := []byte(testLog4jPrp)
-	err = ioutil.WriteFile(path.Join(confDir, "log4j.properties"), log4jPrp, 0644)
-	if err != nil {
-		panic("Can't write log4j.properties: " + err.String())
-	}
-
-	s.StartZK()
+	// N.B. We meed to create a subdirectory because zk.CreateServer
+	// insists on creating its own directory.
+	s.zkTestRoot = c.MkDir() + "/zk"
+
+	port := 21812
+	s.zkAddr = fmt.Sprint("localhost:", port)
+
+	srv, err := zk.CreateServer(port, s.zkTestRoot, "")
+	if err != nil {
+		c.Fatal("Cannot set up server environment: ", err)
+	}
+	s.zkServer = service.New(srv)
+	err = s.zkServer.Start()
+	if err != nil {
+		c.Fatal("Cannot start ZooKeeper server: ", err)
+	}
 }
 
 func (s *S) TearDownSuite(c *C) {
-	s.StopZK()
-	s.zkServerOut.Close()
-}
-
-func (s *S) StartZK() {
-	attr := os.ProcAttr{Files: []*os.File{os.Stdin, s.zkServerOut, os.Stderr}}
-	proc, err := os.StartProcess(s.zkServerSh, []string{s.zkServerSh, "start"}, &attr)
-	if err != nil {
-		panic("Problem executing zkServer.sh start: " + err.String())
-	}
-
-	result, err := proc.Wait(0)
-	if err != nil {
-		panic(err.String())
-	} else if result.ExitStatus() != 0 {
-		panic("'zkServer.sh start' exited with non-zero status")
-	}
-}
-
-func (s *S) StopZK() {
-	attr := os.ProcAttr{Files: []*os.File{os.Stdin, s.zkServerOut, os.Stderr}}
-	proc, err := os.StartProcess(s.zkServerSh, []string{s.zkServerSh, "stop"}, &attr)
-	if err != nil {
-		panic("Problem executing zkServer.sh stop: " + err.String() +
-			" (look for runaway java processes!)")
-	}
-	result, err := proc.Wait(0)
-	if err != nil {
-		panic(err.String())
-	} else if result.ExitStatus() != 0 {
-		panic("'zkServer.sh stop' exited with non-zero status " +
-			"(look for runaway java processes!)")
+	if s.zkServer != nil {
+		s.zkServer.Stop() // TODO Destroy
 	}
 }
 
=== renamed file 'gozk.go' => 'zk.go'
--- gozk.go	2011-08-19 01:56:39 +0000
+++ zk.go	2011-09-30 13:35:26 +0000
@@ -1,12 +1,26 @@
-// gozk - Zookeeper support for the Go language
+// gozk - ZooKeeper support for the Go language
 //
 //   https://wiki.ubuntu.com/gozk
 //
 // Copyright (c) 2010-2011 Canonical Ltd.
-//
 // Written by Gustavo Niemeyer <[e-mail]>
 //
-package gozk
+// A note about channels as used for change notification in gozk.
+//
+// Event channels are used to provide notifications of changes to ZooKeeper
+// nodes, from GetW, ExistsW, etc. These are invariably one-shot - a single
+// value will be sent on the wait channel to signify the change and then the
+// channel will be closed. If the ZooKeeper server goes down when awaiting
+// a change, the channel will be closed without sending a value. Clients
+// can either check explicitly for the closed channel with:
+//
+//	if v, ok := <-wait; !ok {
+//	}
+//
+// or just test the received value against zero, which
+// signifies no value for each of the notification types.
+//
+package zk
 
 /*
 #cgo CFLAGS: -I/usr/include/c-client-src
@@ -27,17 +41,16 @@
 // -----------------------------------------------------------------------
 // Main constants and data types.
 
-// The main ZooKeeper object, created through the Init function.
-// Encapsulates all communication with ZooKeeper.
-type ZooKeeper struct {
-	watchChannels  map[uintptr]chan Event
+// Conn represents a connection to a set of ZooKeeper nodes.
+type Conn struct {
+	watchChannels  map[uintptr]statusChan
 	sessionWatchId uintptr
 	handle         *C.zhandle_t
 	mutex          sync.Mutex
 }
 
-// ClientId represents the established session in ZooKeeper.  This is only
-// useful to be passed back into the ReInit function.
+// ClientId represents an established ZooKeeper session.  It can be
+// passed into Redial to reestablish a connection to an existing session.
 type ClientId struct {
 	cId C.clientid_t
 }
@@ -51,84 +64,63 @@
 	Id     string
 }
 
-// Event channels are used to provide notifications of changes in the
-// ZooKeeper connection state and in specific node aspects.
-// 
-// There are two sources of events: the session channel obtained during
-// initialization with Init, and any watch channels obtained
-// through one of the W-suffixed functions (GetW, ExistsW, etc).
-// 
-// The session channel will only receive session-level events notifying
-// about critical and transient changes in the ZooKeeper connection
-// state (STATE_CONNECTED, STATE_EXPIRED_SESSION, etc).  On long
-// running applications the session channel must *necessarily* be
-// observed since certain events like session expirations require an
-// explicit reconnection and reestablishment of state (or bailing out).
-// Because of that, the buffer used on the session channel has a limited
-// size, and a panic will occur if too many events are not collected.
-//
-// Watch channels enable monitoring state for nodes, and the
-// moment they're fired depends on which function was called to
-// create them.  Note that, unlike in other ZooKeeper interfaces,
-// gozk will NOT dispatch unimportant session events such as
-// STATE_ASSOCIATING, STATE_CONNECTING and STATE_CONNECTED to
-// watch Event channels, since they are transient and disruptive
-// to the workflow.  Critical state changes such as expirations
-// are still delivered to all event channels, though, and the
-// transient events may be obsererved in the session channel.
-//
-// Since every watch channel may receive critical session events, events
-// received must not be handled blindly as if the watch requested has
-// been fired.  To facilitate such tests, Events offer the Ok method,
-// and they also have a good String method so they may be used as an
-// os.Error value if wanted. E.g.:
-//
-//     event := <-watch
-//     if !event.Ok() {
-//         err = event
-//         return
-//     }
-//
-// Note that closed channels will deliver zeroed Event, which means
-// event.Type is set to EVENT_CLOSED and event.State is set to STATE_CLOSED,
-// to facilitate handling.
-type Event struct {
-	Type  int
-	Path  string
-	State int
-}
+// Error represents a ZooKeeper error.
+type Error int
 
-// Error codes that may be used to verify the result of the
-// Code method from Error.
 const (
-	ZOK                      = C.ZOK
-	ZSYSTEMERROR             = C.ZSYSTEMERROR
-	ZRUNTIMEINCONSISTENCY    = C.ZRUNTIMEINCONSISTENCY
-	ZDATAINCONSISTENCY       = C.ZDATAINCONSISTENCY
-	ZCONNECTIONLOSS          = C.ZCONNECTIONLOSS
-	ZMARSHALLINGERROR        = C.ZMARSHALLINGERROR
-	ZUNIMPLEMENTED           = C.ZUNIMPLEMENTED
-	ZOPERATIONTIMEOUT        = C.ZOPERATIONTIMEOUT
-	ZBADARGUMENTS            = C.ZBADARGUMENTS
-	ZINVALIDSTATE            = C.ZINVALIDSTATE
-	ZAPIERROR                = C.ZAPIERROR
-	ZNONODE                  = C.ZNONODE
-	ZNOAUTH                  = C.ZNOAUTH
-	ZBADVERSION              = C.ZBADVERSION
-	ZNOCHILDRENFOREPHEMERALS = C.ZNOCHILDRENFOREPHEMERALS
-	ZNODEEXISTS              = C.ZNODEEXISTS
-	ZNOTEMPTY                = C.ZNOTEMPTY
-	ZSESSIONEXPIRED          = C.ZSESSIONEXPIRED
-	ZINVALIDCALLBACK         = C.ZINVALIDCALLBACK
-	ZINVALIDACL              = C.ZINVALIDACL
-	ZAUTHFAILED              = C.ZAUTHFAILED
-	ZCLOSING                 = C.ZCLOSING
-	ZNOTHING                 = C.ZNOTHING
-	ZSESSIONMOVED            = C.ZSESSIONMOVED
+	ZOK                      Error = C.ZOK
+	ZSYSTEMERROR             Error = C.ZSYSTEMERROR
+	ZRUNTIMEINCONSISTENCY    Error = C.ZRUNTIMEINCONSISTENCY
+	ZDATAINCONSISTENCY       Error = C.ZDATAINCONSISTENCY
+	ZCONNECTIONLOSS          Error = C.ZCONNECTIONLOSS
+	ZMARSHALLINGERROR        Error = C.ZMARSHALLINGERROR
+	ZUNIMPLEMENTED           Error = C.ZUNIMPLEMENTED
+	ZOPERATIONTIMEOUT        Error = C.ZOPERATIONTIMEOUT
+	ZBADARGUMENTS            Error = C.ZBADARGUMENTS
+	ZINVALIDSTATE            Error = C.ZINVALIDSTATE
+	ZAPIERROR                Error = C.ZAPIERROR
+	ZNONODE                  Error = C.ZNONODE
+	ZNOAUTH                  Error = C.ZNOAUTH
+	ZBADVERSION              Error = C.ZBADVERSION
+	ZNOCHILDRENFOREPHEMERALS Error = C.ZNOCHILDRENFOREPHEMERALS
+	ZNODEEXISTS              Error = C.ZNODEEXISTS
+	ZNOTEMPTY                Error = C.ZNOTEMPTY
+	ZSESSIONEXPIRED          Error = C.ZSESSIONEXPIRED
+	ZINVALIDCALLBACK         Error = C.ZINVALIDCALLBACK
+	ZINVALIDACL              Error = C.ZINVALIDACL
+	ZAUTHFAILED              Error = C.ZAUTHFAILED
+	ZCLOSING                 Error = C.ZCLOSING
+	ZNOTHING                 Error = C.ZNOTHING
+	ZSESSIONMOVED            Error = C.ZSESSIONMOVED
 )
 
+func (error Error) String() string {
+	return C.GoString(C.zerror(C.int(error))) // Static, no need to free it.
+}
+
+// zkError creates an appropriate error return from
+// a ZooKeeper status and the errno return from a C API
+// call.
+func zkError(rc C.int, cerr os.Error) os.Error {
+	code := Error(rc)
+	switch code {
+	case ZOK:
+		return nil
+
+	case ZSYSTEMERROR:
+		// If a ZooKeeper call returns ZSYSTEMERROR, then
+		// errno becomes significant. If errno has not been
+		// set, then we will return ZSYSTEMERROR nonetheless.
+		if cerr != nil {
+			return cerr
+		}
+	}
+	return code
+}
+
 // Constants for SetLogLevel.
 const (
+	LOG_NONE = 0,
 	LOG_ERROR = C.ZOO_LOG_LEVEL_ERROR
 	LOG_WARN  = C.ZOO_LOG_LEVEL_WARN
 	LOG_INFO  = C.ZOO_LOG_LEVEL_INFO
@@ -155,33 +147,6 @@
 	PERM_ALL = 0x1f
 )
 
-// Constants for Event Type.
-const (
-	EVENT_CREATED = iota + 1
-	EVENT_DELETED
-	EVENT_CHANGED
-	EVENT_CHILD
-	EVENT_SESSION     = -1
-	EVENT_NOTWATCHING = -2
-
-	// Doesn't really exist in zk, but handy for use in zeroed Event
-	// values (e.g. closed channels).
-	EVENT_CLOSED = 0
-)
-
-// Constants for Event State.
-const (
-	STATE_EXPIRED_SESSION = -112
-	STATE_AUTH_FAILED     = -113
-	STATE_CONNECTING      = 1
-	STATE_ASSOCIATING     = 2
-	STATE_CONNECTED       = 3
-
-	// Doesn't really exist in zk, but handy for use in zeroed Event
-	// values (e.g. closed channels).
-	STATE_CLOSED = 0
-)
-
 func init() {
 	if EPHEMERAL != C.ZOO_EPHEMERAL ||
 		SEQUENCE != C.ZOO_SEQUENCE ||
@@ -191,17 +156,17 @@
 		PERM_DELETE != C.ZOO_PERM_DELETE ||
 		PERM_ADMIN != C.ZOO_PERM_ADMIN ||
 		PERM_ALL != C.ZOO_PERM_ALL ||
-		EVENT_CREATED != C.ZOO_CREATED_EVENT ||
-		EVENT_DELETED != C.ZOO_DELETED_EVENT ||
-		EVENT_CHANGED != C.ZOO_CHANGED_EVENT ||
-		EVENT_CHILD != C.ZOO_CHILD_EVENT ||
-		EVENT_SESSION != C.ZOO_SESSION_EVENT ||
-		EVENT_NOTWATCHING != C.ZOO_NOTWATCHING_EVENT ||
-		STATE_EXPIRED_SESSION != C.ZOO_EXPIRED_SESSION_STATE ||
-		STATE_AUTH_FAILED != C.ZOO_AUTH_FAILED_STATE ||
-		STATE_CONNECTING != C.ZOO_CONNECTING_STATE ||
-		STATE_ASSOCIATING != C.ZOO_ASSOCIATING_STATE ||
-		STATE_CONNECTED != C.ZOO_CONNECTED_STATE {
+		event_CREATED != C.ZOO_CREATED_EVENT ||
+		event_DELETED != C.ZOO_DELETED_EVENT ||
+		event_CHANGED != C.ZOO_CHANGED_EVENT ||
+		event_CHILD != C.ZOO_CHILD_EVENT ||
+		event_SESSION != C.ZOO_SESSION_EVENT ||
+		event_NOTWATCHING != C.ZOO_NOTWATCHING_EVENT ||
+		SessionExpired != SessionStatus(C.ZOO_EXPIRED_SESSION_STATE) ||
+		SessionAuthFailed != SessionStatus(C.ZOO_AUTH_FAILED_STATE) ||
+		SessionConnecting != SessionStatus(C.ZOO_CONNECTING_STATE) ||
+		SessionAssociating != SessionStatus(C.ZOO_ASSOCIATING_STATE) ||
+		SessionConnected != SessionStatus(C.ZOO_CONNECTED_STATE) {
 
 		panic("OOPS: Constants don't match C counterparts")
 	}
@@ -222,155 +187,97 @@
 }
 
 // -----------------------------------------------------------------------
-// Event methods.
+// Session status methods.
 
 // Ok returns true in case the event reports zk as being in a usable state.
-func (e Event) Ok() bool {
+func (status SessionStatus) Ok() bool {
 	// That's really it for now. Anything else seems to mean zk
 	// can't be used at the moment.
-	return e.State == STATE_CONNECTED
+	return status == SessionConnected
 }
 
-func (e Event) String() (s string) {
-	switch e.State {
-	case STATE_EXPIRED_SESSION:
+func (status SessionStatus) String() (s string) {
+	switch status {
+	case 0:
+		s = "ZooKeeper connection closed"
+	case SessionExpired:
 		s = "ZooKeeper session expired"
-	case STATE_AUTH_FAILED:
+	case SessionAuthFailed:
 		s = "ZooKeeper authentication failed"
-	case STATE_CONNECTING:
+	case SessionConnecting:
 		s = "ZooKeeper connecting"
-	case STATE_ASSOCIATING:
+	case SessionAssociating:
 		s = "ZooKeeper still associating"
-	case STATE_CONNECTED:
+	case SessionConnected:
 		s = "ZooKeeper connected"
-	case STATE_CLOSED:
-		s = "ZooKeeper connection closed"
 	default:
-		s = fmt.Sprintf("unknown ZooKeeper state %d", e.State)
-	}
-	if e.Type == -1 || e.Type == EVENT_SESSION {
-		return
-	}
-	if s != "" {
-		s += "; "
-	}
-	switch e.Type {
-	case EVENT_CREATED:
-		s += "path created: "
-	case EVENT_DELETED:
-		s += "path deleted: "
-	case EVENT_CHANGED:
-		s += "path changed: "
-	case EVENT_CHILD:
-		s += "path children changed: "
-	case EVENT_NOTWATCHING:
-		s += "not watching: " // !?
-	case EVENT_SESSION:
-		// nothing
-	}
-	s += e.Path
-	return
-}
-
-// -----------------------------------------------------------------------
-// Error interface which maps onto the ZooKeeper error codes.
-
-type Error interface {
-	String() string
-	Code() int
-}
-
-type errorType struct {
-	zkrc C.int
-	err  os.Error
-}
-
-func newError(zkrc C.int, err os.Error) Error {
-	return &errorType{zkrc, err}
-}
-
-func (error *errorType) String() (result string) {
-	if error.zkrc == ZSYSTEMERROR && error.err != nil {
-		result = error.err.String()
-	} else {
-		result = C.GoString(C.zerror(error.zkrc)) // Static, no need to free it.
-	}
-	return
-}
-
-// Code returns the error code that may be compared against one of
-// the gozk.Z* constants.
-func (error *errorType) Code() int {
-	return int(error.zkrc)
-}
-
-// -----------------------------------------------------------------------
-// Stat interface which maps onto the ZooKeeper Stat struct.
-
-// We declare this as an interface rather than an actual struct because
-// this way we don't have to copy data around between the real C struct
-// and the Go one on every call.  Most uses will only touch a few elements,
-// or even ignore the stat entirely, so that's a win.
+		s = fmt.Sprintf("unknown ZooKeeper state %d", status)
+	}
+	return s
+}
+
+// -----------------------------------------------------------------------
 
 // Stat contains detailed information about a node.
-type Stat interface {
-	Czxid() int64
-	Mzxid() int64
-	CTime() int64
-	MTime() int64
-	Version() int32
-	CVersion() int32
-	AVersion() int32
-	EphemeralOwner() int64
-	DataLength() int32
-	NumChildren() int32
-	Pzxid() int64
-}
-
-type resultStat C.struct_Stat
-
-func (stat *resultStat) Czxid() int64 {
-	return int64(stat.czxid)
-}
-
-func (stat *resultStat) Mzxid() int64 {
-	return int64(stat.mzxid)
-}
-
-func (stat *resultStat) CTime() int64 {
-	return int64(stat.ctime)
-}
-
-func (stat *resultStat) MTime() int64 {
-	return int64(stat.mtime)
-}
-
-func (stat *resultStat) Version() int32 {
-	return int32(stat.version)
-}
-
-func (stat *resultStat) CVersion() int32 {
-	return int32(stat.cversion)
-}
-
-func (stat *resultStat) AVersion() int32 {
-	return int32(stat.aversion)
-}
-
-func (stat *resultStat) EphemeralOwner() int64 {
-	return int64(stat.ephemeralOwner)
-}
-
-func (stat *resultStat) DataLength() int32 {
-	return int32(stat.dataLength)
-}
-
-func (stat *resultStat) NumChildren() int32 {
-	return int32(stat.numChildren)
-}
-
-func (stat *resultStat) Pzxid() int64 {
-	return int64(stat.pzxid)
+type Stat struct {
+	c C.struct_Stat
+}
+
+func (stat *Stat) String() string {
+	return fmt.Sprintf(
+		"{Czxid: %d; Mzxid: %d; CTime: %d; MTime: %d; "+
+			"Version: %d; CVersion: %d; AVersion: %d; "+
+			"EphemeralOwner: %d; DataLength: %d; "+
+			"NumChildren: %d; Pzxid: %d}",
+		stat.Czxid(), stat.Mzxid(), stat.CTime(), stat.MTime(),
+		stat.Version(), stat.CVersion(), stat.AVersion(),
+		stat.EphemeralOwner(), stat.DataLength(),
+		stat.NumChildren(), stat.Pzxid(),
+	)
+}
+
+func (stat *Stat) Czxid() int64 {
+	return int64(stat.c.czxid)
+}
+
+func (stat *Stat) Mzxid() int64 {
+	return int64(stat.c.mzxid)
+}
+
+func (stat *Stat) CTime() int64 {
+	return int64(stat.c.ctime)
+}
+
+func (stat *Stat) MTime() int64 {
+	return int64(stat.c.mtime)
+}
+
+func (stat *Stat) Version() int32 {
+	return int32(stat.c.version)
+}
+
+func (stat *Stat) CVersion() int32 {
+	return int32(stat.c.cversion)
+}
+
+func (stat *Stat) AVersion() int32 {
+	return int32(stat.c.aversion)
+}
+
+func (stat *Stat) EphemeralOwner() int64 {
+	return int64(stat.c.ephemeralOwner)
+}
+
+func (stat *Stat) DataLength() int32 {
+	return int32(stat.c.dataLength)
+}
+
+func (stat *Stat) NumChildren() int32 {
+	return int32(stat.c.numChildren)
+}
+
+func (stat *Stat) Pzxid() int64 {
+	return int64(stat.c.pzxid)
 }
 
 // -----------------------------------------------------------------------
@@ -380,11 +287,12 @@
 
 // SetLogLevel changes the minimum level of logging output generated
 // to adjust the amount of information provided.
+// The default logging level is LOG_ERROR.
 func SetLogLevel(level int) {
 	C.zoo_set_debug_level(C.ZooLogLevel(level))
 }
 
-// Init initializes the communication with a ZooKeeper cluster. The provided
+// Dial initializes the communication with a ZooKeeper cluster. The provided
 // servers parameter may include multiple server addresses, separated
 // by commas, so that the client will automatically attempt to connect
 // to another server if one of them stops working for whatever reason.
@@ -395,82 +303,80 @@
 //
 // Session establishment is asynchronous, meaning that this function
 // will return before the communication with ZooKeeper is fully established.
-// The watch channel receives events of type SESSION_EVENT when any change
-// to the state of the established connection happens.  See the documentation
-// for the Event type for more details.
-func Init(servers string, recvTimeoutNS int64) (zk *ZooKeeper, watch chan Event, err Error) {
-	zk, watch, err = internalInit(servers, recvTimeoutNS, nil)
-	return
+// The returned channel is used to receive events about the current
+// status of the connection. On long running applications the session channel
+// must *necessarily* be observed since certain events like session expirations require an
+// explicit reconnection and reestablishment of state (or bailing out).
+// Because of that, the buffer used on the session channel has a limited
+// size, and a panic will occur if too many events are not collected.
+func Dial(servers string, recvTimeoutNS int64) (*Conn, <-chan SessionStatus, os.Error) {
+	return dial(servers, recvTimeoutNS, nil)
 }
 
-// Equivalent to Init, but attempt to reestablish an existing session
+// Redial is equivalent to Dial, but attempts to reestablish an existing session
 // identified via the clientId parameter.
-func ReInit(servers string, recvTimeoutNS int64, clientId *ClientId) (zk *ZooKeeper, watch chan Event, err Error) {
-	zk, watch, err = internalInit(servers, recvTimeoutNS, clientId)
-	return
+func Redial(servers string, recvTimeoutNS int64, clientId *ClientId) (*Conn, <-chan SessionStatus, os.Error) {
+	return dial(servers, recvTimeoutNS, clientId)
 }
 
-func internalInit(servers string, recvTimeoutNS int64, clientId *ClientId) (*ZooKeeper, chan Event, Error) {
-
-	zk := &ZooKeeper{}
-	zk.watchChannels = make(map[uintptr]chan Event)
+func dial(servers string, recvTimeoutNS int64, clientId *ClientId) (*Conn, <-chan SessionStatus, os.Error) {
+	conn := &Conn{}
+	conn.watchChannels = make(map[uintptr]statusChan)
 
 	var cId *C.clientid_t
 	if clientId != nil {
 		cId = &clientId.cId
 	}
 
-	watchId, watchChannel := zk.createWatch(true)
-	zk.sessionWatchId = watchId
+	watchChannel := make(sessionStatusChan, 32)
+	watchId := conn.createWatch(watchChannel)
+	conn.sessionWatchId = watchId
 
 	cservers := C.CString(servers)
-	handle, cerr := C.zookeeper_init(cservers, C.watch_handler, C.int(recvTimeoutNS / 1e6), cId, unsafe.Pointer(watchId), 0)
+	handle, cerr := C.zookeeper_init(cservers, C.watch_handler, C.int(recvTimeoutNS/1e6), cId, unsafe.Pointer(watchId), 0)
 	C.free(unsafe.Pointer(cservers))
 	if handle == nil {
-		zk.closeAllWatches()
-		return nil, nil, newError(ZSYSTEMERROR, cerr)
+		conn.closeAllWatches()
+		return nil, nil, zkError(C.int(ZSYSTEMERROR), cerr)
 	}
-	zk.handle = handle
+	conn.handle = handle
 	runWatchLoop()
-	return zk, watchChannel, nil
+	return conn, watchChannel, nil
 }
 
 // ClientId returns the client ID for the existing session with ZooKeeper.
 // This is useful to reestablish an existing session via ReInit.
-func (zk *ZooKeeper) ClientId() *ClientId {
-	return &ClientId{*C.zoo_client_id(zk.handle)}
+func (conn *Conn) ClientId() *ClientId {
+	return &ClientId{*C.zoo_client_id(conn.handle)}
 }
 
 // Close terminates the ZooKeeper interaction.
-func (zk *ZooKeeper) Close() Error {
-
-	// Protect from concurrency around zk.handle change.
-	zk.mutex.Lock()
-	defer zk.mutex.Unlock()
-
-	if zk.handle == nil {
+func (conn *Conn) Close() os.Error {
+
+	// Protect from concurrency around conn.handle change.
+	conn.mutex.Lock()
+	defer conn.mutex.Unlock()
+
+	if conn.handle == nil {
 		// ZooKeeper may hang indefinitely if a handler is closed twice,
 		// so we get in the way and prevent it from happening.
-		return newError(ZCLOSING, nil)
+		return ZCLOSING
 	}
-	rc, cerr := C.zookeeper_close(zk.handle)
+	rc, cerr := C.zookeeper_close(conn.handle)
 
-	zk.closeAllWatches()
+	conn.closeAllWatches()
 	stopWatchLoop()
 
-	// At this point, nothing else should need zk.handle.
-	zk.handle = nil
+	// At this point, nothing else should need conn.handle.
+	conn.handle = nil
 
-	if rc != C.ZOK {
-		return newError(rc, cerr)
-	}
-	return nil
+	return zkError(rc, cerr)
 }
 
 // Get returns the data and status from an existing node.  err will be nil,
 // unless an error is found. Attempting to retrieve data from a non-existing
 // node is an error.
-func (zk *ZooKeeper) Get(path string) (data string, stat Stat, err Error) {
+func (conn *Conn) Get(path string) (data string, stat *Stat, err os.Error) {
 
 	cpath := C.CString(path)
 	cbuffer := (*C.char)(C.malloc(bufferSize))
@@ -478,22 +384,22 @@
 	defer C.free(unsafe.Pointer(cpath))
 	defer C.free(unsafe.Pointer(cbuffer))
 
-	cstat := C.struct_Stat{}
-	rc, cerr := C.zoo_wget(zk.handle, cpath, nil, nil,
-		cbuffer, &cbufferLen, &cstat)
+	var cstat Stat
+	rc, cerr := C.zoo_wget(conn.handle, cpath, nil, nil, cbuffer, &cbufferLen, &cstat.c)
 	if rc != C.ZOK {
-		return "", nil, newError(rc, cerr)
+		return "", nil, zkError(rc, cerr)
 	}
+
 	result := C.GoStringN(cbuffer, cbufferLen)
-
-	return result, (*resultStat)(&cstat), nil
+	return result, &cstat, nil
 }
 
 // GetW works like Get but also returns a channel that will receive
-// a single Event value when the data or existence of the given ZooKeeper
-// node changes or when critical session events happen.  See the
-// documentation of the Event type for more details.
-func (zk *ZooKeeper) GetW(path string) (data string, stat Stat, watch chan Event, err Error) {
+// a single value, one of NodeChanged or NodeDeleted, when the
+// node contents change or it is deleted.
+// If the ZooKeeper server goes down before an
+// event happens, the channel will be closed.
+func (conn *Conn) GetW(path string) (data string, stat *Stat, watch <-chan NodeStatus, err os.Error) {
 
 	cpath := C.CString(path)
 	cbuffer := (*C.char)(C.malloc(bufferSize))
@@ -501,82 +407,94 @@
 	defer C.free(unsafe.Pointer(cpath))
 	defer C.free(unsafe.Pointer(cbuffer))
 
-	watchId, watchChannel := zk.createWatch(true)
+	watchChannel := newNodeStatusChan(path)
+	watchId := conn.createWatch(watchChannel)
 
-	cstat := C.struct_Stat{}
-	rc, cerr := C.zoo_wget(zk.handle, cpath,
-		C.watch_handler, unsafe.Pointer(watchId),
-		cbuffer, &cbufferLen, &cstat)
+	var cstat Stat
+	rc, cerr := C.zoo_wget(conn.handle, cpath, C.watch_handler, unsafe.Pointer(watchId), cbuffer, &cbufferLen, &cstat.c)
 	if rc != C.ZOK {
-		zk.forgetWatch(watchId)
-		return "", nil, nil, newError(rc, cerr)
+		conn.forgetWatch(watchId)
+		return "", nil, nil, zkError(rc, cerr)
 	}
 
 	result := C.GoStringN(cbuffer, cbufferLen)
-	return result, (*resultStat)(&cstat), watchChannel, nil
+	return result, &cstat, watchChannel.c, nil
 }
 
 // Children returns the children list and status from an existing node.
-// err will be nil, unless an error is found. Attempting to retrieve the
-// children list from a non-existent node is an error.
-func (zk *ZooKeeper) Children(path string) (children []string, stat Stat, err Error) {
+// Attempting to retrieve the children list from a non-existent node is an error.
+func (conn *Conn) Children(path string) (children []string, stat *Stat, err os.Error) {
 
 	cpath := C.CString(path)
 	defer C.free(unsafe.Pointer(cpath))
 
 	cvector := C.struct_String_vector{}
-	cstat := C.struct_Stat{}
-	rc, cerr := C.zoo_wget_children2(zk.handle, cpath, nil, nil,
-		&cvector, &cstat)
+	var cstat Stat
+	rc, cerr := C.zoo_wget_children2(conn.handle, cpath, nil, nil, &cvector, &cstat.c)
 
 	// Can't happen if rc != 0, but avoid potential memory leaks in the future.
 	if cvector.count != 0 {
 		children = parseStringVector(&cvector)
 	}
-	if rc != C.ZOK {
-		err = newError(rc, cerr)
+	if rc == C.ZOK {
+		stat = &cstat
 	} else {
-		stat = (*resultStat)(&cstat)
+		err = zkError(rc, cerr)
 	}
 	return
 }
 
 // ChildrenW works like Children but also returns a channel that will
-// receive a single Event value when a node is added or removed under the
-// provided path or when critical session events happen.  See the documentation
-// of the Event type for more details.
-func (zk *ZooKeeper) ChildrenW(path string) (children []string, stat Stat, watch chan Event, err Error) {
+// receive a single value, one of NodeDeleted or NodeChangedChild,
+// when a node is added or removed under the
+// provided path or when the node itself is deleted. If the ZooKeeper server
+// goes down before an event happens, the channel will be closed.
+func (conn *Conn) ChildrenW(path string) (children []string, stat *Stat, watch <-chan NodeStatus, err os.Error) {
 
 	cpath := C.CString(path)
 	defer C.free(unsafe.Pointer(cpath))
 
-	watchId, watchChannel := zk.createWatch(true)
+	watchChannel := newNodeStatusChan(path)
+	watchId := conn.createWatch(watchChannel)
 
 	cvector := C.struct_String_vector{}
-	cstat := C.struct_Stat{}
-	rc, cerr := C.zoo_wget_children2(zk.handle, cpath,
-		C.watch_handler, unsafe.Pointer(watchId),
-		&cvector, &cstat)
+	var cstat Stat
+	rc, cerr := C.zoo_wget_children2(conn.handle, cpath, C.watch_handler, unsafe.Pointer(watchId), &cvector, &cstat.c)
 
 	// Can't happen if rc != 0, but avoid potential memory leaks in the future.
 	if cvector.count != 0 {
 		children = parseStringVector(&cvector)
 	}
-	if rc != C.ZOK {
-		zk.forgetWatch(watchId)
-		err = newError(rc, cerr)
+	if rc == C.ZOK {
+		stat = &cstat
+		watch = watchChannel.c
 	} else {
-		stat = (*resultStat)(&cstat)
-		watch = watchChannel
+		conn.forgetWatch(watchId)
+		err = zkError(rc, cerr)
 	}
 	return
 }
 
+// an arguably nicer version of parseStringVector (untested)
+//func parseStringVector(c *C.struct_String_vector) []string {
+//	// make a Go slice onto the C data.
+//	cv := *(*[]*C.char)(&reflect.SliceHeader{
+//		Data: uintptr(unsafe.Pointer(c.data)),
+//		Len: c.count,
+//		Cap: c.count,
+//	})
+//	v := make([]string, c.count)
+//	for i, s := range cv {
+//		v[i] = C.GoString(s)
+//	}
+//	return v
+//}
+
 func parseStringVector(cvector *C.struct_String_vector) []string {
 	vector := make([]string, cvector.count)
 	dataStart := uintptr(unsafe.Pointer(cvector.data))
 	uintptrSize := unsafe.Sizeof(dataStart)
-	for i := 0; i != len(vector); i++ {
+	for i := range vector {
 		cpathPos := dataStart + uintptr(i)*uintptrSize
 		cpath := *(**C.char)(unsafe.Pointer(cpathPos))
 		vector[i] = C.GoString(cpath)
@@ -588,51 +506,50 @@
 // Exists checks if a node exists at the given path.  If it does,
 // stat will contain meta information on the existing node, otherwise
 // it will be nil.
-func (zk *ZooKeeper) Exists(path string) (stat Stat, err Error) {
+func (conn *Conn) Exists(path string) (stat *Stat, err os.Error) {
 	cpath := C.CString(path)
 	defer C.free(unsafe.Pointer(cpath))
 
-	cstat := C.struct_Stat{}
-	rc, cerr := C.zoo_wexists(zk.handle, cpath, nil, nil, &cstat)
+	var cstat Stat
+	rc, cerr := C.zoo_wexists(conn.handle, cpath, nil, nil, &stat.c)
 
 	// We diverge a bit from the usual here: a ZNONODE is not an error
 	// for an exists call, otherwise every Exists call would have to check
 	// for err != nil and err.Code() != ZNONODE.
 	if rc == C.ZOK {
-		stat = (*resultStat)(&cstat)
+		stat = &cstat
 	} else if rc != C.ZNONODE {
-		err = newError(rc, cerr)
+		err = zkError(rc, cerr)
 	}
 	return
 }
 
 // ExistsW works like Exists but also returns a channel that will
-// receive an Event value when a node is created in case the returned
-// stat is nil and the node didn't exist, or when the existing node
-// is removed. It will also receive critical session events. See the
-// documentation of the Event type for more details.
-func (zk *ZooKeeper) ExistsW(path string) (stat Stat, watch chan Event, err Error) {
+// receive a single value, one of NodeCreated, NodeDeleted or NodeChanged,
+// when the node is next created, removed, or its contents change.
+// If the ZooKeeper server goes down before an event happens, the channel will be closed.
+func (conn *Conn) ExistsW(path string) (stat *Stat, watch <-chan NodeStatus, err os.Error) {
 	cpath := C.CString(path)
 	defer C.free(unsafe.Pointer(cpath))
 
-	watchId, watchChannel := zk.createWatch(true)
+	watchChannel := newNodeStatusChan(path)
+	watchId := conn.createWatch(watchChannel)
 
-	cstat := C.struct_Stat{}
-	rc, cerr := C.zoo_wexists(zk.handle, cpath,
-		C.watch_handler, unsafe.Pointer(watchId), &cstat)
+	var cstat Stat
+	rc, cerr := C.zoo_wexists(conn.handle, cpath, C.watch_handler, unsafe.Pointer(watchId), &cstat.c)
 
 	// We diverge a bit from the usual here: a ZNONODE is not an error
 	// for an exists call, otherwise every Exists call would have to check
 	// for err != nil and err.Code() != ZNONODE.
-	switch rc {
+	switch Error(rc) {
 	case ZOK:
-		stat = (*resultStat)(&cstat)
-		watch = watchChannel
+		stat = &cstat
+		watch = watchChannel.c
 	case ZNONODE:
-		watch = watchChannel
+		watch = watchChannel.c
 	default:
-		zk.forgetWatch(watchId)
-		err = newError(rc, cerr)
+		conn.forgetWatch(watchId)
+		err = zkError(rc, cerr)
 	}
 	return
 }
@@ -646,7 +563,7 @@
 // The returned path is useful in cases where the created path may differ
 // from the requested one, such as when a sequence number is appended
 // to it due to the use of the gozk.SEQUENCE flag.
-func (zk *ZooKeeper) Create(path, value string, flags int, aclv []ACL) (pathCreated string, err Error) {
+func (conn *Conn) Create(path, value string, flags int, aclv []ACL) (pathCreated string, err os.Error) {
 	cpath := C.CString(path)
 	cvalue := C.CString(value)
 	defer C.free(unsafe.Pointer(cpath))
@@ -660,13 +577,13 @@
 	cpathCreated := (*C.char)(C.malloc(cpathLen))
 	defer C.free(unsafe.Pointer(cpathCreated))
 
-	rc, cerr := C.zoo_create(zk.handle, cpath, cvalue, C.int(len(value)),
-		caclv, C.int(flags), cpathCreated, C.int(cpathLen))
-	if rc != C.ZOK {
-		return "", newError(rc, cerr)
+	rc, cerr := C.zoo_create(conn.handle, cpath, cvalue, C.int(len(value)), caclv, C.int(flags), cpathCreated, C.int(cpathLen))
+	if rc == C.ZOK {
+		pathCreated = C.GoString(cpathCreated)
+	} else {
+		err = zkError(rc, cerr)
 	}
-
-	return C.GoString(cpathCreated), nil
+	return
 }
 
 // Set modifies the data for the existing node at the given path, replacing it
@@ -677,35 +594,31 @@
 //
 // It is an error to attempt to set the data of a non-existing node with
 // this function. In these cases, use Create instead.
-func (zk *ZooKeeper) Set(path, value string, version int32) (stat Stat, err Error) {
+func (conn *Conn) Set(path, value string, version int32) (stat *Stat, err os.Error) {
 
 	cpath := C.CString(path)
 	cvalue := C.CString(value)
 	defer C.free(unsafe.Pointer(cpath))
 	defer C.free(unsafe.Pointer(cvalue))
 
-	cstat := C.struct_Stat{}
-
-	rc, cerr := C.zoo_set2(zk.handle, cpath, cvalue, C.int(len(value)),
-		C.int(version), &cstat)
-	if rc != C.ZOK {
-		return nil, newError(rc, cerr)
+	var cstat Stat
+	rc, cerr := C.zoo_set2(conn.handle, cpath, cvalue, C.int(len(value)), C.int(version), &cstat.c)
+	if rc == C.ZOK {
+		stat = &cstat
+	} else {
+		err = zkError(rc, cerr)
 	}
-
-	return (*resultStat)(&cstat), nil
+	return
 }
 
 // Delete removes the node at path. If version is not -1, the operation
 // will only succeed if the node is still at this version when the
 // node is deleted as an atomic operation.
-func (zk *ZooKeeper) Delete(path string, version int32) (err Error) {
+func (conn *Conn) Delete(path string, version int32) (err os.Error) {
 	cpath := C.CString(path)
 	defer C.free(unsafe.Pointer(cpath))
-	rc, cerr := C.zoo_delete(zk.handle, cpath, C.int(version))
-	if rc != C.ZOK {
-		return newError(rc, cerr)
-	}
-	return nil
+	rc, cerr := C.zoo_delete(conn.handle, cpath, C.int(version))
+	return zkError(rc, cerr)
 }
 
 // AddAuth adds a new authentication certificate to the ZooKeeper
@@ -713,7 +626,7 @@
 // authentication information, while the cert parameter provides the
 // identity data itself. For instance, the "digest" scheme requires
 // a pair like "username:password" to be provided as the certificate.
-func (zk *ZooKeeper) AddAuth(scheme, cert string) Error {
+func (conn *Conn) AddAuth(scheme, cert string) os.Error {
 	cscheme := C.CString(scheme)
 	ccert := C.CString(cert)
 	defer C.free(unsafe.Pointer(cscheme))
@@ -725,43 +638,38 @@
 	}
 	defer C.destroy_completion_data(data)
 
-	rc, cerr := C.zoo_add_auth(zk.handle, cscheme, ccert, C.int(len(cert)),
-		C.handle_void_completion, unsafe.Pointer(data))
+	rc, cerr := C.zoo_add_auth(conn.handle, cscheme, ccert, C.int(len(cert)), C.handle_void_completion, unsafe.Pointer(data))
 	if rc != C.ZOK {
-		return newError(rc, cerr)
+		return zkError(rc, cerr)
 	}
 
 	C.wait_for_completion(data)
 
 	rc = C.int(uintptr(data.data))
-	if rc != C.ZOK {
-		return newError(rc, nil)
-	}
-
-	return nil
+	return zkError(rc, nil)
 }
 
 // ACL returns the access control list for path.
-func (zk *ZooKeeper) ACL(path string) ([]ACL, Stat, Error) {
+func (conn *Conn) ACL(path string) ([]ACL, *Stat, os.Error) {
 
 	cpath := C.CString(path)
 	defer C.free(unsafe.Pointer(cpath))
 
 	caclv := C.struct_ACL_vector{}
-	cstat := C.struct_Stat{}
 
-	rc, cerr := C.zoo_get_acl(zk.handle, cpath, &caclv, &cstat)
+	var cstat Stat
+	rc, cerr := C.zoo_get_acl(conn.handle, cpath, &caclv, &cstat.c)
 	if rc != C.ZOK {
-		return nil, nil, newError(rc, cerr)
+		return nil, nil, zkError(rc, cerr)
 	}
 
 	aclv := parseACLVector(&caclv)
 
-	return aclv, (*resultStat)(&cstat), nil
+	return aclv, &cstat, nil
 }
 
 // SetACL changes the access control list for path.
-func (zk *ZooKeeper) SetACL(path string, aclv []ACL, version int32) Error {
+func (conn *Conn) SetACL(path string, aclv []ACL, version int32) os.Error {
 
 	cpath := C.CString(path)
 	defer C.free(unsafe.Pointer(cpath))
@@ -769,12 +677,8 @@
 	caclv := buildACLVector(aclv)
 	defer C.deallocate_ACL_vector(caclv)
 
-	rc, cerr := C.zoo_set_acl(zk.handle, cpath, C.int(version), caclv)
-	if rc != C.ZOK {
-		return newError(rc, cerr)
-	}
-
-	return nil
+	rc, cerr := C.zoo_set_acl(conn.handle, cpath, C.int(version), caclv)
+	return zkError(rc, cerr)
 }
 
 func parseACLVector(caclv *C.struct_ACL_vector) []ACL {
@@ -822,7 +726,7 @@
 // -----------------------------------------------------------------------
 // RetryChange utility method.
 
-type ChangeFunc func(oldValue string, oldStat Stat) (newValue string, err os.Error)
+type ChangeFunc func(oldValue string, oldStat *Stat) (newValue string, err os.Error)
 
 // RetryChange runs changeFunc to attempt to atomically change path
 // in a lock free manner, and retries in case there was another
@@ -831,8 +735,7 @@
 // changeFunc must work correctly if called multiple times in case
 // the modification fails due to concurrent changes, and it may return
 // an error that will cause the the RetryChange function to stop and
-// return an Error with code ZSYSTEMERROR and the same .String() result
-// as the provided error.
+// return the same error.
 //
 // This mechanism is not suitable for a node that is frequently modified
 // concurrently. For those cases, consider using a pessimistic locking
@@ -845,8 +748,7 @@
 //
 // 2. Call the changeFunc with the current node value and stat,
 // or with an empty string and nil stat, if the node doesn't yet exist.
-// If the changeFunc returns an error, stop and return an Error with
-// ZSYSTEMERROR Code() and the same String() as the changeFunc error.
+// If the changeFunc returns an error, stop and return the same error.
 //
 // 3. If the changeFunc returns no errors, use the string returned as
 // the new candidate value for the node, and attempt to either create
@@ -855,36 +757,32 @@
 // in the same node), repeat from step 1.  If this procedure fails with any
 // other error, stop and return the error found.
 //
-func (zk *ZooKeeper) RetryChange(path string, flags int, acl []ACL, changeFunc ChangeFunc) (err Error) {
+func (conn *Conn) RetryChange(path string, flags int, acl []ACL, changeFunc ChangeFunc) os.Error {
 	for {
-		oldValue, oldStat, getErr := zk.Get(path)
-		if getErr != nil && getErr.Code() != ZNONODE {
-			err = getErr
-			break
-		}
-		newValue, osErr := changeFunc(oldValue, oldStat)
-		if osErr != nil {
-			return newError(ZSYSTEMERROR, osErr)
-		} else if oldStat == nil {
-			_, err = zk.Create(path, newValue, flags, acl)
-			if err == nil || err.Code() != ZNODEEXISTS {
-				break
+		oldValue, oldStat, err := conn.Get(path)
+		if err != nil && err != ZNONODE {
+			return err
+		}
+		newValue, err := changeFunc(oldValue, oldStat)
+		if err != nil {
+			return err
+		}
+		if oldStat == nil {
+			_, err := conn.Create(path, newValue, flags, acl)
+			if err == nil || err != ZNODEEXISTS {
+				return err
 			}
-		} else if newValue == oldValue {
+			continue
+		}
+		if newValue == oldValue {
 			return nil // Nothing to do.
-		} else {
-			_, err = zk.Set(path, newValue, oldStat.Version())
-			if err == nil {
-				break
-			} else {
-				code := err.Code()
-				if code != ZBADVERSION && code != ZNONODE {
-					break
-				}
-			}
+		}
+		_, err = conn.Set(path, newValue, oldStat.Version())
+		if err == nil || (err != ZBADVERSION && err != ZNONODE) {
+			return err
 		}
 	}
-	return err
+	panic("not reached")
 }
 
 // -----------------------------------------------------------------------
@@ -897,7 +795,7 @@
 // Whenever a *W method is called, it will return a channel which
 // outputs Event values.  Internally, a map is used to maintain references
 // between an unique integer key (the watchId), and the event channel. The
-// watchId is then handed to the C zookeeper library as the watch context,
+// watchId is then handed to the C ZooKeeper library as the watch context,
 // so that we get it back when events happen.  Using an integer key as the
 // watch context rather than a pointer is needed because there's no guarantee
 // that in the future the GC will not move objects around, and also because
@@ -910,13 +808,13 @@
 // Since Cgo doesn't allow calling back into Go, we actually fire a new
 // goroutine the very first time Init is called, and allow it to block
 // in a pthread condition variable within a C function. This condition
-// will only be notified once a zookeeper watch callback appends new
+// will only be notified once a ZooKeeper watch callback appends new
 // entries to the event list.  When this happens, the C function returns
 // and we get back into Go land with the pointer to the watch data,
 // including the watchId and other event details such as type and path.
 
 var watchMutex sync.Mutex
-var watchZooKeepers = make(map[uintptr]*ZooKeeper)
+var watchConns = make(map[uintptr]*Conn)
 var watchCounter uintptr
 var watchLoopCounter int
 
@@ -925,25 +823,20 @@
 // mostly as a debugging and testing aid.
 func CountPendingWatches() int {
 	watchMutex.Lock()
-	count := len(watchZooKeepers)
+	count := len(watchConns)
 	watchMutex.Unlock()
 	return count
 }
 
-// createWatch creates and registers a watch, returning the watch id
-// and channel.
-func (zk *ZooKeeper) createWatch(session bool) (watchId uintptr, watchChannel chan Event) {
-	buf := 1 // session/watch event
-	if session {
-		buf = 32
-	}
-	watchChannel = make(chan Event, buf)
+// createWatch creates a watch that will send events down the provided
+// channel, which must be one of the four possible event channel types.
+func (conn *Conn) createWatch(watchChannel statusChan) (watchId uintptr) {
 	watchMutex.Lock()
 	defer watchMutex.Unlock()
 	watchId = watchCounter
 	watchCounter += 1
-	zk.watchChannels[watchId] = watchChannel
-	watchZooKeepers[watchId] = zk
+	conn.watchChannels[watchId] = watchChannel
+	watchConns[watchId] = conn
 	return
 }
 
@@ -951,67 +844,71 @@
 // from ever getting delivered. It shouldn't be used if there's any
 // chance the watch channel is still visible and not closed, since
 // it might mean a goroutine would be blocked forever.
-func (zk *ZooKeeper) forgetWatch(watchId uintptr) {
+func (conn *Conn) forgetWatch(watchId uintptr) {
 	watchMutex.Lock()
 	defer watchMutex.Unlock()
-	zk.watchChannels[watchId] = nil, false
-	watchZooKeepers[watchId] = nil, false
+	conn.watchChannels[watchId] = nil, false
+	watchConns[watchId] = nil, false
 }
 
-// closeAllWatches closes all watch channels for zk.
-func (zk *ZooKeeper) closeAllWatches() {
+// closeAllWatches closes all watch channels for conn.
+func (conn *Conn) closeAllWatches() {
 	watchMutex.Lock()
 	defer watchMutex.Unlock()
-	for watchId, ch := range zk.watchChannels {
-		close(ch)
-		zk.watchChannels[watchId] = nil, false
-		watchZooKeepers[watchId] = nil, false
+	for watchId, ch := range conn.watchChannels {
+		ch.close()
+		conn.watchChannels[watchId] = nil, false
+		watchConns[watchId] = nil, false
 	}
 }
 
 // sendEvent delivers the event to the watchId event channel.  If the
 // event channel is a watch event channel, the event is delivered,
 // the channel is closed, and resources are freed.
-func sendEvent(watchId uintptr, event Event) {
-	if event.State == STATE_CLOSED {
-		panic("Attempted to send a CLOSED event")
+func sendEvent(watchId uintptr, eventType, connectionState int, path string) {
+	if eventType == event_SESSION && connectionState == 0 {
+		panic("Attempted to send a CLOSED connection status")
 	}
 	watchMutex.Lock()
 	defer watchMutex.Unlock()
-	zk, ok := watchZooKeepers[watchId]
+	conn, ok := watchConns[watchId]
 	if !ok {
 		return
 	}
-	if event.Type == EVENT_SESSION && watchId != zk.sessionWatchId {
-		switch event.State {
-		case STATE_EXPIRED_SESSION, STATE_AUTH_FAILED:
+	ch := conn.watchChannels[watchId]
+	if ch == nil {
+		return
+	}
+	if eventType == event_SESSION && watchId != conn.sessionWatchId {
+		switch SessionStatus(connectionState) {
+		case SessionExpired, SessionAuthFailed:
+			// The channel will be closed below.
 		default:
+			// Ignore non-critical session events.
 			// WTF? Feels like TCP saying "dropped a dup packet, ok?"
 			return
 		}
-	}
-	ch := zk.watchChannels[watchId]
-	if ch == nil {
-		return
-	}
-	select {
-	case ch <- event:
-	default:
-		// Channel not available for sending, which means session
-		// events are necessarily involved (trivial events go
-		// straight to the buffer), and the application isn't paying
-		// attention for long enough to have the buffer filled up.
-		// Break down now rather than leaking forever.
-		if watchId == zk.sessionWatchId {
-			panic("Session event channel buffer is full")
-		} else {
-			panic("Watch event channel buffer is full")
-		}
-	}
-	if watchId != zk.sessionWatchId {
-		zk.watchChannels[watchId] = nil, false
-		watchZooKeepers[watchId] = nil, false
-		close(ch)
+	} else {
+		if eventType != event_SESSION && watchId == conn.sessionWatchId {
+			panic(fmt.Errorf("unexpected non-session event %v %v", eventType, connectionState))
+		}
+		if !ch.send(eventType, connectionState, path) {
+			// Channel not available for sending, which means session
+			// events are necessarily involved (trivial events go
+			// straight to the buffer), and the application isn't paying
+			// attention for long enough to have the buffer filled up.
+			// Break down now rather than leaking forever.
+			if watchId == conn.sessionWatchId {
+				panic("Session event channel buffer is full")
+			} else {
+				panic("Watch event channel buffer is full (impossible!)")
+			}
+		}
+	}
+	if watchId != conn.sessionWatchId {
+		conn.watchChannels[watchId] = nil, false
+		watchConns[watchId] = nil, false
+		ch.close()
 	}
 }
 
@@ -1049,13 +946,8 @@
 	for {
 		// This will block until there's a watch event is available.
 		data := C.wait_for_watch()
-		event := Event{
-			Type:  int(data.event_type),
-			Path:  C.GoString(data.event_path),
-			State: int(data.connection_state),
-		}
 		watchId := uintptr(data.watch_context)
+		sendEvent(watchId, int(data.event_type), int(data.connection_state), C.GoString(data.event_path))
 		C.destroy_watch_data(data)
-		sendEvent(watchId, event)
 	}
 }
 
=== renamed file 'gozk_test.go' => 'zk_test.go'
--- gozk_test.go	2011-08-19 01:51:37 +0000
+++ zk_test.go	2011-09-30 13:35:26 +0000
@@ -1,17 +1,17 @@
-package gozk_test
+package zk_test
 
 import (
 	. "launchpad.net/gocheck"
-	"gozk"
+	"launchpad.net/gozk/zk"
 	"time"
 )
 
 // This error will be delivered via C errno, since ZK unfortunately
 // only provides the handler back from zookeeper_init().
 func (s *S) TestInitErrorThroughErrno(c *C) {
-	zk, watch, err := gozk.Init("bad-domain-without-port", 5e9)
-	if zk != nil {
-		zk.Close()
+	conn, watch, err := zk.Dial("bad-domain-without-port", 5e9)
+	if conn != nil {
+		conn.Close()
 	}
 	if watch != nil {
 		go func() {
@@ -23,15 +23,15 @@
 			}
 		}()
 	}
-	c.Assert(zk, IsNil)
+	c.Assert(conn, IsNil)
 	c.Assert(watch, IsNil)
 	c.Assert(err, Matches, "invalid argument")
 }
 
 func (s *S) TestRecvTimeoutInitParameter(c *C) {
-	zk, watch, err := gozk.Init(s.zkAddr, 0)
+	conn, watch, err := zk.Dial(s.zkAddr, 0)
 	c.Assert(err, IsNil)
-	defer zk.Close()
+	defer conn.Close()
 
 	select {
 	case <-watch:
@@ -40,7 +40,7 @@
 	}
 
 	for i := 0; i != 1000; i++ {
-		_, _, err := zk.Get("/zookeeper")
+		_, _, err := conn.Get("/zookeeper")
 		if err != nil {
 			c.Assert(err, Matches, "operation timeout")
 			c.SucceedNow()
@@ -51,88 +51,75 @@
 }
 
 func (s *S) TestSessionWatches(c *C) {
-	c.Assert(gozk.CountPendingWatches(), Equals, 0)
+	c.Assert(zk.CountPendingWatches(), Equals, 0)
 
 	zk1, watch1 := s.init(c)
 	zk2, watch2 := s.init(c)
 	zk3, watch3 := s.init(c)
 
-	c.Assert(gozk.CountPendingWatches(), Equals, 3)
+	c.Assert(zk.CountPendingWatches(), Equals, 3)
 
 	event1 := <-watch1
-	c.Assert(event1.Type, Equals, gozk.EVENT_SESSION)
-	c.Assert(event1.State, Equals, gozk.STATE_CONNECTED)
+	c.Assert(event1, Equals, zk.SessionConnected)
 
-	c.Assert(gozk.CountPendingWatches(), Equals, 3)
+	c.Assert(zk.CountPendingWatches(), Equals, 3)
 
 	event2 := <-watch2
-	c.Assert(event2.Type, Equals, gozk.EVENT_SESSION)
-	c.Assert(event2.State, Equals, gozk.STATE_CONNECTED)
+	c.Assert(event2, Equals, zk.SessionConnected)
 
-	c.Assert(gozk.CountPendingWatches(), Equals, 3)
+	c.Assert(zk.CountPendingWatches(), Equals, 3)
 
 	event3 := <-watch3
-	c.Assert(event3.Type, Equals, gozk.EVENT_SESSION)
-	c.Assert(event3.State, Equals, gozk.STATE_CONNECTED)
+	c.Assert(event3, Equals, zk.SessionConnected)
 
-	c.Assert(gozk.CountPendingWatches(), Equals, 3)
+	c.Assert(zk.CountPendingWatches(), Equals, 3)
 
 	zk1.Close()
-	c.Assert(gozk.CountPendingWatches(), Equals, 2)
+	c.Assert(zk.CountPendingWatches(), Equals, 2)
 	zk2.Close()
-	c.Assert(gozk.CountPendingWatches(), Equals, 1)
+	c.Assert(zk.CountPendingWatches(), Equals, 1)
 	zk3.Close()
-	c.Assert(gozk.CountPendingWatches(), Equals, 0)
+	c.Assert(zk.CountPendingWatches(), Equals, 0)
 }
 
-// Gozk injects a STATE_CLOSED event when zk.Close() is called, right
+// Gozk injects a STATE_CLOSED event when conn.Close() is called, right
 // before the channel is closed.  Closing the channel injects a nil
 // pointer, as usual for Go, so the STATE_CLOSED gives a chance to
 // know that a nil pointer is coming, and to stop the procedure.
 // Hopefully this procedure will avoid some nil-pointer references by
 // mistake.
 func (s *S) TestClosingStateInSessionWatch(c *C) {
-	zk, watch := s.init(c)
+	conn, watch := s.init(c)
 
 	event := <-watch
-	c.Assert(event.Type, Equals, gozk.EVENT_SESSION)
-	c.Assert(event.State, Equals, gozk.STATE_CONNECTED)
+	c.Assert(event, Equals, zk.SessionConnected)
 
-	zk.Close()
+	conn.Close()
 	event, ok := <-watch
 	c.Assert(ok, Equals, false)
-	c.Assert(event.Type, Equals, gozk.EVENT_CLOSED)
-	c.Assert(event.State, Equals, gozk.STATE_CLOSED)
-}
-
-func (s *S) TestEventString(c *C) {
-	var event gozk.Event
-	event = gozk.Event{gozk.EVENT_SESSION, "/path", gozk.STATE_CONNECTED}
-	c.Assert(event, Matches, "ZooKeeper connected")
-	event = gozk.Event{gozk.EVENT_CREATED, "/path", gozk.STATE_CONNECTED}
-	c.Assert(event, Matches, "ZooKeeper connected; path created: /path")
-	event = gozk.Event{-1, "/path", gozk.STATE_CLOSED}
-	c.Assert(event, Matches, "ZooKeeper connection closed")
-}
-
-var okTests = []struct{gozk.Event; Ok bool}{
-	{gozk.Event{gozk.EVENT_SESSION, "", gozk.STATE_CONNECTED}, true},
-	{gozk.Event{gozk.EVENT_CREATED, "", gozk.STATE_CONNECTED}, true},
-	{gozk.Event{0, "", gozk.STATE_CLOSED}, false},
-	{gozk.Event{0, "", gozk.STATE_EXPIRED_SESSION}, false},
-	{gozk.Event{0, "", gozk.STATE_AUTH_FAILED}, false},
+	c.Assert(event, Equals, zk.SessionStatus(0))
+}
+
+var okTests = []struct {
+	zk.SessionStatus
+	Ok bool
+}{
+	{zk.SessionConnected, true},
+	{0, false},
+	{zk.SessionExpired, false},
+	{zk.SessionAuthFailed, false},
 }
 
 func (s *S) TestEventOk(c *C) {
 	for _, t := range okTests {
-		c.Assert(t.Event.Ok(), Equals, t.Ok)
+		c.Assert(t.SessionStatus.Ok(), Equals, t.Ok)
 	}
 }
 
 func (s *S) TestGetAndStat(c *C) {
-	zk, _ := s.init(c)
+	conn, _ := s.init(c)
 
-	data, stat, err := zk.Get("/zookeeper")
+	data, stat, err := conn.Get("/zookeeper")
 	c.Assert(err, IsNil)
 	c.Assert(data, Equals, "")
 	c.Assert(stat.Czxid(), Equals, int64(0))
@@ -149,58 +136,58 @@
 }
 
 func (s *S) TestGetAndError(c *C) {
-	zk, _ := s.init(c)
+	conn, _ := s.init(c)
 
-	data, stat, err := zk.Get("/non-existent")
+	data, stat, err := conn.Get("/non-existent")
 
 	c.Assert(data, Equals, "")
 	c.Assert(stat, IsNil)
 	c.Assert(err, Matches, "no node")
-	c.Assert(err.Code(), Equals, gozk.ZNONODE)
+	c.Assert(err, Equals, zk.ZNONODE)
 }
 
 func (s *S) TestCreateAndGet(c *C) {
-	zk, _ := s.init(c)
+	conn, _ := s.init(c)
 
-	path, err := zk.Create("/test-", "bababum", gozk.SEQUENCE|gozk.EPHEMERAL, gozk.WorldACL(gozk.PERM_ALL))
+	path, err := conn.Create("/test-", "bababum", zk.SEQUENCE|zk.EPHEMERAL, zk.WorldACL(zk.PERM_ALL))
 	c.Assert(err, IsNil)
 	c.Assert(path, Matches, "/test-[0-9]+")
 
 	// Check the error condition from Create().
-	_, err = zk.Create(path, "", gozk.EPHEMERAL, gozk.WorldACL(gozk.PERM_ALL))
+	_, err = conn.Create(path, "", zk.EPHEMERAL, zk.WorldACL(zk.PERM_ALL))
 	c.Assert(err, Matches, "node exists")
 
-	data, _, err := zk.Get(path)
+	data, _, err := conn.Get(path)
 	c.Assert(err, IsNil)
 	c.Assert(data, Equals, "bababum")
 }
 
 func (s *S) TestCreateSetAndGet(c *C) {
-	zk, _ := s.init(c)
+	conn, _ := s.init(c)
 
-	_, err := zk.Create("/test", "", gozk.EPHEMERAL, gozk.WorldACL(gozk.PERM_ALL))
+	_, err := conn.Create("/test", "", zk.EPHEMERAL, zk.WorldACL(zk.PERM_ALL))
 	c.Assert(err, IsNil)
 
-	stat, err := zk.Set("/test", "bababum", -1) // Any version.
+	stat, err := conn.Set("/test", "bababum", -1) // Any version.
 	c.Assert(err, IsNil)
 	c.Assert(stat.Version(), Equals, int32(1))
 
-	data, _, err := zk.Get("/test")
+	data, _, err := conn.Get("/test")
 	c.Assert(err, IsNil)
 	c.Assert(data, Equals, "bababum")
 }
 
 func (s *S) TestGetAndWatch(c *C) {
-	c.Check(gozk.CountPendingWatches(), Equals, 0)
-
-	zk, _ := s.init(c)
-
-	c.Check(gozk.CountPendingWatches(), Equals, 1)
-
-	_, err := zk.Create("/test", "one", gozk.EPHEMERAL, gozk.WorldACL(gozk.PERM_ALL))
+	c.Check(zk.CountPendingWatches(), Equals, 0)
+
+	conn, _ := s.init(c)
+
+	c.Check(zk.CountPendingWatches(), Equals, 1)
+
+	_, err := conn.Create("/test", "one", zk.EPHEMERAL, zk.WorldACL(zk.PERM_ALL))
 	c.Assert(err, IsNil)
 
-	data, stat, watch, err := zk.GetW("/test")
+	data, stat, watch, err := conn.GetW("/test")
 	c.Assert(err, IsNil)
 	c.Assert(data, Equals, "one")
 	c.Assert(stat.Version(), Equals, int32(0))
@@ -211,17 +198,17 @@
 	default:
 	}
 
-	c.Check(gozk.CountPendingWatches(), Equals, 2)
+	c.Check(zk.CountPendingWatches(), Equals, 2)
 
-	_, err = zk.Set("/test", "two", -1)
+	_, err = conn.Set("/test", "two", -1)
 	c.Assert(err, IsNil)
 
 	event := <-watch
-	c.Assert(event.Type, Equals, gozk.EVENT_CHANGED)
-
-	c.Check(gozk.CountPendingWatches(), Equals, 1)
-
-	data, _, watch, err = zk.GetW("/test")
+	c.Assert(event, Equals, zk.NodeChanged)
+
+	c.Check(zk.CountPendingWatches(), Equals, 1)
+
+	data, _, watch, err = conn.GetW("/test")
 	c.Assert(err, IsNil)
 	c.Assert(data, Equals, "two")
 
@@ -231,86 +218,86 @@
 	default:
 	}
 
-	c.Check(gozk.CountPendingWatches(), Equals, 2)
+	c.Check(zk.CountPendingWatches(), Equals, 2)
 
-	_, err = zk.Set("/test", "three", -1)
+	_, err = conn.Set("/test", "three", -1)
 	c.Assert(err, IsNil)
 
 	event = <-watch
-	c.Assert(event.Type, Equals, gozk.EVENT_CHANGED)
+	c.Assert(event, Equals, zk.NodeChanged)
 
-	c.Check(gozk.CountPendingWatches(), Equals, 1)
+	c.Check(zk.CountPendingWatches(), Equals, 1)
 }
 
 func (s *S) TestGetAndWatchWithError(c *C) {
-	c.Check(gozk.CountPendingWatches(), Equals, 0)
-
-	zk, _ := s.init(c)
-
-	c.Check(gozk.CountPendingWatches(), Equals, 1)
-
-	_, _, watch, err := zk.GetW("/test")
+	c.Check(zk.CountPendingWatches(), Equals, 0)
+
+	conn, _ := s.init(c)
+
+	c.Check(zk.CountPendingWatches(), Equals, 1)
+
+	_, _, watch, err := conn.GetW("/test")
 	c.Assert(err, NotNil)
-	c.Assert(err.Code(), Equals, gozk.ZNONODE)
+	c.Assert(err, Equals, zk.ZNONODE)
 	c.Assert(watch, IsNil)
 
-	c.Check(gozk.CountPendingWatches(), Equals, 1)
+	c.Check(zk.CountPendingWatches(), Equals, 1)
 }
 
 func (s *S) TestCloseReleasesWatches(c *C) {
-	c.Check(gozk.CountPendingWatches(), Equals, 0)
-
-	zk, _ := s.init(c)
-
-	c.Check(gozk.CountPendingWatches(), Equals, 1)
-
-	_, err := zk.Create("/test", "one", gozk.EPHEMERAL, gozk.WorldACL(gozk.PERM_ALL))
-	c.Assert(err, IsNil)
-
-	_, _, _, err = zk.GetW("/test")
-	c.Assert(err, IsNil)
-
-	c.Assert(gozk.CountPendingWatches(), Equals, 2)
-
-	zk.Close()
-
-	c.Assert(gozk.CountPendingWatches(), Equals, 0)
+	c.Check(zk.CountPendingWatches(), Equals, 0)
+
+	conn, _ := s.init(c)
+
+	c.Check(zk.CountPendingWatches(), Equals, 1)
+
+	_, err := conn.Create("/test", "one", zk.EPHEMERAL, zk.WorldACL(zk.PERM_ALL))
+	c.Assert(err, IsNil)
+
+	_, _, _, err = conn.GetW("/test")
+	c.Assert(err, IsNil)
+
+	c.Assert(zk.CountPendingWatches(), Equals, 2)
+
+	conn.Close()
+
+	c.Assert(zk.CountPendingWatches(), Equals, 0)
 }
 
 // By default, the ZooKeeper C client will hang indefinitely if a
 // handler is closed twice.  We get in the way and prevent it.
 func (s *S) TestClosingTwiceDoesntHang(c *C) {
-	zk, _ := s.init(c)
-	err := zk.Close()
+	conn, _ := s.init(c)
+	err := conn.Close()
 	c.Assert(err, IsNil)
-	err = zk.Close()
+	err = conn.Close()
 	c.Assert(err, NotNil)
-	c.Assert(err.Code(), Equals, gozk.ZCLOSING)
+	c.Assert(err, Equals, zk.ZCLOSING)
 }
 
 func (s *S) TestChildren(c *C) {
-	zk, _ := s.init(c)
+	conn, _ := s.init(c)
 
-	children, stat, err := zk.Children("/")
+	children, stat, err := conn.Children("/")
 	c.Assert(err, IsNil)
 	c.Assert(children, Equals, []string{"zookeeper"})
 	c.Assert(stat.NumChildren(), Equals, int32(1))
 
-	children, stat, err = zk.Children("/non-existent")
+	children, stat, err = conn.Children("/non-existent")
 	c.Assert(err, NotNil)
-	c.Assert(err.Code(), Equals, gozk.ZNONODE)
+	c.Assert(err, Equals, zk.ZNONODE)
 	c.Assert(children, Equals, []string{})
-	c.Assert(stat, Equals, nil)
+	c.Assert(stat, IsNil)
 }
 
 func (s *S) TestChildrenAndWatch(c *C) {
-	c.Check(gozk.CountPendingWatches(), Equals, 0)
-
-	zk, _ := s.init(c)
-
-	c.Check(gozk.CountPendingWatches(), Equals, 1)
-
-	children, stat, watch, err := zk.ChildrenW("/")
+	c.Check(zk.CountPendingWatches(), Equals, 0)
+
+	conn, _ := s.init(c)
+
+	c.Check(zk.CountPendingWatches(), Equals, 1)
+
+	children, stat, watch, err := conn.ChildrenW("/")
 	c.Assert(err, IsNil)
 	c.Assert(children, Equals, []string{"zookeeper"})
 	c.Assert(stat.NumChildren(), Equals, int32(1))
@@ -321,18 +308,17 @@
 	default:
 	}
 
-	c.Check(gozk.CountPendingWatches(), Equals, 2)
+	c.Check(zk.CountPendingWatches(), Equals, 2)
 
-	_, err = zk.Create("/test1", "", gozk.EPHEMERAL, gozk.WorldACL(gozk.PERM_ALL))
+	_, err = conn.Create("/test1", "", zk.EPHEMERAL, zk.WorldACL(zk.PERM_ALL))
 	c.Assert(err, IsNil)
 
 	event := <-watch
-	c.Assert(event.Type, Equals, gozk.EVENT_CHILD)
-	c.Assert(event.Path, Equals, "/")
-
-	c.Check(gozk.CountPendingWatches(), Equals, 1)
-
-	children, stat, watch, err = zk.ChildrenW("/")
+	c.Assert(event, Equals, zk.NodeChangedChild)
+
+	c.Check(zk.CountPendingWatches(), Equals, 1)
+
+	children, stat, watch, err = conn.ChildrenW("/")
 	c.Assert(err, IsNil)
 	c.Assert(stat.NumChildren(), Equals, int32(2))
 
@@ -345,57 +331,56 @@
 	default:
 	}
 
-	c.Check(gozk.CountPendingWatches(), Equals, 2)
+	c.Check(zk.CountPendingWatches(), Equals, 2)
 
-	_, err = zk.Create("/test2", "", gozk.EPHEMERAL, gozk.WorldACL(gozk.PERM_ALL))
+	_, err = conn.Create("/test2", "", zk.EPHEMERAL, zk.WorldACL(zk.PERM_ALL))
 	c.Assert(err, IsNil)
 
 	event = <-watch
-	c.Assert(event.Type, Equals, gozk.EVENT_CHILD)
+	c.Assert(event, Equals, zk.NodeChangedChild)
 
-	c.Check(gozk.CountPendingWatches(), Equals, 1)
+	c.Check(zk.CountPendingWatches(), Equals, 1)
 }
 
 func (s *S) TestChildrenAndWatchWithError(c *C) {
-	c.Check(gozk.CountPendingWatches(), Equals, 0)
-
-	zk, _ := s.init(c)
-
-	c.Check(gozk.CountPendingWatches(), Equals, 1)
-
-	_, stat, watch, err := zk.ChildrenW("/test")
+	c.Check(zk.CountPendingWatches(), Equals, 0)
+
+	conn, _ := s.init(c)
+
+	c.Check(zk.CountPendingWatches(), Equals, 1)
+
+	_, stat, watch, err := conn.ChildrenW("/test")
 	c.Assert(err, NotNil)
-	c.Assert(err.Code(), Equals, gozk.ZNONODE)
+	c.Assert(err, Equals, zk.ZNONODE)
 	c.Assert(watch, IsNil)
 	c.Assert(stat, IsNil)
 
-	c.Check(gozk.CountPendingWatches(), Equals, 1)
+	c.Check(zk.CountPendingWatches(), Equals, 1)
 }
 
 func (s *S) TestExists(c *C) {
-	zk, _ := s.init(c)
-
-	stat, err := zk.Exists("/zookeeper")
-	c.Assert(err, IsNil)
-	c.Assert(stat.NumChildren(), Equals, int32(1))
-
-	stat, err = zk.Exists("/non-existent")
+	conn, _ := s.init(c)
+
+	stat, err := conn.Exists("/non-existent")
 	c.Assert(err, IsNil)
 	c.Assert(stat, IsNil)
+
+	stat, err = conn.Exists("/zookeeper")
+	c.Assert(err, IsNil)
 }
 
 func (s *S) TestExistsAndWatch(c *C) {
-	c.Check(gozk.CountPendingWatches(), Equals, 0)
-
-	zk, _ := s.init(c)
-
-	c.Check(gozk.CountPendingWatches(), Equals, 1)
-
-	stat, watch, err := zk.ExistsW("/test")
+	c.Check(zk.CountPendingWatches(), Equals, 0)
+
+	conn, _ := s.init(c)
+
+	c.Check(zk.CountPendingWatches(), Equals, 1)
+
+	stat, watch, err := conn.ExistsW("/test")
 	c.Assert(err, IsNil)
 	c.Assert(stat, IsNil)
 
-	c.Check(gozk.CountPendingWatches(), Equals, 2)
+	c.Check(zk.CountPendingWatches(), Equals, 2)
 
 	select {
 	case <-watch:
@@ -403,62 +388,61 @@
 	default:
 	}
 
-	_, err = zk.Create("/test", "", gozk.EPHEMERAL, gozk.WorldACL(gozk.PERM_ALL))
+	_, err = conn.Create("/test", "", zk.EPHEMERAL, zk.WorldACL(zk.PERM_ALL))
 	c.Assert(err, IsNil)
 
 	event := <-watch
-	c.Assert(event.Type, Equals, gozk.EVENT_CREATED)
-	c.Assert(event.Path, Equals, "/test")
-
-	c.Check(gozk.CountPendingWatches(), Equals, 1)
-
-	stat, watch, err = zk.ExistsW("/test")
+	c.Assert(event, Equals, zk.NodeCreated)
+
+	c.Check(zk.CountPendingWatches(), Equals, 1)
+
+	stat, watch, err = conn.ExistsW("/test")
 	c.Assert(err, IsNil)
 	c.Assert(stat, NotNil)
 	c.Assert(stat.NumChildren(), Equals, int32(0))
 
-	c.Check(gozk.CountPendingWatches(), Equals, 2)
+	c.Check(zk.CountPendingWatches(), Equals, 2)
 }
 
 func (s *S) TestExistsAndWatchWithError(c *C) {
-	c.Check(gozk.CountPendingWatches(), Equals, 0)
-
-	zk, _ := s.init(c)
-
-	c.Check(gozk.CountPendingWatches(), Equals, 1)
-
-	stat, watch, err := zk.ExistsW("///")
+	c.Check(zk.CountPendingWatches(), Equals, 0)
+
+	conn, _ := s.init(c)
+
+	c.Check(zk.CountPendingWatches(), Equals, 1)
+
+	stat, watch, err := conn.ExistsW("///")
 	c.Assert(err, NotNil)
-	c.Assert(err.Code(), Equals, gozk.ZBADARGUMENTS)
+	c.Assert(err, Equals, zk.ZBADARGUMENTS)
 	c.Assert(stat, IsNil)
 	c.Assert(watch, IsNil)
 
-	c.Check(gozk.CountPendingWatches(), Equals, 1)
+	c.Check(zk.CountPendingWatches(), Equals, 1)
 }
 
 func (s *S) TestDelete(c *C) {
-	zk, _ := s.init(c)
-
-	_, err := zk.Create("/test", "", gozk.EPHEMERAL, gozk.WorldACL(gozk.PERM_ALL))
-	c.Assert(err, IsNil)
-
-	err = zk.Delete("/test", 5)
-	c.Assert(err, NotNil)
-	c.Assert(err.Code(), Equals, gozk.ZBADVERSION)
-
-	err = zk.Delete("/test", -1)
-	c.Assert(err, IsNil)
-
-	err = zk.Delete("/test", -1)
-	c.Assert(err, NotNil)
-	c.Assert(err.Code(), Equals, gozk.ZNONODE)
+	conn, _ := s.init(c)
+
+	_, err := conn.Create("/test", "", zk.EPHEMERAL, zk.WorldACL(zk.PERM_ALL))
+	c.Assert(err, IsNil)
+
+	err = conn.Delete("/test", 5)
+	c.Assert(err, NotNil)
+	c.Assert(err, Equals, zk.ZBADVERSION)
+
+	err = conn.Delete("/test", -1)
+	c.Assert(err, IsNil)
+
+	err = conn.Delete("/test", -1)
+	c.Assert(err, NotNil)
+	c.Assert(err, Equals, zk.ZNONODE)
 }
 
 func (s *S) TestClientIdAndReInit(c *C) {
 	zk1, _ := s.init(c)
 	clientId1 := zk1.ClientId()
 
-	zk2, _, err := gozk.ReInit(s.zkAddr, 5e9, clientId1)
+	zk2, _, err := zk.Redial(s.zkAddr, 5e9, clientId1)
 	c.Assert(err, IsNil)
 	defer zk2.Close()
 	clientId2 := zk2.ClientId()
@@ -469,110 +453,113 @@
 // Surprisingly for some (including myself, initially), the watch
 // returned by the exists method actually fires on data changes too.
 func (s *S) TestExistsWatchOnDataChange(c *C) {
-	zk, _ := s.init(c)
-
-	_, err := zk.Create("/test", "", gozk.EPHEMERAL, gozk.WorldACL(gozk.PERM_ALL))
-	c.Assert(err, IsNil)
-
-	_, watch, err := zk.ExistsW("/test")
-	c.Assert(err, IsNil)
-
-	_, err = zk.Set("/test", "new", -1)
+	conn, _ := s.init(c)
+
+	_, err := conn.Create("/test", "", zk.EPHEMERAL, zk.WorldACL(zk.PERM_ALL))
+	c.Assert(err, IsNil)
+
+	_, watch, err := conn.ExistsW("/test")
+	c.Assert(err, IsNil)
+
+	_, err = conn.Set("/test", "new", -1)
 	c.Assert(err, IsNil)
 
 	event := <-watch
 
-	c.Assert(event.Path, Equals, "/test")
-	c.Assert(event.Type, Equals, gozk.EVENT_CHANGED)
+	c.Assert(event, Equals, zk.NodeChanged)
 }
 
 func (s *S) TestACL(c *C) {
-	zk, _ := s.init(c)
-
-	_, err := zk.Create("/test", "", gozk.EPHEMERAL, gozk.WorldACL(gozk.PERM_ALL))
-	c.Assert(err, IsNil)
-
-	acl, stat, err := zk.ACL("/test")
-	c.Assert(err, IsNil)
-	c.Assert(acl, Equals, gozk.WorldACL(gozk.PERM_ALL))
+	conn, _ := s.init(c)
+
+	_, err := conn.Create("/test", "", zk.EPHEMERAL, zk.WorldACL(zk.PERM_ALL))
+	c.Assert(err, IsNil)
+
+	acl, stat, err := conn.ACL("/test")
+	c.Assert(err, IsNil)
+	c.Assert(acl, Equals, zk.WorldACL(zk.PERM_ALL))
 	c.Assert(stat, NotNil)
 	c.Assert(stat.Version(), Equals, int32(0))
 
-	acl, stat, err = zk.ACL("/non-existent")
+	acl, stat, err = conn.ACL("/non-existent")
 	c.Assert(err, NotNil)
-	c.Assert(err.Code(), Equals, gozk.ZNONODE)
+	c.Assert(err, Equals, zk.ZNONODE)
 	c.Assert(acl, IsNil)
 	c.Assert(stat, IsNil)
+	c.Log("finished TestACL")
 }
 
 func (s *S) TestSetACL(c *C) {
-	zk, _ := s.init(c)
+	conn, _ := s.init(c)
 
-	_, err := zk.Create("/test", "", gozk.EPHEMERAL, gozk.WorldACL(gozk.PERM_ALL))
+	_, err := conn.Create("/test", "", zk.EPHEMERAL, zk.WorldACL(zk.PERM_ALL))
 	c.Assert(err, IsNil)
 
-	err = zk.SetACL("/test", gozk.WorldACL(gozk.PERM_ALL), 5)
+	err = conn.SetACL("/test", zk.WorldACL(zk.PERM_ALL), 5)
 	c.Assert(err, NotNil)
-	c.Assert(err.Code(), Equals, gozk.ZBADVERSION)
-
-	err = zk.SetACL("/test", gozk.WorldACL(gozk.PERM_READ), -1)
-	c.Assert(err, IsNil)
-
-	acl, _, err := zk.ACL("/test")
-	c.Assert(err, IsNil)
-	c.Assert(acl, Equals, gozk.WorldACL(gozk.PERM_READ))
+	c.Assert(err, Equals, zk.ZBADVERSION)
+
+	err = conn.SetACL("/test", zk.WorldACL(zk.PERM_READ), -1)
+	c.Assert(err, IsNil)
+
+	acl, _, err := conn.ACL("/test")
+	c.Assert(err, IsNil)
+	c.Assert(acl, Equals, zk.WorldACL(zk.PERM_READ))
 }
 
 func (s *S) TestAddAuth(c *C) {
-	zk, _ := s.init(c)
-
-	acl := []gozk.ACL{{gozk.PERM_READ, "digest", "joe:enQcM3mIEHQx7IrPNStYBc0qfs8="}}
-
-	_, err := zk.Create("/test", "", gozk.EPHEMERAL, acl)
+	conn, _ := s.init(c)
+
+	acl := []zk.ACL{{zk.PERM_READ, "digest", "joe:enQcM3mIEHQx7IrPNStYBc0qfs8="}}
+
+	_, err := conn.Create("/test", "", zk.EPHEMERAL, acl)
 	c.Assert(err, IsNil)
 
-	_, _, err = zk.Get("/test")
+	_, _, err = conn.Get("/test")
 	c.Assert(err, NotNil)
-	c.Assert(err.Code(), Equals, gozk.ZNOAUTH)
+	c.Assert(err, Equals, zk.ZNOAUTH)
 
-	err = zk.AddAuth("digest", "joe:passwd")
+	err = conn.AddAuth("digest", "joe:passwd")
 	c.Assert(err, IsNil)
 
-	_, _, err = zk.Get("/test")
+	_, _, err = conn.Get("/test")
 	c.Assert(err, IsNil)
 }
 
 func (s *S) TestWatchOnReconnection(c *C) {
-	c.Check(gozk.CountPendingWatches(), Equals, 0)
+	c.Check(zk.CountPendingWatches(), Equals, 0)
 
-	zk, session := s.init(c)
+	conn, session := s.init(c)
 
 	event := <-session
-	c.Assert(event.Type, Equals, gozk.EVENT_SESSION)
-	c.Assert(event.State, Equals, gozk.STATE_CONNECTED)
-
-	c.Check(gozk.CountPendingWatches(), Equals, 1)
-
-	stat, watch, err := zk.ExistsW("/test")
+	c.Assert(event, Equals, zk.SessionConnected)
+
+	c.Check(zk.CountPendingWatches(), Equals, 1)
+
+	stat, watch, err := conn.ExistsW("/test")
 	c.Assert(err, IsNil)
 	c.Assert(stat, IsNil)
 
-	c.Check(gozk.CountPendingWatches(), Equals, 2)
-
-	s.StopZK()
+	c.Check(zk.CountPendingWatches(), Equals, 2)
+
+	err = s.zkServer.Stop()
+	c.Assert(err, IsNil)
+
 	time.Sleep(2e9)
-	s.StartZK()
-
-	// The session channel should receive the reconnection notification,
+
+	err = s.zkServer.Start()
+	c.Assert(err, IsNil)
+
+	// The session channel should receive the reconnection notification.
 	select {
 	case event := <-session:
-		c.Assert(event.State, Equals, gozk.STATE_CONNECTING)
+		c.Assert(event, Equals, zk.SessionConnecting)
 	case <-time.After(3e9):
 		c.Fatal("Session watch didn't fire")
 	}
 	select {
 	case event := <-session:
-		c.Assert(event.State, Equals, gozk.STATE_CONNECTED)
+		c.Assert(event, Equals, zk.SessionConnected)
 	case <-time.After(3e9):
 		c.Fatal("Session watch didn't fire")
 	}
@@ -585,40 +572,38 @@
 	}
 
 	// And it should still work.
-	_, err = zk.Create("/test", "", gozk.EPHEMERAL, gozk.WorldACL(gozk.PERM_ALL))
+	_, err = conn.Create("/test", "", zk.EPHEMERAL, zk.WorldACL(zk.PERM_ALL))
 	c.Assert(err, IsNil)
 
-	event = <-watch
-	c.Assert(event.Type, Equals, gozk.EVENT_CREATED)
-	c.Assert(event.Path, Equals, "/test")
+	nodeStatus := <-watch
+	c.Assert(nodeStatus, Equals, zk.NodeCreated)
 
-	c.Check(gozk.CountPendingWatches(), Equals, 1)
+	c.Check(zk.CountPendingWatches(), Equals, 1)
 }
 
 func (s *S) TestWatchOnSessionExpiration(c *C) {
-	c.Check(gozk.CountPendingWatches(), Equals, 0)
+	c.Check(zk.CountPendingWatches(), Equals, 0)
 
-	zk, session := s.init(c)
+	conn, session := s.init(c)
 
 	event := <-session
-	c.Assert(event.Type, Equals, gozk.EVENT_SESSION)
-	c.Assert(event.State, Equals, gozk.STATE_CONNECTED)
-
-	c.Check(gozk.CountPendingWatches(), Equals, 1)
-
-	stat, watch, err := zk.ExistsW("/test")
+	c.Assert(event, Equals, zk.SessionConnected)
+
+	c.Check(zk.CountPendingWatches(), Equals, 1)
+
+	stat, watch, err := conn.ExistsW("/test")
 	c.Assert(err, IsNil)
 	c.Assert(stat, IsNil)
 
-	c.Check(gozk.CountPendingWatches(), Equals, 2)
+	c.Check(zk.CountPendingWatches(), Equals, 2)
 
 	// Use expiration trick described in the FAQ.
-	clientId := zk.ClientId()
-	zk2, session2, err := gozk.ReInit(s.zkAddr, 5e9, clientId)
+	clientId := conn.ClientId()
+	zk2, session2, err := zk.Redial(s.zkAddr, 5e9, clientId)
 
 	for event := range session2 {
 		c.Log("Event from overlapping session: ", event)
-		if event.State == gozk.STATE_CONNECTED {
+		if event == zk.SessionConnected {
 			// Wait for zk to process the connection.
 			// Not reliable without this. :-(
 			time.Sleep(1e9)
@@ -627,21 +612,21 @@
 	}
 	for event := range session {
 		c.Log("Event from primary session: ", event)
-		if event.State == gozk.STATE_EXPIRED_SESSION {
+		if event == zk.SessionExpired {
 			break
 		}
 	}
 
 	select {
-	case event := <-watch:
-		c.Assert(event.State, Equals, gozk.STATE_EXPIRED_SESSION)
+	case event, ok := <-watch:
+		c.Assert(event, Equals, zk.NodeStatus(0))
+		c.Assert(ok, Equals, false)
 	case <-time.After(3e9):
 		c.Fatal("Watch event didn't fire")
 	}
 
-	event = <-watch
-	c.Assert(event.Type, Equals, gozk.EVENT_CLOSED)
-	c.Assert(event.State, Equals, gozk.STATE_CLOSED)
+	nodeStatus := <-watch
+	c.Assert(nodeStatus, Equals, zk.NodeStatus(0))
 
-	c.Check(gozk.CountPendingWatches(), Equals, 1)
+	c.Check(zk.CountPendingWatches(), Equals, 1)
 }