func (ws *watcherSyncer) sendUpdates(updates []api.Update) []api.Update {
	log.WithField("NumUpdates", len(updates)).Debug("Sending syncer updates (if any to send)")
	if len(updates) > 0 {
		ws.callbacks.OnUpdates(updates)
	}
	return nil
}
sendUpdates is used to send the consolidated set of updates.  Returns nil.
package watchersyncer
import (
	log "github.com/sirupsen/logrus"
	"context"
	"sync"
	"github.com/projectcalico/libcalico-go/lib/backend/api"
	"github.com/projectcalico/libcalico-go/lib/backend/model"
	cerrors "github.com/projectcalico/libcalico-go/lib/errors"
)
const (
	maxUpdatesToConsolidate = 1000
)
type ResourceType struct {
	ListInterface model.ListInterface
	UpdateProcessor SyncerUpdateProcessor
}
type SyncerUpdateProcessor interface {
	Process(*model.KVPair) ([]*model.KVPair, error)
	OnSyncerStarting()
}
func New(client api.Client, resourceTypes []ResourceType, callbacks api.SyncerCallbacks) api.Syncer {
	rs := &watcherSyncer{
		watcherCaches: make([]*watcherCache, len(resourceTypes)),
		results:       make(chan interface{}, 2000),
		callbacks:     callbacks,
	}
	for i, r := range resourceTypes {
		rs.watcherCaches[i] = newWatcherCache(client, r, rs.results)
	}
	return rs
}
type watcherSyncer struct {
	status        api.SyncStatus
	watcherCaches []*watcherCache
	results       chan interface{}
	numSynced     int
	callbacks     api.SyncerCallbacks
	wgwc          *sync.WaitGroup
	wgws          *sync.WaitGroup
	cancel        context.CancelFunc
}
func (ws *watcherSyncer) Start() {
	log.Info("Start called")
	ctx, cancel := context.WithCancel(context.Background())
	ws.cancel = cancel
	ws.wgwc = &sync.WaitGroup{}
	ws.wgws = &sync.WaitGroup{}
	go func() {
		ws.run(ctx)
		log.Debug("Watcher syncer run completed")
	}()
}
func (ws *watcherSyncer) Stop() {
	ws.cancel()
	log.Debug("Waiting for watcher caches to stop")
	ws.wgwc.Wait()
	log.Debug("Watcher caches have stopped")
	close(ws.results)
	ws.wgws.Wait()
}
func (ws *watcherSyncer) sendStatusUpdate(status api.SyncStatus) {
	log.WithField("Status", status).Info("Sending status update")
	ws.callbacks.OnStatusUpdated(status)
	ws.status = status
}
func (ws *watcherSyncer) run(ctx context.Context) {
	log.Debug("Sending initial status event and starting watchers")
	ws.wgws.Add(1)
	ws.sendStatusUpdate(api.WaitForDatastore)
	for _, wc := range ws.watcherCaches {
		ws.wgwc.Add(1)
		go func(wc *watcherCache) {
			wc.run(ctx)
			log.Debug("Watcher cache run completed")
			ws.wgwc.Done()
		}(wc)
	}
	log.Info("Starting main event processing loop")
	var updates []api.Update
	for result := range ws.results {
		updates := ws.processResult(updates, result)
	consolidatationloop:
		for ii := 0; ii < maxUpdatesToConsolidate; ii++ {
			select {
			case next := <-ws.results:
				updates = ws.processResult(updates, next)
			default:
				break consolidatationloop
			}
		}
		updates = ws.sendUpdates(updates)
	}
	ws.wgws.Done()
}
func (ws *watcherSyncer) processResult(updates []api.Update, result interface{}) []api.Update {
	switch r := result.(type) {
	case []api.Update:
		if len(updates) == 0 && ws.status == api.WaitForDatastore {
			ws.sendStatusUpdate(api.ResyncInProgress)
		}
		updates = append(updates, r...)
	case error:
		updates = ws.sendUpdates(updates)
		log.WithError(r).Debug("Error received in main syncer event processing loop")
		if ec, ok := ws.callbacks.(api.SyncerParseFailCallbacks); ok {
			log.Debug("syncer receiver can receive parse failed callbacks")
			if pe, ok := r.(cerrors.ErrorParsingDatastoreEntry); ok {
				ec.ParseFailed(pe.RawKey, pe.RawValue)
			}
		}
	case api.SyncStatus:
		log.WithField("SyncUpdate", r).Debug("Received sync status event from watcher")
		if r == api.InSync {
			log.Info("Received InSync event from one of the watcher caches")
			if ws.status == api.WaitForDatastore {
				ws.sendStatusUpdate(api.ResyncInProgress)
			}
			ws.numSynced++
			if ws.numSynced == len(ws.watcherCaches) {
				log.Info("All watchers have sync'd data - sending data and final sync")
				updates = ws.sendUpdates(updates)
				ws.sendStatusUpdate(api.InSync)
			}
		}
	}
	return updates
}