(entriesChannel chan<- *common.EntryContext)
| 807 | } |
| 808 | |
| 809 | func (b *BinlogReader) sendEntry(entriesChannel chan<- *common.EntryContext) { |
| 810 | isBig := b.entryContext.Entry.IsPartOfBigTx() |
| 811 | coordinate := b.entryContext.Entry.Coordinates.(*common.MySQLCoordinateTx) |
| 812 | if isBig { |
| 813 | newVal := atomic.AddInt32(&b.BigTxCount, 1) |
| 814 | if newVal == 1 { |
| 815 | g.AddBigTxJob() |
| 816 | } |
| 817 | b.logger.Info("send bigtx entry", "gno", coordinate.GNO, |
| 818 | "index", b.entryContext.Entry.Index, "final", b.entryContext.Entry.Final, "count", newVal, |
| 819 | "rows", b.entryContext.Rows) |
| 820 | } |
| 821 | b.logger.Debug("sendEntry", "gno", coordinate.GNO, "events", len(b.entryContext.Entry.Events), |
| 822 | "isBig", isBig, "index", b.entryContext.Entry.Index, "final", b.entryContext.Entry.Final, |
| 823 | "rows", b.entryContext.Rows) |
| 824 | atomic.AddInt64(b.memory, int64(b.entryContext.Entry.Size())) |
| 825 | select { |
| 826 | case <-b.shutdownCh: |
| 827 | return |
| 828 | case entriesChannel <- b.entryContext: |
| 829 | if b.entryContext.Entry.Final { |
| 830 | atomic.AddUint32(&b.extractedTxCount, 1) |
| 831 | } |
| 832 | } |
| 833 | } |
| 834 | |
| 835 | func (b *BinlogReader) loadMapping(sql, currentSchema string, schemasRenameMap map[string]string, |
| 836 | oldSchemaNameToTablesRenameMap map[string]map[string]string, stmt ast.StmtNode, |
no test coverage detected