(execCtx *common.ExecContext, cfg *common.MySQLDriverConfig, logger g.LoggerType, storeManager *common.StoreManager, waitCh chan *drivers.ExitResult, ctx context.Context)
| 122 | } |
| 123 | |
| 124 | func NewExtractorOracle(execCtx *common.ExecContext, cfg *common.MySQLDriverConfig, logger g.LoggerType, storeManager *common.StoreManager, waitCh chan *drivers.ExitResult, ctx context.Context) (*ExtractorOracle, error) { |
| 125 | logger.Info("NewExtractorOracle", "job", execCtx.Subject) |
| 126 | |
| 127 | e := &ExtractorOracle{ |
| 128 | ctx: ctx, |
| 129 | logger: logger.Named("ExtractorOracle").With("job", execCtx.Subject), |
| 130 | execCtx: execCtx, |
| 131 | subject: execCtx.Subject, |
| 132 | mysqlContext: cfg, |
| 133 | //rowCopyComplete: make(chan bool), |
| 134 | waitCh: waitCh, |
| 135 | shutdownCh: make(chan struct{}), |
| 136 | testStub1Delay: 0, |
| 137 | context: sqle.NewContext(nil), |
| 138 | gotCoordinateCh: make(chan struct{}), |
| 139 | streamerReadyCh: make(chan error), |
| 140 | fullCopyDone: make(chan struct{}), |
| 141 | storeManager: storeManager, |
| 142 | memory1: new(int64), |
| 143 | memory2: new(int64), |
| 144 | OracleContext: new(OracleContext), |
| 145 | } |
| 146 | e.dataChannel = make(chan *common.EntryContext, cfg.ReplChanBufferSize*4) |
| 147 | e.timestampCtx = NewTimestampContext(e.shutdownCh, e.logger, func() bool { |
| 148 | return len(e.dataChannel) == 0 |
| 149 | // TODO need a more reliable method to determine queue.empty. |
| 150 | }) |
| 151 | |
| 152 | return e, nil |
| 153 | } |
| 154 | |
| 155 | // Run executes the complete extract logic. |
| 156 | func (e *ExtractorOracle) Run() { |
no test coverage detected