Method Watch

kv/etcd/mock.go:87–131

Watch implements the Clientv3Facade interface

(ctx context.Context, key string, opts ...clientv3.OpOption)

// Watch implements the Clientv3Facade interface
func (m *mockKV) Watch(ctx context.Context, key string, opts ...clientv3.OpOption) clientv3.WatchChan {
	watcher := make(chan clientv3.WatchResponse, channelBufferSize)
	consumer := m.createEventConsumer(channelBufferSize)

	go func() {
		defer func() {
			// When this goroutine ends, remove and close the channel written to by the
			// Put and Delete methods as well as closing the channel read by the caller
			// of the Watch method
			m.destroyEventConsumer(consumer)

			// non-blocking send
			select {
			case watcher <- clientv3.WatchResponse{Canceled: true}:
			default:
			}

			close(watcher)
		}()

		for {
			select {
			case <-ctx.Done():
				// Context cancelled for this watcher, run cleanup logic and exit
				return
			case <-m.close:
				// Close method called for all watchers, run cleanup logic and exit
				return
			case e := <-consumer:
				op := clientv3.OpGet(key, opts...)
				match := m.isMatch(op, *e.Kv)

				if match {
					// non-blocking send
					select {
					case watcher <- clientv3.WatchResponse{Events: []*clientv3.Event{&e}}:
					default:
					}
				}
			}
		}
	}()

	return watcher
}

// createEventConsumer creates and returns a new channel that is registered to receive
// events for Puts and Deletes.