Merge pull request #5448 from xiang90/fix_refrsh

etcd: fix refresh feature

Documentation/v2/api.md

@@ -233,10 +233,11 @@ curl http://127.0.0.1:2379/v2/keys/foo -XPUT -d value=bar -d ttl= -d prevExist=t
 
 ### Refreshing key TTL
 
-Keys in etcd can be refreshed without notifying watchers
-this can be achieved by setting the refresh to true when updating a TTL
+Keys in etcd can be refreshed without notifying current watchers.
 
-You cannot update the value of a key when refreshing it
+This can be achieved by setting the refresh to true when updating a TTL.
+
+You cannot update the value of a key when refreshing it.
 
 ```sh
 curl http://127.0.0.1:2379/v2/keys/foo -XPUT -d value=bar -d ttl=5

store/event.go

@@ -30,6 +30,7 @@ type Event struct {
 	Node      *NodeExtern `json:"node,omitempty"`
 	PrevNode  *NodeExtern `json:"prevNode,omitempty"`
 	EtcdIndex uint64      `json:"-"`
+	Refresh   bool        `json:"refresh,omitempty"`
 }
 
 func newEvent(action string, key string, modifiedIndex, createdIndex uint64) *Event {
@@ -64,3 +65,7 @@ func (e *Event) Clone() *Event {
 		PrevNode:  e.PrevNode.Clone(),
 	}
 }
+
+func (e *Event) SetRefresh() {
+	e.Refresh = true
+}

store/event_history.go

@@ -78,24 +78,26 @@ func (eh *EventHistory) scan(key string, recursive bool, index uint64) (*Event,
 	for {
 		e := eh.Queue.Events[i]
 
-		ok := (e.Node.Key == key)
+		if !e.Refresh {
+			ok := (e.Node.Key == key)
 
-		if recursive {
-			// add tailing slash
-			key = path.Clean(key)
-			if key[len(key)-1] != '/' {
-				key = key + "/"
-			}
+			if recursive {
+				// add tailing slash
+				key = path.Clean(key)
+				if key[len(key)-1] != '/' {
+					key = key + "/"
+				}
 
-			ok = ok || strings.HasPrefix(e.Node.Key, key)
-		}
+				ok = ok || strings.HasPrefix(e.Node.Key, key)
+			}
 
-		if (e.Action == Delete || e.Action == Expire) && e.PrevNode != nil && e.PrevNode.Dir {
-			ok = ok || strings.HasPrefix(key, e.PrevNode.Key)
-		}
+			if (e.Action == Delete || e.Action == Expire) && e.PrevNode != nil && e.PrevNode.Dir {
+				ok = ok || strings.HasPrefix(key, e.PrevNode.Key)
+			}
 
-		if ok {
-			return e, nil
+			if ok {
+				return e, nil
+			}
 		}
 
 		i = (i + 1) % eh.Queue.Capacity

store/store.go

@@ -236,6 +236,9 @@ func (s *store) Set(nodePath string, dir bool, value string, expireOpts TTLOptio
 
 	if !expireOpts.Refresh {
 		s.WatcherHub.notify(e)
+	} else {
+		e.SetRefresh()
+		s.WatcherHub.add(e)
 	}
 
 	return e, nil
@@ -314,6 +317,9 @@ func (s *store) CompareAndSwap(nodePath string, prevValue string, prevIndex uint
 
 	if !expireOpts.Refresh {
 		s.WatcherHub.notify(e)
+	} else {
+		e.SetRefresh()
+		s.WatcherHub.add(e)
 	}
 
 	return e, nil
@@ -539,6 +545,9 @@ func (s *store) Update(nodePath string, newValue string, expireOpts TTLOptionSet
 
 	if !expireOpts.Refresh {
 		s.WatcherHub.notify(e)
+	} else {
+		e.SetRefresh()
+		s.WatcherHub.add(e)
 	}
 
 	s.CurrentIndex = nextIndex

store/watcher_hub.go

@@ -115,6 +115,10 @@ func (wh *watcherHub) watch(key string, recursive, stream bool, index, storeInde
 	return w, nil
 }
 
+func (wh *watcherHub) add(e *Event) {
+	e = wh.EventHistory.addEvent(e)
+}
+
 // notify function accepts an event and notify to the watchers.
 func (wh *watcherHub) notify(e *Event) {
 	e = wh.EventHistory.addEvent(e) // add event into the eventHistory