(tx *LogMinerTx)
| 579 | } |
| 580 | |
| 581 | func (e *ExtractorOracle) handleSQLs(tx *LogMinerTx) *common.DataEntry { |
| 582 | entry := common.NewBinlogEntry() |
| 583 | entry.Final = true |
| 584 | oracleCoordinateTx := &common.OracleCoordinateTx{ |
| 585 | OldestUncommittedScn: tx.oldestUncommittedScn, |
| 586 | EndSCN: tx.endScn, |
| 587 | } |
| 588 | entry.Coordinates = oracleCoordinateTx |
| 589 | for _, row := range tx.records { |
| 590 | dataEvent, err := e.parseToDataEvent(row) |
| 591 | if err != nil { |
| 592 | e.logger.Error("parseOracleToMySQL", "err", err) |
| 593 | continue |
| 594 | } |
| 595 | times, err := time.Parse(time.RFC3339, row.StartTime) |
| 596 | if err != nil { |
| 597 | e.logger.Error("parse timestamp", "err", err) |
| 598 | } |
| 599 | dataEvent.Timestamp = uint32(times.Unix()) |
| 600 | entry.Events = append(entry.Events, dataEvent) |
| 601 | } |
| 602 | return entry |
| 603 | } |
| 604 | |
| 605 | type LogMinerStream struct { |
| 606 | ctx context.Context |
no test coverage detected