Method Start

watch_manager.go:609–639

Start starts the watch manager Implements Component interface

(ctx context.Context)

// Start starts the watch manager
// Implements Component interface
func (wm *watchManager) Start(ctx context.Context) error {
	if wm.isClosed() {
		return ErrWatchManagerClosed
	}

	wm.muStarted.RLock()
	if wm.started {
		wm.muStarted.RUnlock()
		return nil
	}

	wm.started = true
	wm.muStarted.RUnlock()

	// use a local ready channel to wait for goroutine startup
	ready := make(chan struct{})

	go func() {
		close(ready) // signal that goroutine has started
		wm.startDistributor()
	}()

	// wait for distributor goroutine to start before returning
	select {
	case <-ready:
		utils.GetLogger().Printf("[watch_manager] Watch manager distributor started\n")
		return nil
	case <-time.After(5 * time.Second):
		return fmt.Errorf("timeout waiting for watch manager distributor to start")
	}
}

// Stop stops the watch manager
// Notifies all subscribers that the database is closing and closes all subscription channels