Function TestWatchPrefix

kv/kv_test.go:172–263

(t *testing.T)

}

func TestWatchPrefix(t *testing.T) {
	withFixtures(t, func(t *testing.T, client Client) {
		const prefix = "test/"
		const prefix2 = "ignore/"

		// We are going to generate this number of updates, sleeping between each update.
		const max = 100
		const sleep = time.Millisecond * 10
		// etcd seems to be quite slow. If we finish faster, test will end sooner.
		// (We regularly see generators taking up to 5 seconds to produce all messages on some platforms!)
		const totalTestTimeout = 10 * time.Second

		observedKeysCh := make(chan string, max)

		ctx, cancel := context.WithCancel(context.Background())
		defer cancel()

		wg := sync.WaitGroup{}

		wg.Add(1)
		go func() {
			defer wg.Done()

			// start watching before we even start generating values. values will be buffered
			client.WatchPrefix(ctx, prefix, func(key string, _ interface{}) bool {
				observedKeysCh <- key
				return true
			})
		}()

		gen := func(p string) {
			defer wg.Done()

			start := time.Now()
			for i := 0; i < max && ctx.Err() == nil; i++ {
				// Start with sleeping, so that watching client can see empty KV store at the beginning.
				time.Sleep(sleep)

				key := fmt.Sprintf("%s%d", p, i)
				err := client.CAS(ctx, key, func(interface{}) (out interface{}, retry bool, err error) {
					return key, true, nil
				})

				if ctx.Err() != nil {
					break
				}
				require.NoError(t, err)
			}
			t.Log("Generator finished in", time.Since(start))
		}

		wg.Add(2)
		go gen(prefix)
		go gen(prefix2) // we don't want to see these keys reported

		observedKeys := map[string]int{}

		totalDeadline := time.After(totalTestTimeout)