func (zs *ZKStorage) createZPath(conn *zk.Conn, zpath string) error {
	_, err := conn.Create(zpath, nil, 0, zk.WorldACL(zk.PermAll))
	if err != nil {
		if zk.ErrNodeExists == err {
			return nil
		} else {
			parent, _ := path.Split(zpath)
			if len(parent) == 0 {
				return errors.New("Specified blank path")
			}
			err = zs.createZPath(conn, parent[:len(parent)-1])
			if err != nil {
				return err
			}

			_, err = conn.Create(zpath, nil, 0, zk.WorldACL(zk.PermAll))
			if err == zk.ErrNodeExists {
				err = nil
			}
		}
	}

	if zk.ErrNodeExists == err {
		return nil
	} else {
		return err
	}
}

// 创建一个临时节点，并监听自己
func CreateTempW(conn *zk.Conn, fpath, data string, watchFunc func(bool, zk.Event)) error {
	tpath, err := conn.Create(path.Join(fpath)+"/", []byte(data), zk.FlagEphemeral|zk.FlagSequence, zk.WorldACL(zk.PermAll))
	if err != nil {
		glog.Errorf("conn.Create(\"%s\", \"%s\", zk.FlagEphemeral|zk.FlagSequence) error(%v)", fpath, data, err)
		return err
	}
	glog.V(1).Infof("create a zookeeper node:%s", tpath)
	// watch self
	if watchFunc != nil {
		go func() {
			for {
				glog.Infof("zk path: \"%s\" set a watch", tpath)
				exist, _, watch, err := conn.ExistsW(tpath)
				if err != nil {
					glog.Errorf("zk.ExistsW(%s) error(%v)", tpath, err)
					return
				}
				event := <-watch
				watchFunc(exist, event)
				glog.Infof("zk path: \"%s\" receive a event %v", tpath, event)
			}
		}()
	}
	return nil
}

func (a *HostAgent) processChildrenAndWait(conn *zk.Conn) bool {
	processing := make(map[string]chan int)
	ssDone := make(chan stateResult, 25)

	hostPath := zzk.HostPath(a.hostId)

	for {

		children, _, zkEvent, err := conn.ChildrenW(hostPath)
		if err != nil {
			glog.V(0).Infoln("Unable to read children, retrying.")
			time.Sleep(3 * time.Second)
			return true
		}
		a.startMissingChildren(conn, children, processing, ssDone)

		select {

		case errc := <-a.closing:
			glog.V(1).Info("Agent received interrupt")
			err = waitForSsNodes(processing, ssDone)
			errc <- err
			return false

		case ssResult := <-ssDone:
			glog.V(1).Infof("Goroutine finished %s", ssResult.id)
			delete(processing, ssResult.id)

		case evt := <-zkEvent:
			glog.V(1).Info("Agent event: ", evt)
		}
	}
}

// NewExplorer creates a new Explorer instance with a name,
// application and balancer finders and zookeeper connection
// to persist versioning information
func NewExplorer(name string, af application.Finder, bf balancer.Finder, zc *zk.Conn, zp string) (*Explorer, error) {
	s := state.State{}

	ss, _, err := zc.Get(zp)
	if err != nil {
		if err != zk.ErrNoNode {
			return nil, err
		}

		s.Versions = map[string]state.Versions{}
	} else {
		err := json.Unmarshal(ss, &s)
		if err != nil {
			return nil, err
		}
	}

	return &Explorer{
		name:      name,
		af:        af,
		bf:        bf,
		zookeeper: zc,
		zp:        zp,
		state:     s,
		mutex:     sync.Mutex{},
	}, nil
}

func NewExplorer(d Discoverer, zc *zk.Conn, zp string, location ExplorerLocation) (*Explorer, error) {
	state := State{}

	ss, _, err := zc.Get(zp)
	if err != nil {
		if err != zk.ErrNoNode {
			return nil, err
		}

		state.Versions = map[string]Versions{}
	} else {
		err := json.Unmarshal(ss, &state)
		if err != nil {
			return nil, err
		}
	}

	return &Explorer{
		discoverer: d,
		zookeeper:  zc,
		zp:         zp,
		state:      state,
		location:   location,
		mutex:      sync.Mutex{},
	}, nil
}

func All(conn *zk.Conn, zkConf conf.Zookeeper) (map[string]Service, error) {

	err := ensurePathExists(conn, zkConf.Path)
	if err != nil {
		return nil, err
	}

	services := map[string]Service{}
	keys, _, err2 := conn.Children(zkConf.Path)

	if err2 != nil {
		return nil, err2
	}

	for _, childPath := range keys {
		bite, _, e := conn.Get(zkConf.Path + "/" + childPath)
		if e != nil {
			return nil, e
			break
		}
		appId, _ := unescapeSlashes(childPath)
		services[appId] = Service{Id: appId, Acl: string(bite)}
	}
	return services, nil
}

// RegisterTmp create a ephemeral node, and watch it, if node droped then send a SIGQUIT to self.
func RegisterTemp(conn *zk.Conn, fpath, data string) error {
	tpath, err := conn.Create(path.Join(fpath)+"/", []byte(data), zk.FlagEphemeral|zk.FlagSequence, zk.WorldACL(zk.PermAll))
	if err != nil {
		glog.Errorf("conn.Create(\"%s\", \"%s\", zk.FlagEphemeral|zk.FlagSequence) error(%v)", fpath, data, err)
		return err
	}
	glog.V(1).Infof("create a zookeeper node:%s", tpath)
	// watch self
	go func() {
		for {
			glog.Infof("zk path: \"%s\" set a watch", tpath)
			exist, _, watch, err := conn.ExistsW(tpath)
			if err != nil {
				glog.Errorf("zk.ExistsW(\"%s\") error(%v)", tpath, err)
				glog.Warningf("zk path: \"%s\" set watch failed, kill itself", tpath)
				killSelf()
				return
			}
			if !exist {
				glog.Warningf("zk path: \"%s\" not exist, kill itself", tpath)
				killSelf()
				return
			}
			event := <-watch
			glog.Infof("zk path: \"%s\" receive a event %v", tpath, event)
		}
	}()
	return nil
}

func registerNode(conn *zk.Conn, node, path string) (*NodeInfo, error) {
	path = path + "/" + node
	data, _, err := conn.Get(path)
	if err != nil {
		Log.Error("zk.Get(\"%s\") error(%v)", path, err)
		return nil, err
	}
	// fetch and parse comet info
	addr, err := parseNode(string(data))
	if err != nil {
		Log.Error("parseNode(\"%s\") error(%v)", string(data), err)
		return nil, err
	}
	info := &NodeInfo{Addr: addr}
	rpcAddr, ok := addr[ProtocolRPC]
	if !ok || len(rpcAddr) == 0 {
		Log.Crit("zk nodes: \"%s\" don't have rpc addr", path)
		return nil, ErrCometRPC
	}
	// init comet rpc
	// TODO support many rpc
	r, err := rpc.Dial("tcp", rpcAddr[0])
	if err != nil {
		Log.Crit("rpc.Dial(\"%s\") error(%v)", rpcAddr[0], err)
		return nil, ErrCometRPC
	}
	info.PubRPC = r
	Log.Info("zk path: \"%s\" register nodes: \"%s\"", path, node)
	return info, nil
}

func mirrorNode(conn *zk.Conn, node *NodeTree) error {
	var err error
	var children []string

	nodePath := appendPath(node.PathPrefix, node.Name)

	if children, _, err = conn.Children(nodePath); err != nil {
		return err
	}

	for _, child := range children {
		var data []byte

		childPath := appendPath(nodePath, child)

		if data, _, err = conn.Get(childPath); err != nil {
			return err
		}

		childNode := NewNodeTree(child, nodePath, data)

		node.AddChild(childNode)

		if err := mirrorNode(conn, childNode); err != nil {
			return err
		}
	}

	return nil
}

func GetRouters(conn *zk.Conn) []string {
	nodes, _, _, err := conn.ChildrenW(ZOOKEEPER_ROUTER_PATH)
	if err != nil {
		log.Error("get router from zk error:%v", err)
		return []string{}
	}
	return nodes
}

/*
   Read ZK ACL:
   http://zookeeper.apache.org/doc/trunk/zookeeperProgrammers.html#sc_ACLPermissions
*/
func Create(conn *zk.Conn, zkConf conf.Zookeeper, appId string, domainValue string) (string, error) {
	path := concatPath(zkConf.Path, appId)
	resPath, err := conn.Create(path, []byte(domainValue), 0, defaultACL())
	if err != nil {
		return "", err
	}

	return resPath, nil
}

func GetRoomConnector(conn *zk.Conn, roomid string) []string {
	fpath := path.Join(ZOOKEEPER_ROOM_PATH, roomid)
	nodes, _, _, err := conn.ChildrenW(fpath)
	if err != nil {
		log.Error("get roomid:%s from zk error:%v", roomid, err)
		return []string{}
	}
	return nodes
}

func syncConnections(conf *bloorConfig, conn *zk.Conn) {
	// Sync up again
	_, err := conn.Sync(conf.rootZnode)
	if err != nil {
		log.Fatal(err)
	} else {
		log.Printf("Synced connection")
	}
}

// RegisterTmp create a ephemeral node, and watch it, if node droped then send a SIGQUIT to self.
func RegisterTemp(conn *zk.Conn, fpath string, data []byte) error {
	tpath, err := conn.Create(path.Join(fpath), data, zk.FlagEphemeral, zk.WorldACL(zk.PermAll))
	if err != nil {
		log.Error("conn.Create(\"%s\", \"%s\", zk.FlagEphemeral) error(%v)", fpath, string(data), err)
		return err
	}
	log.Debug("create a zookeeper node:%s", tpath)
	return nil
}

func watchService(cpDao dao.ControlPlane, conn *zk.Conn, shutdown <-chan int, done chan<- string, serviceId string) {
	defer func() {
		glog.V(3).Info("Exiting function watchService ", serviceId)
		done <- serviceId
	}()
	for {
		var service dao.Service
		_, zkEvent, err := zzk.LoadServiceW(conn, serviceId, &service)
		if err != nil {
			glog.Errorf("Unable to load service %s: %v", serviceId, err)
			return
		}
		_, _, childEvent, err := conn.ChildrenW(zzk.ServicePath(serviceId))

		glog.V(1).Info("Leader watching for changes to service ", service.Name)

		// check current state
		var serviceStates []*dao.ServiceState
		err = zzk.GetServiceStates(conn, &serviceStates, serviceId)
		if err != nil {
			glog.Error("Unable to retrieve running service states: ", err)
			return
		}

		// Is the service supposed to be running at all?
		switch {
		case service.DesiredState == dao.SVC_STOP:
			shutdownServiceInstances(conn, serviceStates, len(serviceStates))
		case service.DesiredState == dao.SVC_RUN:
			updateServiceInstances(cpDao, conn, &service, serviceStates)
		default:
			glog.Warningf("Unexpected desired state %d for service %s", service.DesiredState, service.Name)
		}

		select {
		case evt := <-zkEvent:
			if evt.Type == zk.EventNodeDeleted {
				glog.V(0).Info("Shutting down due to node delete ", serviceId)
				shutdownServiceInstances(conn, serviceStates, len(serviceStates))
				return
			}
			glog.V(1).Infof("Service %s received event: %v", service.Name, evt)
			continue

		case evt := <-childEvent:
			glog.V(1).Infof("Service %s received child event: %v", service.Name, evt)
			continue

		case <-shutdown:
			glog.V(1).Info("Leader stopping watch on ", service.Name)
			return

		}
	}

}

// watchMessageRoot watch the message root path.
func watchMessageRoot(conn *zk.Conn, fpath string, ch chan *MessageNodeEvent) error {
	for {
		nodes, watch, err := myzk.GetNodesW(conn, fpath)
		if err == myzk.ErrNodeNotExist {
			glog.Warningf("zk don't have node \"%s\", retry in %d second", fpath, waitNodeDelay)
			time.Sleep(waitNodeDelaySecond)
			continue
		} else if err == myzk.ErrNoChild {
			glog.Warningf("zk don't have any children in \"%s\", retry in %d second", fpath, waitNodeDelay)
			// all child died, kick all the nodes
			for _, client := range MessageRPC.Clients {
				glog.V(1).Infof("node: \"%s\" send del node event", client.Addr)
				ch <- &MessageNodeEvent{Event: eventNodeDel, Key: &RPCClient{Addr: client.Addr, Weight: client.Weight}}
			}
			time.Sleep(waitNodeDelaySecond)
			continue
		} else if err != nil {
			glog.Errorf("getNodes error(%v), retry in %d second", err, waitNodeDelay)
			time.Sleep(waitNodeDelaySecond)
			continue
		}
		nodesMap := map[string]bool{}
		// handle new add nodes
		for _, node := range nodes {
			data, _, err := conn.Get(path.Join(fpath, node))
			if err != nil {
				glog.Errorf("zk.Get(\"%s\") error(%v)", path.Join(fpath, node), err)
				continue
			}
			// may contains many addrs split by ,
			//addrs := strings.Split(string(data), ",")
			addrInfos, err := parseMessageAddr(string(data))
			if err != nil {
				glog.Errorf("parseMessageAddr(\"%s\") error(%v)", string(data), err)
				continue
			}
			for _, addInfo := range addrInfos {
				// if not exists in old map then trigger a add event
				if _, ok := MessageRPC.Clients[addInfo.Addr]; !ok {
					ch <- &MessageNodeEvent{Event: eventNodeAdd, Key: addInfo}
				}
				nodesMap[addInfo.Addr] = true
			}
		}
		// handle delete nodes
		for _, client := range MessageRPC.Clients {
			if _, ok := nodesMap[client.Addr]; !ok {
				ch <- &MessageNodeEvent{Event: eventNodeDel, Key: client}
			}
		}
		// blocking wait node changed
		event := <-watch
		glog.Infof("zk path: \"%s\" receive a event %v", fpath, event)
	}
}

// watchMessageRoot watch the message root path.
func watchMessageRoot(conn *zk.Conn, fpath string, ch chan *MessageNodeEvent) error {
	for {
		nodes, watch, err := myzk.GetNodesW(conn, fpath)
		if err == myzk.ErrNodeNotExist {
			log.Warn("zk don't have node \"%s\", retry in %d second", fpath, waitNodeDelay)
			time.Sleep(waitNodeDelaySecond)
			continue
		} else if err == myzk.ErrNoChild {
			log.Warn("zk don't have any children in \"%s\", retry in %d second", fpath, waitNodeDelay)
			// all child died, kick all the nodes
			for _, client := range MessageRPC.Clients {
				log.Debug("node: \"%s\" send del node event", client.Addr)
				ch <- &MessageNodeEvent{Event: eventNodeDel, Key: &WeightRpc{Addr: client.Addr, Weight: client.Weight}}
			}
			time.Sleep(waitNodeDelaySecond)
			continue
		} else if err != nil {
			log.Error("getNodes error(%v), retry in %d second", err, waitNodeDelay)
			time.Sleep(waitNodeDelaySecond)
			continue
		}
		nodesMap := map[string]bool{}
		// handle new add nodes
		for _, node := range nodes {
			data, _, err := conn.Get(path.Join(fpath, node))
			if err != nil {
				log.Error("zk.Get(\"%s\") error(%v)", path.Join(fpath, node), err)
				continue
			}
			// parse message node info
			nodeInfo := &MessageNodeInfo{}
			if err := json.Unmarshal(data, nodeInfo); err != nil {
				log.Error("json.Unmarshal(\"%s\", nodeInfo) error(%v)", string(data), err)
				continue
			}
			for _, addr := range nodeInfo.Rpc {
				// if not exists in old map then trigger a add event
				if _, ok := MessageRPC.Clients[addr]; !ok {
					ch <- &MessageNodeEvent{Event: eventNodeAdd, Key: &WeightRpc{Addr: addr, Weight: nodeInfo.Weight}}
				}
				nodesMap[addr] = true
			}
		}
		// handle delete nodes
		for _, client := range MessageRPC.Clients {
			if _, ok := nodesMap[client.Addr]; !ok {
				ch <- &MessageNodeEvent{Event: eventNodeDel, Key: client}
			}
		}
		// blocking wait node changed
		event := <-watch
		log.Info("zk path: \"%s\" receive a event %v", fpath, event)
	}
}

func CreateNode(path string, conn *zk.Conn) error {
	var err error
	for i := 0; i < 5; i++ {
		_, err = conn.Create(path, []byte{}, 0, zk.WorldACL(zk.PermAll))
		if err == zk.ErrNodeExists || err == nil {
			return nil
		}
		glog.Warningf("Could not create node:%s, %v", path, err)
	}
	return err
}

// create the parent znodes up to path. does not error if any of the
// parent znodes exist.
func createParentPath(path string, conn *zk.Conn, acl []zk.ACL) error {
	parts := strings.Split(path, "/")
	prePath := ""
	for _, p := range parts[1 : len(parts)-1] {
		prePath += "/" + p
		_, err := conn.Create(prePath, []byte{}, 0, acl)
		if err != nil && err != zk.ErrNodeExists {
			return err
		}
	}
	return nil
}

func CreateOrUpdate(zconn zookeeper.Conn, zkPath, value string, flags int, aclv []zookeeper.ACL, recursive bool) (pathCreated string, err error) {
	if recursive {
		pathCreated, err = CreateRecursive(zconn, zkPath, value, 0, zookeeper.WorldACL(zookeeper.PermAll))
	} else {
		pathCreated, err = zconn.Create(zkPath, []byte(value), 0, zookeeper.WorldACL(zookeeper.PermAll))
	}
	if err != nil && ZkErrorEqual(err, zookeeper.ErrNodeExists) {
		pathCreated = ""
		_, err = zconn.Set(zkPath, []byte(value), -1)
	}
	return
}