// Watches a given key prefix for changes.
func WatchKeyHandler(w http.ResponseWriter, req *http.Request, s Server) error {
	var err error
	vars := mux.Vars(req)
	key := "/" + vars["key"]

	// Create a command to watch from a given index (default 0).
	var sinceIndex uint64 = 0
	if req.Method == "POST" {
		sinceIndex, err = strconv.ParseUint(string(req.FormValue("index")), 10, 64)
		if err != nil {
			return etcdErr.NewError(203, "Watch From Index", s.Store().Index())
		}
	}

	// Start the watcher on the store.
	watcher, err := s.Store().Watch(key, false, false, sinceIndex)
	if err != nil {
		return etcdErr.NewError(500, key, s.Store().Index())
	}
	event := <-watcher.EventChan

	// Convert event to a response and write to client.
	w.WriteHeader(http.StatusOK)
	if req.Method == "HEAD" {
		return nil
	}
	b, _ := json.Marshal(event.Response(s.Store().Index()))
	w.Write(b)
	return nil
}

// Sets the value for a given key.
func SetKeyHandler(w http.ResponseWriter, req *http.Request, s Server) error {
	vars := mux.Vars(req)
	key := "/" + vars["key"]

	req.ParseForm()

	// Parse non-blank value.
	value := req.Form.Get("value")
	if len(value) == 0 {
		return etcdErr.NewError(200, "Set", s.Store().Index())
	}

	// Convert time-to-live to an expiration time.
	expireTime, err := store.TTL(req.Form.Get("ttl"))
	if err != nil {
		return etcdErr.NewError(202, "Set", s.Store().Index())
	}

	// If the "prevValue" is specified then test-and-set. Otherwise create a new key.
	var c raft.Command
	if prevValueArr, ok := req.Form["prevValue"]; ok {
		if len(prevValueArr[0]) > 0 {
			// test against previous value
			c = s.Store().CommandFactory().CreateCompareAndSwapCommand(key, value, prevValueArr[0], 0, expireTime)
		} else {
			// test against existence
			c = s.Store().CommandFactory().CreateCreateCommand(key, false, value, expireTime, false)
		}

	} else {
		c = s.Store().CommandFactory().CreateSetCommand(key, false, value, expireTime)
	}

	return s.Dispatch(c, w, req)
}

// getIndexHandler retrieves the current lock index.
// The "field" parameter specifies to read either the lock "index" or lock "value".
func (h *handler) getIndexHandler(w http.ResponseWriter, req *http.Request) error {
	h.client.SyncCluster()

	vars := mux.Vars(req)
	keypath := path.Join(prefix, vars["key"])
	field := req.FormValue("field")
	if len(field) == 0 {
		field = "value"
	}

	// Read all indices.
	resp, err := h.client.Get(keypath, true, true)
	if err != nil {
		return err
	}
	nodes := lockNodes{resp.Node.Nodes}

	// Write out the requested field.
	if node := nodes.First(); node != nil {
		switch field {
		case "index":
			w.Write([]byte(path.Base(node.Key)))

		case "value":
			w.Write([]byte(node.Value))

		default:
			return etcdErr.NewError(etcdErr.EcodeInvalidField, "Get", 0)
		}
	}

	return nil
}

// deleteHandler remove a given leader.
func (h *handler) deleteHandler(w http.ResponseWriter, req *http.Request) error {
	vars := mux.Vars(req)
	name := req.FormValue("name")
	if name == "" {
		return etcdErr.NewError(etcdErr.EcodeNameRequired, "Delete", 0)
	}

	// Proxy the request to the the lock service.
	u, err := url.Parse(fmt.Sprintf("%s/mod/v2/lock/%s", h.addr, vars["key"]))
	if err != nil {
		return err
	}
	q := u.Query()
	q.Set("value", name)
	u.RawQuery = q.Encode()

	r, err := http.NewRequest("DELETE", u.String(), nil)
	if err != nil {
		return err
	}

	// Read from the leader lock.
	resp, err := h.client.Do(r)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	w.WriteHeader(resp.StatusCode)
	io.Copy(w, resp.Body)

	return nil
}

func GetHandler(w http.ResponseWriter, req *http.Request, s Server) error {
	vars := mux.Vars(req)
	key := "/" + vars["key"]

	// Help client to redirect the request to the current leader
	if req.FormValue("consistent") == "true" && s.State() != raft.Leader {
		leader := s.Leader()
		hostname, _ := s.ClientURL(leader)

		url, err := url.Parse(hostname)
		if err != nil {
			log.Warn("Redirect cannot parse hostName ", hostname)
			return err
		}
		url.RawQuery = req.URL.RawQuery
		url.Path = req.URL.Path

		log.Debugf("Redirect consistent get to %s", url.String())
		http.Redirect(w, req, url.String(), http.StatusTemporaryRedirect)
		return nil
	}

	recursive := (req.FormValue("recursive") == "true")
	sort := (req.FormValue("sorted") == "true")
	waitIndex := req.FormValue("waitIndex")
	stream := (req.FormValue("stream") == "true")

	if req.FormValue("wait") == "true" {
		return handleWatch(key, recursive, stream, waitIndex, w, s)
	}

	return handleGet(key, recursive, sort, w, s)
}

// Checks whether a given version is supported.
func (ps *PeerServer) VersionCheckHttpHandler(w http.ResponseWriter, req *http.Request) {
	log.Debugf("[recv] Get %s%s ", ps.Config.URL, req.URL.Path)
	vars := mux.Vars(req)
	version, _ := strconv.Atoi(vars["version"])
	if version >= store.MinVersion() && version <= store.MaxVersion() {
		w.WriteHeader(http.StatusOK)
	} else {
		w.WriteHeader(http.StatusForbidden)
	}
}

// acquireHandler attempts to acquire a lock on the given key.
// The "key" parameter specifies the resource to lock.
// The "value" parameter specifies a value to associate with the lock.
// The "ttl" parameter specifies how long the lock will persist for.
// The "timeout" parameter specifies how long the request should wait for the lock.
func (h *handler) acquireHandler(w http.ResponseWriter, req *http.Request) error {
	h.client.SyncCluster()

	// Setup connection watcher.
	closeNotifier, _ := w.(http.CloseNotifier)
	closeChan := closeNotifier.CloseNotify()
	stopChan := make(chan bool)

	// Parse the lock "key".
	vars := mux.Vars(req)
	keypath := path.Join(prefix, vars["key"])
	value := req.FormValue("value")

	// Parse "timeout" parameter.
	var timeout int
	var err error
	if req.FormValue("timeout") == "" {
		timeout = -1
	} else if timeout, err = strconv.Atoi(req.FormValue("timeout")); err != nil {
		return etcdErr.NewError(etcdErr.EcodeTimeoutNaN, "Acquire", 0)
	}
	timeout = timeout + 1

	// Parse TTL.
	ttl, err := strconv.Atoi(req.FormValue("ttl"))
	if err != nil {
		return etcdErr.NewError(etcdErr.EcodeTTLNaN, "Acquire", 0)
	}

	// If node exists then just watch it. Otherwise create the node and watch it.
	node, index, pos := h.findExistingNode(keypath, value)
	if index > 0 {
		if pos == 0 {
			// If lock is already acquired then update the TTL.
			h.client.Update(node.Key, node.Value, uint64(ttl))
		} else {
			// Otherwise watch until it becomes acquired (or errors).
			err = h.watch(keypath, index, nil)
		}
	} else {
		index, err = h.createNode(keypath, value, ttl, closeChan, stopChan)
	}

	// Stop all goroutines.
	close(stopChan)

	// Check for an error.
	if err != nil {
		return err
	}

	// Write response.
	w.Write([]byte(strconv.Itoa(index)))
	return nil
}

func PostHandler(w http.ResponseWriter, req *http.Request, s Server) error {
	vars := mux.Vars(req)
	key := "/" + vars["key"]

	value := req.FormValue("value")
	dir := (req.FormValue("dir") == "true")
	expireTime, err := store.TTL(req.FormValue("ttl"))
	if err != nil {
		return etcdErr.NewError(etcdErr.EcodeTTLNaN, "Create", s.Store().Index())
	}

	c := s.Store().CommandFactory().CreateCreateCommand(key, dir, value, expireTime, true)
	return s.Dispatch(c, w, req)
}

// renewLockHandler attempts to update the TTL on an existing lock.
// Returns a 200 OK if successful. Returns non-200 on error.
func (h *handler) renewLockHandler(w http.ResponseWriter, req *http.Request) error {
	h.client.SyncCluster()

	// Read the lock path.
	vars := mux.Vars(req)
	keypath := path.Join(prefix, vars["key"])

	// Parse new TTL parameter.
	ttl, err := strconv.Atoi(req.FormValue("ttl"))
	if err != nil {
		return etcdErr.NewError(etcdErr.EcodeTTLNaN, "Renew", 0)
	}

	// Read and set defaults for index and value.
	index := req.FormValue("index")
	value := req.FormValue("value")
	if len(index) == 0 && len(value) == 0 {
		return etcdErr.NewError(etcdErr.EcodeIndexOrValueRequired, "Renew", 0)
	}

	if len(index) == 0 {
		// If index is not specified then look it up by value.
		resp, err := h.client.Get(keypath, true, true)
		if err != nil {
			return err
		}
		nodes := lockNodes{resp.Node.Nodes}
		node, _ := nodes.FindByValue(value)
		if node == nil {
			return etcdErr.NewError(etcdErr.EcodeKeyNotFound, "Renew", 0)
		}
		index = path.Base(node.Key)

	} else if len(value) == 0 {
		// If value is not specified then default it to the previous value.
		resp, err := h.client.Get(path.Join(keypath, index), true, false)
		if err != nil {
			return err
		}
		value = resp.Node.Value
	}

	// Renew the lock, if it exists.
	if _, err = h.client.Update(path.Join(keypath, index), value, uint64(ttl)); err != nil {
		return err
	}

	return nil
}

// getHandler retrieves the current leader.
func (h *handler) getHandler(w http.ResponseWriter, req *http.Request) error {
	vars := mux.Vars(req)

	// Proxy the request to the lock service.
	url := fmt.Sprintf("%s/mod/v2/lock/%s?field=value", h.addr, vars["key"])
	resp, err := h.client.Get(url)
	if err != nil {
		return err
	}
	defer resp.Body.Close()

	w.WriteHeader(resp.StatusCode)
	io.Copy(w, resp.Body)
	return nil
}

// Response to remove request
func (ps *PeerServer) RemoveHttpHandler(w http.ResponseWriter, req *http.Request) {
	if req.Method != "DELETE" {
		w.WriteHeader(http.StatusMethodNotAllowed)
		return
	}

	vars := mux.Vars(req)
	command := &RemoveCommand{
		Name: vars["name"],
	}

	log.Debugf("[recv] Remove Request [%s]", command.Name)

	ps.server.Dispatch(command, w, req)
}

// setHandler attempts to set the current leader.
func (h *handler) setHandler(w http.ResponseWriter, req *http.Request) error {
	vars := mux.Vars(req)
	name := req.FormValue("name")
	if name == "" {
		return etcdErr.NewError(etcdErr.EcodeNameRequired, "Set", 0)
	}

	// Proxy the request to the the lock service.
	u, err := url.Parse(fmt.Sprintf("%s/mod/v2/lock/%s", h.addr, vars["key"]))
	if err != nil {
		return err
	}
	q := u.Query()
	q.Set("value", name)
	q.Set("ttl", req.FormValue("ttl"))
	q.Set("timeout", req.FormValue("timeout"))
	u.RawQuery = q.Encode()

	r, err := http.NewRequest("POST", u.String(), nil)
	if err != nil {
		return err
	}

	// Close request if this connection disconnects.
	closeNotifier, _ := w.(http.CloseNotifier)
	stopChan := make(chan bool)
	defer close(stopChan)
	go func() {
		select {
		case <-closeNotifier.CloseNotify():
			h.transport.CancelRequest(r)
		case <-stopChan:
		}
	}()

	// Read from the leader lock.
	resp, err := h.client.Do(r)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	w.WriteHeader(resp.StatusCode)
	io.Copy(w, resp.Body)
	return nil
}

// Retrieves the value for a given key.
func GetKeyHandler(w http.ResponseWriter, req *http.Request, s Server) error {
	vars := mux.Vars(req)
	key := "/" + vars["key"]

	// Retrieve the key from the store.
	event, err := s.Store().Get(key, false, false)
	if err != nil {
		return err
	}

	// Convert event to a response and write to client.
	b, _ := json.Marshal(event.Response(s.Store().Index()))

	w.WriteHeader(http.StatusOK)
	w.Write(b)

	return nil
}

func DeleteHandler(w http.ResponseWriter, req *http.Request, s Server) error {
	vars := mux.Vars(req)
	key := "/" + vars["key"]

	recursive := (req.FormValue("recursive") == "true")
	dir := (req.FormValue("dir") == "true")

	req.ParseForm()
	_, valueOk := req.Form["prevValue"]
	_, indexOk := req.Form["prevIndex"]

	if !valueOk && !indexOk {
		c := s.Store().CommandFactory().CreateDeleteCommand(key, dir, recursive)
		return s.Dispatch(c, w, req)
	}

	var err error
	prevIndex := uint64(0)
	prevValue := req.Form.Get("prevValue")

	if indexOk {
		prevIndexStr := req.Form.Get("prevIndex")
		prevIndex, err = strconv.ParseUint(prevIndexStr, 10, 64)

		// bad previous index
		if err != nil {
			return etcdErr.NewError(etcdErr.EcodeIndexNaN, "CompareAndDelete", s.Store().Index())
		}
	}

	if valueOk {
		if prevValue == "" {
			return etcdErr.NewError(etcdErr.EcodePrevValueRequired, "CompareAndDelete", s.Store().Index())
		}
	}

	c := s.Store().CommandFactory().CreateCompareAndDeleteCommand(key, prevValue, prevIndex)
	return s.Dispatch(c, w, req)
}

// releaseLockHandler deletes the lock.
func (h *handler) releaseLockHandler(w http.ResponseWriter, req *http.Request) error {
	h.client.SyncCluster()

	vars := mux.Vars(req)
	keypath := path.Join(prefix, vars["key"])

	// Read index and value parameters.
	index := req.FormValue("index")
	value := req.FormValue("value")
	if len(index) == 0 && len(value) == 0 {
		return etcdErr.NewError(etcdErr.EcodeIndexOrValueRequired, "Release", 0)
	} else if len(index) != 0 && len(value) != 0 {
		return etcdErr.NewError(etcdErr.EcodeIndexValueMutex, "Release", 0)
	}

	// Look up index by value if index is missing.
	if len(index) == 0 {
		resp, err := h.client.Get(keypath, true, true)
		if err != nil {
			return err
		}
		nodes := lockNodes{resp.Node.Nodes}
		node, _ := nodes.FindByValue(value)
		if node == nil {
			return etcdErr.NewError(etcdErr.EcodeKeyNotFound, "Release", 0)
		}
		index = path.Base(node.Key)
	}

	// Delete the lock.
	if _, err := h.client.Delete(path.Join(keypath, index), false); err != nil {
		return err
	}

	return nil
}

func PutHandler(w http.ResponseWriter, req *http.Request, s Server) error {
	var c raft.Command

	vars := mux.Vars(req)
	key := "/" + vars["key"]

	req.ParseForm()

	value := req.Form.Get("value")
	dir := (req.FormValue("dir") == "true")

	expireTime, err := store.TTL(req.Form.Get("ttl"))
	if err != nil {
		return etcdErr.NewError(etcdErr.EcodeTTLNaN, "Update", s.Store().Index())
	}

	_, valueOk := req.Form["prevValue"]
	prevValue := req.FormValue("prevValue")

	_, indexOk := req.Form["prevIndex"]
	prevIndexStr := req.FormValue("prevIndex")

	_, existOk := req.Form["prevExist"]
	prevExist := req.FormValue("prevExist")

	// Set handler: create a new node or replace the old one.
	if !valueOk && !indexOk && !existOk {
		return SetHandler(w, req, s, key, dir, value, expireTime)
	}

	// update with test
	if existOk {
		if prevExist == "false" {
			// Create command: create a new node. Fail, if a node already exists
			// Ignore prevIndex and prevValue
			return CreateHandler(w, req, s, key, dir, value, expireTime)
		}

		if prevExist == "true" && !indexOk && !valueOk {
			return UpdateHandler(w, req, s, key, value, expireTime)
		}
	}

	var prevIndex uint64

	if indexOk {
		prevIndex, err = strconv.ParseUint(prevIndexStr, 10, 64)

		// bad previous index
		if err != nil {
			return etcdErr.NewError(etcdErr.EcodeIndexNaN, "CompareAndSwap", s.Store().Index())
		}
	} else {
		prevIndex = 0
	}

	if valueOk {
		if prevValue == "" {
			return etcdErr.NewError(etcdErr.EcodePrevValueRequired, "CompareAndSwap", s.Store().Index())
		}
	}

	c = s.Store().CommandFactory().CreateCompareAndSwapCommand(key, value, prevValue, prevIndex, expireTime)
	return s.Dispatch(c, w, req)
}

// Retrieve single peer or standby.
func (ps *PeerServer) getMachineHttpHandler(w http.ResponseWriter, req *http.Request) {
	vars := mux.Vars(req)
	m := ps.getMachineMessage(vars["name"], ps.raftServer.Leader())
	json.NewEncoder(w).Encode(m)
}

// acquireHandler attempts to acquire a lock on the given key.
// The "key" parameter specifies the resource to lock.
// The "value" parameter specifies a value to associate with the lock.
// The "ttl" parameter specifies how long the lock will persist for.
// The "timeout" parameter specifies how long the request should wait for the lock.
func (h *handler) acquireHandler(w http.ResponseWriter, req *http.Request) error {
	h.client.SyncCluster()

	// Setup connection watcher.
	closeNotifier, _ := w.(http.CloseNotifier)
	closeChan := closeNotifier.CloseNotify()

	// Wrap closeChan so we can pass it to subsequent components
	timeoutChan := make(chan bool)
	stopChan := make(chan bool)
	go func() {
		select {
		case <-closeChan:
			// Client closed connection
			stopChan <- true
		case <-timeoutChan:
			// Timeout expired
			stopChan <- true
		case <-stopChan:
		}
		close(stopChan)
	}()

	// Parse the lock "key".
	vars := mux.Vars(req)
	keypath := path.Join(prefix, vars["key"])
	value := req.FormValue("value")

	// Parse "timeout" parameter.
	var timeout int
	var err error
	if req.FormValue("timeout") == "" {
		timeout = -1
	} else if timeout, err = strconv.Atoi(req.FormValue("timeout")); err != nil {
		return etcdErr.NewError(etcdErr.EcodeTimeoutNaN, "Acquire", 0)
	}

	// Parse TTL.
	ttl, err := strconv.Atoi(req.FormValue("ttl"))
	if err != nil {
		return etcdErr.NewError(etcdErr.EcodeTTLNaN, "Acquire", 0)
	}

	// Search for the node
	_, index, pos := h.findExistingNode(keypath, value)
	if index == 0 {
		// Node doesn't exist; Create it
		pos = -1 // Invalidate previous position
		index, err = h.createNode(keypath, value, ttl)
		if err != nil {
			return err
		}
	}

	indexpath := path.Join(keypath, strconv.Itoa(index))

	// If pos != 0, we do not already have the lock
	if pos != 0 {
		if timeout == 0 {
			// Attempt to get lock once, no waiting
			err = h.get(keypath, index)
		} else {
			// Keep updating TTL while we wait
			go h.ttlKeepAlive(keypath, value, ttl, stopChan)

			// Start timeout
			go h.timeoutExpire(timeout, timeoutChan, stopChan)

			// wait for lock
			err = h.watch(keypath, index, stopChan)
		}
	}

	// Return on error, deleting our lock request on the way
	if err != nil {
		if index > 0 {
			h.client.Delete(indexpath, false)
		}
		return err
	}

	// Check for connection disconnect before we write the lock index.
	select {
	case <-stopChan:
		err = errors.New("user interrupted")
	default:
	}

	// Update TTL one last time if lock was acquired. Otherwise delete.
	if err == nil {
		h.client.Update(indexpath, value, uint64(ttl))
	} else {
		h.client.Delete(indexpath, false)
	}

//.........这里部分代码省略.........

// Removes a key from the store.
func DeleteKeyHandler(w http.ResponseWriter, req *http.Request, s Server) error {
	vars := mux.Vars(req)
	key := "/" + vars["key"]
	c := s.Store().CommandFactory().CreateDeleteCommand(key, false, false)
	return s.Dispatch(c, w, req)
}

// Removes a machine from the cluster.
func (ps *PeerServer) removeMachineHttpHandler(w http.ResponseWriter, req *http.Request) {
	vars := mux.Vars(req)
	c := &RemoveCommandV2{Name: vars["name"]}
	log.Debugf("[recv] Remove Request [%s]", c.Name)
	ps.server.Dispatch(c, w, req)
}