| 228 | } |
| 229 | |
| 230 | func (w *Watcher) dispatchLoop(ctx context.Context) { |
| 231 | for { |
| 232 | batch, ok := w.nextPendingBatch(ctx) |
| 233 | if !ok { |
| 234 | return |
| 235 | } |
| 236 | queue := w.getAuthQueue() |
| 237 | if queue == nil { |
| 238 | if ctx.Err() != nil { |
| 239 | return |
| 240 | } |
| 241 | time.Sleep(10 * time.Millisecond) |
| 242 | continue |
| 243 | } |
| 244 | for _, update := range batch { |
| 245 | select { |
| 246 | case queue <- update: |
| 247 | case <-ctx.Done(): |
| 248 | return |
| 249 | } |
| 250 | } |
| 251 | } |
| 252 | } |
| 253 | |
| 254 | func (w *Watcher) nextPendingBatch(ctx context.Context) ([]AuthUpdate, bool) { |
| 255 | w.dispatchMu.Lock() |