etcdserver/api/v2store/watcher_hub.go
func (wh *watcherHub) watch(key string, recursive, stream bool, index, storeIndex uint64) (Watcher, *v2error.Error) {
	reportWatchRequest()
	event, err := wh.EventHistory.scan(key, recursive, index)

	if err != nil {
		err.Index = storeIndex
		return nil, err
	}

	w := &watcher{
		eventChan:  make(chan *Event, 100), // use a buffered channel
		recursive:  recursive,
		stream:     stream,
		sinceIndex: index,
		startIndex: storeIndex,
		hub:        wh,
	}

	wh.mutex.Lock()
	defer wh.mutex.Unlock()
	// If the event exists in the known history, append the EtcdIndex and return immediately
	if event != nil {
		ne := event.Clone()
		ne.EtcdIndex = storeIndex
		w.eventChan <- ne
		return w, nil
	}

	l, ok := wh.watchers[key]

	var elem *list.Element

	if ok { // add the new watcher to the back of the list
		elem = l.PushBack(w)
	} else { // create a new list and add the new watcher
		l = list.New()
		elem = l.PushBack(w)
		wh.watchers[key] = l
	}

	w.remove = func() {
		if w.removed { // avoid removing it twice
			return
		}
		w.removed = true
		l.Remove(elem)
		atomic.AddInt64(&wh.count, -1)
		reportWatcherRemoved()
		if l.Len() == 0 {
			delete(wh.watchers, key)
		}
	}

	atomic.AddInt64(&wh.count, 1)
	reportWatcherAdded()

	return w, nil
}
func (wh *watcherHub) watch(key string, recursive, stream bool, index, storeIndex uint64) (Watcher, *v2error.Error) {
	reportWatchRequest()
	event, err := wh.EventHistory.scan(key, recursive, index)

	if err != nil {
		err.Index = storeIndex
		return nil, err
	}

	w := &watcher{
		eventChan:  make(chan *Event, 100), // use a buffered channel
		recursive:  recursive,
		stream:     stream,
		sinceIndex: index,
		startIndex: storeIndex,
		hub:        wh,
	}

	wh.mutex.Lock()
	defer wh.mutex.Unlock()
	// If the event exists in the known history, append the EtcdIndex and return immediately
	if event != nil {
		ne := event.Clone()
		ne.EtcdIndex = storeIndex
		w.eventChan <- ne
		return w, nil
	}

	l, ok := wh.watchers[key]

	var elem *list.Element

	if ok { // add the new watcher to the back of the list
		elem = l.PushBack(w)
	} else { // create a new list and add the new watcher
		l = list.New()
		elem = l.PushBack(w)
		wh.watchers[key] = l
	}

	w.remove = func() {
		if w.removed { // avoid removing it twice
			return
		}
		w.removed = true
		l.Remove(elem)
		atomic.AddInt64(&wh.count, -1)
		reportWatcherRemoved()
		if l.Len() == 0 {
			delete(wh.watchers, key)
		}
	}

	atomic.AddInt64(&wh.count, 1)
	reportWatcherAdded()

	return w, nil
}
func ( wh * watcherHub ) watch ( key string , recursive , stream bool , index , storeIndex uint64 ) ( Watcher , * v2error . Error ) { reportWatchRequest ( ) event , err := wh . EventHistory . scan ( key , recursive , index ) if err != nil { err . Index = storeIndex return nil , err } w := & watcher { eventChan : make ( chan * Event , 100 ) , // use a buffered channel recursive : recursive , stream : stream , sinceIndex : index , startIndex : storeIndex , hub : wh , } wh . mutex . Lock ( ) defer wh . mutex . Unlock ( ) // If the event exists in the known history, append the EtcdIndex and return immediately if event != nil { ne := event . Clone ( ) ne . EtcdIndex = storeIndex w . eventChan <- ne return w , nil } l , ok := wh . watchers [ key ] var elem * list . Element if ok { // add the new watcher to the back of the list elem = l . PushBack ( w ) } else { // create a new list and add the new watcher l = list . New ( ) elem = l . PushBack ( w ) wh . watchers [ key ] = l } w . remove = func ( ) { if w . removed { // avoid removing it twice return } w . removed = true l . Remove ( elem ) atomic . AddInt64 ( & wh . count , - 1 ) reportWatcherRemoved ( ) if l . Len ( ) == 0 { delete ( wh . watchers , key ) } } atomic . AddInt64 ( & wh . count , 1 ) reportWatcherAdded ( ) return w , nil }
// Watch function returns a Watcher.
// If recursive is true, the first change after index under key will be sent to the event channel of the watcher.
// If recursive is false, the first change after index at key will be sent to the event channel of the watcher.
// If index is zero, watch will start from the current index + 1.
Watch function returns a Watcher . If recursive is true the first change after index under key will be sent to the event channel of the watcher . If recursive is false the first change after index at key will be sent to the event channel of the watcher . If index is zero watch will start from the current index + 1 .