Method handleEvent

driver/mysql/binlog/binlog_reader.go:450–496

(ev *replication.BinlogEvent, entriesChannel chan<- *common.EntryContext)

}

func (b *BinlogReader) handleEvent(ev *replication.BinlogEvent, entriesChannel chan<- *common.EntryContext) error {
	switch ev.Header.EventType {
	case replication.GTID_EVENT:
		evt := ev.Event.(*replication.GTIDEvent)
		b.currentCoordMutex.Lock()
		// TODO this does not unlock until function return. wrap with func() if needed
		defer b.currentCoordMutex.Unlock()
		u, _ := uuid.FromBytes(evt.SID)

		entry := common.NewBinlogEntry()
		entry.Coordinates = &common.MySQLCoordinateTx{
			LogFile:       b.currentCoord.LogFile,
			LogPos:        int64(ev.Header.LogPos),
			SID:           u,
			GNO:           evt.GNO,
			LastCommitted: evt.LastCommitted,
			SeqenceNumber: evt.SequenceNumber,
		}
		entry.Index = 0
		entry.Final = true

		b.hasBeginQuery = false
		b.entryContext = &common.EntryContext{
			Entry:        entry,
			TableItems:   nil,
			OriginalSize: 1, // GroupMaxSize is default to 1 and we send on EntriesSize >= GroupMaxSize
			Rows:         0,
		}
	case replication.QUERY_EVENT:
		return b.handleQueryEvent(ev, entriesChannel)
	case replication.XID_EVENT:
		evt := ev.Event.(*replication.XIDEvent)
		b.currentCoord.LogPos = int64(ev.Header.LogPos)
		// TODO is the pos the start or the end of a event?
		// pos if which event should be use? Do we need +1?
		mysqlCoordinates := b.entryContext.Entry.Coordinates.(*common.MySQLCoordinateTx)
		mysqlCoordinates.LogPos = b.currentCoord.LogPos
		b.sendEntry(entriesChannel)

		b.handleEventGSet(evt.GSet)
	default:
		if rowsEvent, ok := ev.Event.(*replication.RowsEvent); ok {
			return b.handleRowsEvent(ev, rowsEvent, entriesChannel)
		}
	}
	return nil
}

func queryIsBegin(query string) bool {
	return len(query) == 5 && strings.ToUpper(query) == "BEGIN"