Method runDistributor

watch_manager.go:316–340

runDistributor distributes batches to subscribers

()

// runDistributor distributes batches to subscribers
func (wm *watchManager) runDistributor() {
	for {
		select {
		case batch, ok := <-wm.distributeChan:
			if !ok {
				return
			}
			_ = wm.distributeAllMessages(batch)

		case <-wm.workerCtx.Done():
			// drain the distribute channel
			for {
				select {
				case batch, ok := <-wm.distributeChan:
					if !ok {
						return
					}
					_ = wm.distributeAllMessages(batch)
				default:
					return
				}
			}
		}
	}
}

// runVictimCollector collects the victim buckets from the victim channel
// and handle delete bucket operation