src/server/pkg/collection/collection.go
func (c *readonlyCollection) WatchByIndex(index *Index, val interface{}) (watch.Watcher, error) {
	eventCh := make(chan *watch.Event)
	done := make(chan struct{})
	watcher, err := watch.NewWatcher(c.ctx, c.etcdClient, c.prefix, c.indexDir(index, val), c.template)
	if err != nil {
		return nil, err
	}
	go func() (retErr error) {
		defer func() {
			if retErr != nil {
				eventCh <- &watch.Event{
					Type: watch.EventError,
					Err:  retErr,
				}
				watcher.Close()
			}
			close(eventCh)
		}()
		for {
			var ev *watch.Event
			var ok bool
			select {
			case ev, ok = <-watcher.Watch():
			case <-done:
				watcher.Close()
				return nil
			}
			if !ok {
				watcher.Close()
				return nil
			}

			var directEv *watch.Event
			switch ev.Type {
			case watch.EventError:
				// pass along the error
				return ev.Err
			case watch.EventPut:
				resp, err := c.get(c.Path(path.Base(string(ev.Key))))
				if err != nil {
					return err
				}
				if len(resp.Kvs) == 0 {
					// this happens only if the item was deleted shortly after
					// we receive this event.
					continue
				}
				directEv = &watch.Event{
					Key:      []byte(path.Base(string(ev.Key))),
					Value:    resp.Kvs[0].Value,
					Type:     ev.Type,
					Template: c.template,
				}
			case watch.EventDelete:
				directEv = &watch.Event{
					Key:      []byte(path.Base(string(ev.Key))),
					Type:     ev.Type,
					Template: c.template,
				}
			}
			eventCh <- directEv
		}
	}()
	return watch.MakeWatcher(eventCh, done), nil
}
func (c *readonlyCollection) WatchByIndex(index *Index, val interface{}) (watch.Watcher, error) {
	eventCh := make(chan *watch.Event)
	done := make(chan struct{})
	watcher, err := watch.NewWatcher(c.ctx, c.etcdClient, c.prefix, c.indexDir(index, val), c.template)
	if err != nil {
		return nil, err
	}
	go func() (retErr error) {
		defer func() {
			if retErr != nil {
				eventCh <- &watch.Event{
					Type: watch.EventError,
					Err:  retErr,
				}
				watcher.Close()
			}
			close(eventCh)
		}()
		for {
			var ev *watch.Event
			var ok bool
			select {
			case ev, ok = <-watcher.Watch():
			case <-done:
				watcher.Close()
				return nil
			}
			if !ok {
				watcher.Close()
				return nil
			}

			var directEv *watch.Event
			switch ev.Type {
			case watch.EventError:
				// pass along the error
				return ev.Err
			case watch.EventPut:
				resp, err := c.get(c.Path(path.Base(string(ev.Key))))
				if err != nil {
					return err
				}
				if len(resp.Kvs) == 0 {
					// this happens only if the item was deleted shortly after
					// we receive this event.
					continue
				}
				directEv = &watch.Event{
					Key:      []byte(path.Base(string(ev.Key))),
					Value:    resp.Kvs[0].Value,
					Type:     ev.Type,
					Template: c.template,
				}
			case watch.EventDelete:
				directEv = &watch.Event{
					Key:      []byte(path.Base(string(ev.Key))),
					Type:     ev.Type,
					Template: c.template,
				}
			}
			eventCh <- directEv
		}
	}()
	return watch.MakeWatcher(eventCh, done), nil
}
func ( c * readonlyCollection ) WatchByIndex ( index * Index , val interface { } ) ( watch . Watcher , error ) { eventCh := make ( chan * watch . Event ) done := make ( chan struct { } ) watcher , err := watch . NewWatcher ( c . ctx , c . etcdClient , c . prefix , c . indexDir ( index , val ) , c . template ) if err != nil { return nil , err } go func ( ) ( retErr error ) { defer func ( ) { if retErr != nil { eventCh <- & watch . Event { Type : watch . EventError , Err : retErr , } watcher . Close ( ) } close ( eventCh ) } ( ) for { var ev * watch . Event var ok bool select { case ev , ok = <- watcher . Watch ( ) : case <- done : watcher . Close ( ) return nil } if ! ok { watcher . Close ( ) return nil } var directEv * watch . Event switch ev . Type { case watch . EventError : // pass along the error return ev . Err case watch . EventPut : resp , err := c . get ( c . Path ( path . Base ( string ( ev . Key ) ) ) ) if err != nil { return err } if len ( resp . Kvs ) == 0 { // this happens only if the item was deleted shortly after // we receive this event. continue } directEv = & watch . Event { Key : [ ] byte ( path . Base ( string ( ev . Key ) ) ) , Value : resp . Kvs [ 0 ] . Value , Type : ev . Type , Template : c . template , } case watch . EventDelete : directEv = & watch . Event { Key : [ ] byte ( path . Base ( string ( ev . Key ) ) ) , Type : ev . Type , Template : c . template , } } eventCh <- directEv } } ( ) return watch . MakeWatcher ( eventCh , done ) , nil }
// WatchByIndex watches items in a collection that match a particular index
WatchByIndex watches items in a collection that match a particular index