(ctx context.Context)
| 450 | } |
| 451 | |
| 452 | func (s *Service) consumeAuthUpdates(ctx context.Context) { |
| 453 | ctx = coreauth.WithSkipPersist(ctx) |
| 454 | for { |
| 455 | select { |
| 456 | case <-ctx.Done(): |
| 457 | return |
| 458 | case update, ok := <-s.authUpdates: |
| 459 | if !ok { |
| 460 | return |
| 461 | } |
| 462 | updates := []watcher.AuthUpdate{update} |
| 463 | labelDrain: |
| 464 | for { |
| 465 | select { |
| 466 | case nextUpdate := <-s.authUpdates: |
| 467 | updates = append(updates, nextUpdate) |
| 468 | default: |
| 469 | break labelDrain |
| 470 | } |
| 471 | } |
| 472 | s.handleAuthUpdates(ctx, updates) |
| 473 | } |
| 474 | } |
| 475 | } |
| 476 | |
| 477 | func (s *Service) emitAuthUpdate(ctx context.Context, update watcher.AuthUpdate) { |
| 478 | if s == nil { |
no test coverage detected