Method startDistributor

watch_manager.go:217–244

startDistributor starts both the collector and distributor goroutines

()

// startDistributor starts both the collector and distributor goroutines
func (wm *watchManager) startDistributor() {
	defer wm.cleanUpSubscribers()

	// start the victim collector goroutine
	// it collects the victim buckets from the victim channel
	// and handle delete bucket operation
	wm.wg.Add(1)
	go func() {
		defer wm.wg.Done()
		wm.runVictimCollector()
	}()

	// Start the distributor goroutine (consumes from distributeChan)
	wm.wg.Add(1)
	go func() {
		defer wm.wg.Done()
		wm.runDistributor()
	}()

	// start the collector goroutine (collects messages into batches)
	wm.wg.Add(1)
	go func() {
		defer wm.wg.Done()
		wm.runCollector()
	}()

	wm.wg.Wait()
}

// runCollector collects messages from watchChan and batches them
func (wm *watchManager) runCollector() {