| 98 | } |
| 99 | |
| 100 | func TestWatchKey(t *testing.T) { |
| 101 | const key = "test" |
| 102 | const max = 100 |
| 103 | const sleep = 15 * time.Millisecond |
| 104 | const totalTestTimeout = 3 * max * sleep |
| 105 | const expectedFactor = 0.75 // we may not see every single value |
| 106 | |
| 107 | withFixtures(t, func(t *testing.T, client Client) { |
| 108 | observedValuesCh := make(chan string, max) |
| 109 | |
| 110 | ctx, cancel := context.WithCancel(context.Background()) |
| 111 | defer cancel() |
| 112 | go func() { |
| 113 | // Start watching before we even start generating values. |
| 114 | // Values will be buffered in the channel. |
| 115 | t.Log("Watching in background", "key", key) |
| 116 | client.WatchKey(ctx, key, func(value any) bool { |
| 117 | observedValuesCh <- value.(string) |
| 118 | return true |
| 119 | }) |
| 120 | }() |
| 121 | |
| 122 | // update value for the key |
| 123 | go func() { |
| 124 | for i := range max { |
| 125 | // Start with sleeping, so that watching client see empty KV store at the beginning. |
| 126 | time.Sleep(sleep) |
| 127 | |
| 128 | err := client.CAS(ctx, key, func(in any) (out any, retry bool, err error) { |
| 129 | return fmt.Sprintf("%d", i), true, nil |
| 130 | }) |
| 131 | |
| 132 | if ctx.Err() != nil { |
| 133 | break |
| 134 | } |
| 135 | require.NoError(t, err) |
| 136 | } |
| 137 | }() |
| 138 | |
| 139 | lastObservedValue := -1 |
| 140 | observedCount := 0 |
| 141 | |
| 142 | totalDeadline := time.After(totalTestTimeout) |
| 143 | |
| 144 | for watching := true; watching; { |
| 145 | select { |
| 146 | case <-totalDeadline: |
| 147 | watching = false |
| 148 | case valStr := <-observedValuesCh: |
| 149 | val, err := strconv.Atoi(valStr) |
| 150 | if err != nil { |
| 151 | t.Fatal("Unexpected value observed:", valStr) |
| 152 | } |
| 153 | |
| 154 | if val <= lastObservedValue { |
| 155 | t.Fatal("Unexpected value observed:", val, "previous:", lastObservedValue) |
| 156 | } |
| 157 | lastObservedValue = val |