Method Run

driver/oracle/extractor/extractor_oracle.go:156–255

Run executes the complete extract logic.

()

// Run executes the complete extract logic.
func (e *ExtractorOracle) Run() {
	var err error

	// PutConfig before WatchNats
	err = e.storeManager.PutConfig(e.subject, e.mysqlContext)
	if err != nil {
		e.onError(common.TaskStateDead, errors.Wrap(err, "PutConfig"))
		return
	}

	e.logger.Info("src watch Nats")
	e.natsAddr, err = e.storeManager.SrcWatchNats(e.subject, e.shutdownCh, func(err error) {
		e.onError(common.TaskStateDead, err)
	})
	if err != nil {
		e.onError(common.TaskStateDead, errors.Wrap(err, "SrcWatchNats"))
		return
	}
	// init nats
	e.logger.Info("initNatsPubClient")
	e.logger.Debug("begin Connect nats server", "NatAddr", e.natsAddr)
	sc, err := gonats.Connect(e.natsAddr)
	if err != nil {
		e.logger.Error("cannot connect nats server", "natsAddr", e.natsAddr, "err", err)
		e.onError(common.TaskStateDead, errors.Wrap(err, "Connect Nats"))
		return
	}
	e.logger.Info("Connect nats server", "natsAddr", e.natsAddr)
	e.natsConn = sc

	_, err = e.natsConn.Subscribe(fmt.Sprintf("%s_control2", e.subject), func(m *gonats.Msg) {
		if m.Data == nil {
			e.onError(common.TaskStateDead, fmt.Errorf("zero-byte control msg"))
			return
		}

		ctrlMsg := &common.ControlMsg{}
		_, err := ctrlMsg.Unmarshal(m.Data)
		if err != nil {
			e.onError(common.TaskStateDead, fmt.Errorf("failed to unmarshal a control msg"))
			return
		}

		switch ctrlMsg.Type {
		case common.ControlMsgError:
			e.onError(common.TaskStateDead, fmt.Errorf("applier error: %v", ctrlMsg.Msg))
			return
		}
	})
	if err != nil {
		e.onError(common.TaskStateDead, errors.Wrap(err, "Subscribe control2"))
		return
	}

	e.initDBConnections()

	startSCN, committedSCN, err := e.calculateSCNPos()
	if err != nil {