src/server/pkg/watch/watch.go
func NewWatcher(ctx context.Context, client *etcd.Client, trimPrefix, prefix string, template proto.Message, opts ...OpOption) (Watcher, error) {
	eventCh := make(chan *Event)
	done := make(chan struct{})
	// First list the collection to get the current items
	// Sort by mod revision--how the items would have been returned if we watched
	// them from the beginning.
	resp, err := client.Get(ctx, prefix, etcd.WithPrefix(), etcd.WithSort(etcd.SortByModRevision, etcd.SortAscend))
	if err != nil {
		return nil, err
	}

	nextRevision := resp.Header.Revision + 1
	etcdWatcher := etcd.NewWatcher(client)
	// Issue a watch that uses the revision timestamp returned by the
	// Get request earlier.  That way even if some items are added between
	// when we list the collection and when we start watching the collection,
	// we won't miss any items.
	options := []etcd.OpOption{etcd.WithPrefix(), etcd.WithRev(nextRevision)}
	for _, opt := range opts {
		options = append(options, etcd.OpOption(opt))
	}
	rch := etcdWatcher.Watch(ctx, prefix, options...)

	go func() (retErr error) {
		defer func() {
			if retErr != nil {
				select {
				case eventCh <- &Event{
					Err:  retErr,
					Type: EventError,
				}:
				case <-done:
				}
			}
			close(eventCh)
			etcdWatcher.Close()
		}()
		for _, etcdKv := range resp.Kvs {
			eventCh <- &Event{
				Key:      bytes.TrimPrefix(etcdKv.Key, []byte(trimPrefix)),
				Value:    etcdKv.Value,
				Type:     EventPut,
				Rev:      etcdKv.ModRevision,
				Template: template,
			}
		}
		for {
			var resp etcd.WatchResponse
			var ok bool
			select {
			case resp, ok = <-rch:
			case <-done:
				return nil
			}
			if !ok {
				if err := etcdWatcher.Close(); err != nil {
					return err
				}
				etcdWatcher = etcd.NewWatcher(client)
				rch = etcdWatcher.Watch(ctx, prefix, etcd.WithPrefix(), etcd.WithRev(nextRevision))
				continue
			}
			if err := resp.Err(); err != nil {
				return err
			}
			for _, etcdEv := range resp.Events {
				ev := &Event{
					Key:      bytes.TrimPrefix(etcdEv.Kv.Key, []byte(trimPrefix)),
					Value:    etcdEv.Kv.Value,
					Rev:      etcdEv.Kv.ModRevision,
					Template: template,
				}
				if etcdEv.PrevKv != nil {
					ev.PrevKey = bytes.TrimPrefix(etcdEv.PrevKv.Key, []byte(trimPrefix))
					ev.PrevValue = etcdEv.PrevKv.Value
				}
				if etcdEv.Type == etcd.EventTypePut {
					ev.Type = EventPut
				} else {
					ev.Type = EventDelete
				}
				select {
				case eventCh <- ev:
				case <-done:
					return nil
				}
			}
			nextRevision = resp.Header.Revision + 1
		}
	}()

	return &watcher{
		eventCh: eventCh,
		done:    done,
	}, nil
}
func NewWatcher(ctx context.Context, client *etcd.Client, trimPrefix, prefix string, template proto.Message, opts ...OpOption) (Watcher, error) {
	eventCh := make(chan *Event)
	done := make(chan struct{})
	// First list the collection to get the current items
	// Sort by mod revision--how the items would have been returned if we watched
	// them from the beginning.
	resp, err := client.Get(ctx, prefix, etcd.WithPrefix(), etcd.WithSort(etcd.SortByModRevision, etcd.SortAscend))
	if err != nil {
		return nil, err
	}

	nextRevision := resp.Header.Revision + 1
	etcdWatcher := etcd.NewWatcher(client)
	// Issue a watch that uses the revision timestamp returned by the
	// Get request earlier.  That way even if some items are added between
	// when we list the collection and when we start watching the collection,
	// we won't miss any items.
	options := []etcd.OpOption{etcd.WithPrefix(), etcd.WithRev(nextRevision)}
	for _, opt := range opts {
		options = append(options, etcd.OpOption(opt))
	}
	rch := etcdWatcher.Watch(ctx, prefix, options...)

	go func() (retErr error) {
		defer func() {
			if retErr != nil {
				select {
				case eventCh <- &Event{
					Err:  retErr,
					Type: EventError,
				}:
				case <-done:
				}
			}
			close(eventCh)
			etcdWatcher.Close()
		}()
		for _, etcdKv := range resp.Kvs {
			eventCh <- &Event{
				Key:      bytes.TrimPrefix(etcdKv.Key, []byte(trimPrefix)),
				Value:    etcdKv.Value,
				Type:     EventPut,
				Rev:      etcdKv.ModRevision,
				Template: template,
			}
		}
		for {
			var resp etcd.WatchResponse
			var ok bool
			select {
			case resp, ok = <-rch:
			case <-done:
				return nil
			}
			if !ok {
				if err := etcdWatcher.Close(); err != nil {
					return err
				}
				etcdWatcher = etcd.NewWatcher(client)
				rch = etcdWatcher.Watch(ctx, prefix, etcd.WithPrefix(), etcd.WithRev(nextRevision))
				continue
			}
			if err := resp.Err(); err != nil {
				return err
			}
			for _, etcdEv := range resp.Events {
				ev := &Event{
					Key:      bytes.TrimPrefix(etcdEv.Kv.Key, []byte(trimPrefix)),
					Value:    etcdEv.Kv.Value,
					Rev:      etcdEv.Kv.ModRevision,
					Template: template,
				}
				if etcdEv.PrevKv != nil {
					ev.PrevKey = bytes.TrimPrefix(etcdEv.PrevKv.Key, []byte(trimPrefix))
					ev.PrevValue = etcdEv.PrevKv.Value
				}
				if etcdEv.Type == etcd.EventTypePut {
					ev.Type = EventPut
				} else {
					ev.Type = EventDelete
				}
				select {
				case eventCh <- ev:
				case <-done:
					return nil
				}
			}
			nextRevision = resp.Header.Revision + 1
		}
	}()

	return &watcher{
		eventCh: eventCh,
		done:    done,
	}, nil
}
func NewWatcher ( ctx context . Context , client * etcd . Client , trimPrefix , prefix string , template proto . Message , opts ... OpOption ) ( Watcher , error ) { eventCh := make ( chan * Event ) done := make ( chan struct { } ) // First list the collection to get the current items // Sort by mod revision--how the items would have been returned if we watched // them from the beginning. resp , err := client . Get ( ctx , prefix , etcd . WithPrefix ( ) , etcd . WithSort ( etcd . SortByModRevision , etcd . SortAscend ) ) if err != nil { return nil , err } nextRevision := resp . Header . Revision + 1 etcdWatcher := etcd . NewWatcher ( client ) // Issue a watch that uses the revision timestamp returned by the // Get request earlier.  That way even if some items are added between // when we list the collection and when we start watching the collection, // we won't miss any items. options := [ ] etcd . OpOption { etcd . WithPrefix ( ) , etcd . WithRev ( nextRevision ) } for _ , opt := range opts { options = append ( options , etcd . OpOption ( opt ) ) } rch := etcdWatcher . Watch ( ctx , prefix , options ... ) go func ( ) ( retErr error ) { defer func ( ) { if retErr != nil { select { case eventCh <- & Event { Err : retErr , Type : EventError , } : case <- done : } } close ( eventCh ) etcdWatcher . Close ( ) } ( ) for _ , etcdKv := range resp . Kvs { eventCh <- & Event { Key : bytes . TrimPrefix ( etcdKv . Key , [ ] byte ( trimPrefix ) ) , Value : etcdKv . Value , Type : EventPut , Rev : etcdKv . ModRevision , Template : template , } } for { var resp etcd . WatchResponse var ok bool select { case resp , ok = <- rch : case <- done : return nil } if ! ok { if err := etcdWatcher . Close ( ) ; err != nil { return err } etcdWatcher = etcd . NewWatcher ( client ) rch = etcdWatcher . Watch ( ctx , prefix , etcd . WithPrefix ( ) , etcd . WithRev ( nextRevision ) ) continue } if err := resp . Err ( ) ; err != nil { return err } for _ , etcdEv := range resp . Events { ev := & Event { Key : bytes . TrimPrefix ( etcdEv . Kv . Key , [ ] byte ( trimPrefix ) ) , Value : etcdEv . Kv . Value , Rev : etcdEv . Kv . ModRevision , Template : template , } if etcdEv . PrevKv != nil { ev . PrevKey = bytes . TrimPrefix ( etcdEv . PrevKv . Key , [ ] byte ( trimPrefix ) ) ev . PrevValue = etcdEv . PrevKv . Value } if etcdEv . Type == etcd . EventTypePut { ev . Type = EventPut } else { ev . Type = EventDelete } select { case eventCh <- ev : case <- done : return nil } } nextRevision = resp . Header . Revision + 1 } } ( ) return & watcher { eventCh : eventCh , done : done , } , nil }
// NewWatcher watches a given etcd prefix for events.
NewWatcher watches a given etcd prefix for events .