Receive Updates from a particular Zookeeper path on a channel

zk_update_channel.go

package main

import (
	"fmt"
	"github.com/samuel/go-zookeeper/zk"
	"time"
)

func main() {
	c, ech, err := zk.Connect([]string{"localhost"}, time.Minute)
	if err != nil {
		fmt.Println(err.Error())
	}
	isReady := false
	for e := range ech {
		if e.State == zk.StateHasSession {
			isReady = true
			break
		}
	}
	if !isReady {
		panic("Could not connect to Zookeeper")
	}

	values, stopReceiving := getValueChannel(c, "/test")

	go func() {
		time.Sleep(10 * time.Second)
		fmt.Println("Stopping receiving zk updates")
		stopReceiving <- struct{}{}
	}()

	for v := range values {
		fmt.Println(string(v))
	}
}

func getValueChannel(c *zk.Conn, path string) (<-chan []byte, chan<- struct{}) {
	out := make(chan []byte)
	stopReceiving := make(chan struct{})

	go func() {
		value, _, ech, err := c.GetW(path)
		if err != nil {
			panic(fmt.Sprintf("no value set initially for %v", path))
		}
		out <- value
	ReceiveZookeeperUpdates:
		for {
			select {
			case e := <-ech:
				switch e.Type {
				case zk.EventNodeDataChanged:
					value, _, ech, err = c.GetW(path)
					out <- value

				case zk.EventNodeCreated:
					value, _, ech, err = c.GetW(path)
					out <- value
				}
			case <-stopReceiving:
				break ReceiveZookeeperUpdates
			}
		}
	}()

	return out, stopReceiving
}