()
| 731 | } |
| 732 | |
| 733 | func (e *ExtractorOracle) LoopLogminerRecord() error { |
| 734 | l := e.LogMinerStream |
| 735 | err := l.initLogMiner() |
| 736 | if err != nil { |
| 737 | fmt.Println(err) |
| 738 | return err |
| 739 | } |
| 740 | defer l.stopLogMiner() |
| 741 | |
| 742 | records := make(chan *LogMinerRecord, 100) |
| 743 | defer close(records) |
| 744 | go func() { |
| 745 | t := time.NewTicker(time.Second * 5) |
| 746 | defer func() { |
| 747 | t.Stop() |
| 748 | e.logger.Info("Handler Records goroutine exited") |
| 749 | }() |
| 750 | for !e.shutdown { |
| 751 | select { |
| 752 | case r, ok := <-records: |
| 753 | if !ok { |
| 754 | continue |
| 755 | } |
| 756 | atomic.AddInt64(&e.mysqlContext.DeltaEstimate, 1) |
| 757 | switch r.Operation { |
| 758 | case OperationCodeStart: |
| 759 | l.txCache.startTx(r.TxId(), r.SCN) |
| 760 | case OperationCodeCommit: |
| 761 | l.txCache.commitTx(r.TxId(), r.SCN) |
| 762 | case OperationCodeRollback: |
| 763 | l.txCache.rollbackTx(r.TxId(), r.SCN) |
| 764 | case OperationCodeDDL: |
| 765 | l.txCache.Handler(&LogMinerTx{ |
| 766 | transactionId: "", |
| 767 | oldestUncommittedScn: l.txCache.getOldestUncommittedSCN(), |
| 768 | startScn: r.SCN, |
| 769 | endScn: r.SCN, |
| 770 | records: []*LogMinerRecord{r}, |
| 771 | }) |
| 772 | case OperationCodeInsert, OperationCodeDelete, OperationCodeUpdate: |
| 773 | l.txCache.addTxRecord(r) |
| 774 | } |
| 775 | l.startScn = r.SCN |
| 776 | case <-t.C: |
| 777 | continue |
| 778 | } |
| 779 | } |
| 780 | }() |
| 781 | |
| 782 | for !e.shutdown { |
| 783 | time.Sleep(5 * time.Second) |
| 784 | changed, err := l.checkRedoLogChanged() |
| 785 | if err != nil { |
| 786 | return err |
| 787 | } |
| 788 | if changed { |
| 789 | err := l.stopLogMiner() |
| 790 | if err != nil { |
no test coverage detected