Method NewRunner

driver/handle.go:180–246

(d *Driver)

}

func (h *taskHandle) NewRunner(d *Driver) (runner DriverHandle, err error) {
	ctx := &common.ExecContext{
		Subject:  h.taskConfig.JobName,
		StateDir: d.config.DataDir,
	}

	logger := h.logger
	// create a new log file for debug job
	if d.config.DebugJob == h.taskConfig.JobName && hclog.LevelFromString(d.config.LogLevel) != hclog.Debug {
		logger = setupLogger(d.config.LogFile, "dtle_debug_job.log")
		logger.SetLevel(hclog.Debug)
	}

	switch common.TaskTypeFromString(h.taskConfig.Name) {
	case common.TaskTypeSrc:
		if h.driverConfig.SrcOracleConfig != nil {
			h.logger.Debug("found oracle src", "SrcOracleConfig", h.driverConfig.SrcOracleConfig)
			runner, err = extractor.NewExtractorOracle(ctx, h.driverConfig, logger, d.storeManager, h.waitCh, h.ctx)
			if err != nil {
				return nil, errors.Wrap(err, "NewExtractor")
			}
		} else {
			e, err := mysql.NewExtractor(ctx, h.driverConfig, logger, d.storeManager, h.waitCh, h.ctx)
			if err != nil {
				return nil, errors.Wrap(err, "NewOracleExtractor")
			}
			runner = e
			if h.driverConfig.TwoWaySync {
				ctx2 := &common.ExecContext{
					Subject:  ctx.Subject + "_dtrev",
					StateDir: d.config.DataDir,
				}
				cfg2 := &common.MySQLDriverConfig{
					DtleTaskConfig: common.DtleTaskConfig{
						DestType: "mysql",
					},
				}
				e.RevApplier, err = mysql.NewApplier(ctx2, cfg2, logger, d.storeManager, d.config.NatsAdvertise,
					h.waitCh, d.eventer, h.taskConfig, h.ctx)
			}

		}
	case common.TaskTypeDest:
		h.logger.Debug("found dest", "allConfig", h.driverConfig)
		switch strings.ToLower(h.driverConfig.DestType) {
		case "kafka":
			runner, err = kafka.NewKafkaRunner(ctx, logger,
				d.storeManager, d.config.NatsAdvertise, h.waitCh, h.ctx)
			if err != nil {
				return nil, errors.Wrap(err, "NewKafkaRunner")
			}
		case "mysql":
			runner, err = mysql.NewApplier(ctx, h.driverConfig, logger, d.storeManager,
				d.config.NatsAdvertise, h.waitCh, d.eventer, h.taskConfig, h.ctx)
			if err != nil {
				return nil, errors.Wrap(err, "NewApplier")
			}
		case "":