initBinlogReader creates and connects the reader: we hook up to a MySQL server as a replica Cooperate with `initiateStreaming()` using `e.streamerReadyCh`. Any err will be sent thru the chan.
(binlogCoordinates *common.MySQLCoordinates)
| 805 | // initBinlogReader creates and connects the reader: we hook up to a MySQL server as a replica |
| 806 | // Cooperate with `initiateStreaming()` using `e.streamerReadyCh`. Any err will be sent thru the chan. |
| 807 | func (e *Extractor) initBinlogReader(binlogCoordinates *common.MySQLCoordinates) { |
| 808 | binlogReader, err := binlog.NewBinlogReader(e.execCtx, e.mysqlContext, e.logger.ResetNamed("reader"), e.replicateDoDb, e.sqleContext, e.memory2, e.db, e.targetGtid, e.lowerCaseTableNames, e.ctx) |
| 809 | if err != nil { |
| 810 | e.logger.Error("err at initBinlogReader: NewBinlogReader", "err", err) |
| 811 | e.streamerReadyCh <- err |
| 812 | return |
| 813 | } |
| 814 | |
| 815 | e.binlogReader = binlogReader |
| 816 | |
| 817 | go func() { |
| 818 | err = binlogReader.ConnectBinlogStreamer(*binlogCoordinates) |
| 819 | if err != nil { |
| 820 | e.streamerReadyCh <- err |
| 821 | return |
| 822 | } |
| 823 | e.streamerReadyCh <- nil |
| 824 | }() |
| 825 | } |
| 826 | |
| 827 | func (e *Extractor) selectSqlMode() error { |
| 828 | query := `select @@sql_mode` |
no test coverage detected