Method initiateStreaming

go/logic/migrator.go:1443–1478

initiateStreaming begins streaming of binary log events and registers listeners for such events

()

// initiateStreaming begins streaming of binary log events and registers listeners for such events
func (mgtr *Migrator) initiateStreaming() error {
	mgtr.eventsStreamer = NewEventsStreamer(mgtr.migrationContext)
	if err := mgtr.eventsStreamer.InitDBConnections(); err != nil {
		return err
	}
	mgtr.eventsStreamer.AddListener(
		false,
		mgtr.migrationContext.DatabaseName,
		mgtr.migrationContext.GetChangelogTableName(),
		func(dmlEntry *binlog.BinlogEntry) error {
			return mgtr.onChangelogEvent(dmlEntry)
		},
	)

	go func() {
		mgtr.migrationContext.Log.Debugf("Beginning streaming")
		err := mgtr.eventsStreamer.StreamEvents(mgtr.canStopStreaming)
		if err != nil {
			// Use helper to prevent deadlock if listenOnPanicAbort already exited
			_ = base.SendWithContext(mgtr.migrationContext.GetContext(), mgtr.migrationContext.PanicAbort, err)
		}
		mgtr.migrationContext.Log.Debugf("Done streaming")
	}()

	go func() {
		ticker := time.NewTicker(time.Second)
		defer ticker.Stop()
		for range ticker.C {
			if atomic.LoadInt64(&mgtr.finishedMigrating) > 0 {
				return
			}
			mgtr.migrationContext.SetRecentBinlogCoordinates(mgtr.eventsStreamer.GetCurrentBinlogCoordinates())
		}
	}()
	return nil
}

// addDMLEventsListener begins listening for binlog events on the original table,
// and creates & enqueues a write task per such event.