Method initNatsPubClient

driver/mysql/extractor.go:616–695

(natsAddr string)

}

func (e *Extractor) initNatsPubClient(natsAddr string) (err error) {
	e.logger.Debug("begin Connect nats server", "NatAddr", natsAddr)
	sc, err := gonats.Connect(natsAddr)
	if err != nil {
		e.logger.Error("cannot connect nats server", "natsAddr", natsAddr, "err", err)
		return err
	}
	e.logger.Info("Connect nats server", "natsAddr", 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
	}

	_, err = e.natsConn.Subscribe(fmt.Sprintf("%s_bigtx_ack", e.subject), func(m *gonats.Msg) {
		err := e.natsConn.Publish(m.Reply, nil)
		if err != nil {
			e.onError(common.TaskStateDead, errors.Wrap(err, "bigtx_ack. reply"))
			return
		}

		ack := &common.BigTxAck{}
		_, err = ack.Unmarshal(m.Data)
		if err != nil {
			e.onError(common.TaskStateDead, errors.Wrap(err, "bigtx_ack. Unmarshal"))
		}
		e.logger.Info("bigtx_ack", "gno", ack.GNO, "index", ack.Index,
			"count", atomic.LoadInt32(&e.binlogReader.BigTxCount))

		if !e.shutdown {
			newVal := atomic.AddInt32(&e.binlogReader.BigTxCount, -1)
			if newVal == 0 {
				g.SubBigTxJob()
			}
			if newVal < 0 {
				e.onError(common.TaskStateDead, fmt.Errorf("DTLE_BUG: BigTxCount is less than 0. %v", newVal))
			}
		}