Merge branch 'feat-serf'

zhangqian

2023-10-19 eba4eb850f0ecfb5839395aa125955ceaa2a454f

 nsq/nsq.go

@@ -3,66 +3,85 @@
import (
   "apsClient/conf"
   "apsClient/constvar"
   "apsClient/model/common"
   "apsClient/pkg/logx"
   "apsClient/pkg/nsqclient"
   "apsClient/pkg/safe"
   "context"
   "errors"
   "fmt"
   "time"
   "sync"
)

func Init() error {
type consumerManager struct {
   ctx     context.Context
   clients sync.Map
}

var defaultConsumerManager *consumerManager

func init() {
   defaultConsumerManager = new(consumerManager)
}

func (c *consumerManager) init() error {
   if len(conf.Conf.NsqConf.NodeId) <= 0 {
      return errors.New("no NodeId")
   }

   if err := initProducer(); err != nil {
      return err
   }

   safe.Go(func() {
      caller := NewCaller(fmt.Sprintf(constvar.NsqTopicGetPlcAddress, conf.Conf.NsqConf.NodeId), fmt.Sprintf(constvar.NsqTopicSendPlcAddress, conf.Conf.NsqConf.NodeId))
      var addressResult common.ResponsePlcAddress
      err := caller.Call(common.RequestPlcAddress{DeviceId: conf.Conf.System.DeviceId}, &addressResult, time.Second*3)
      if err != nil {
         logx.Infof("get plc address err: %v", err.Error())
      }
   })

   safe.Go(func() {
      err := Consume(fmt.Sprintf(constvar.NsqTopicScheduleTask, conf.Conf.NsqConf.NodeId), conf.Conf.System.DeviceId)
      if err != nil {
         logx.Errorf("start nsq consume err: %v", err)
      }
   })

   safe.Go(func() {
      err := Consume(fmt.Sprintf(constvar.NsqTopicSendPlcAddress, conf.Conf.NsqConf.NodeId), conf.Conf.System.DeviceId)
      if err != nil {
         logx.Errorf("start nsq consume err: %v", err)
      }
   })

   safe.Go(func() {
      err := Consume(fmt.Sprintf(constvar.NsqTopicProcessParamsResponse, conf.Conf.NsqConf.NodeId), conf.Conf.System.DeviceId)
      if err != nil {
         logx.Errorf("start nsq consume err: %v", err)
      }
   })

   safe.Go(func() {
      err := Consume(fmt.Sprintf(constvar.NsqTopicApsProcessParams, conf.Conf.NsqConf.NodeId), conf.Conf.System.DeviceId)
      if err != nil {
         logx.Errorf("start nsq consume err: %v", err)
      }
   })

   safe.Go(func() {
      err := Consume(fmt.Sprintf(constvar.NsqTopicDeviceUpdate, conf.Conf.NsqConf.NodeId), conf.Conf.System.DeviceId)
      if err != nil {
         logx.Errorf("start nsq consume err: %v", err)
      }
   })

   var topics = []string{
      constvar.NsqTopicScheduleTask,
      constvar.NsqTopicSendPlcAddress,
      constvar.NsqTopicProcessParamsResponse,
      constvar.NsqTopicApsProcessParams,
      constvar.NsqTopicDeviceUpdate,
      constvar.NsqTopicPullDataResponse,
   }
   for _, t := range topics {
      topic := fmt.Sprintf(t, conf.Conf.NsqConf.NodeId)
      c.AddConsumer(topic)
   }
   return nil
}

func (c *consumerManager) AddConsumer(topic string) {
   safe.Go(func() {
      client, err := NewConsumer(topic, conf.Conf.System.DeviceId)
      if err != nil {
         logx.Errorf("start nsq consume err: %v", err)
      }
      c.clients.Store(topic, client)
      if len(conf.Conf.NsqConf.NsqlookupdAddr) > 0 {
         if err = client.RunLookupd(conf.Conf.NsqConf.NsqlookupdAddr, 1); err != nil {
            logx.Errorf("RunLookupd err:%v", err)
            return
         }
      } else {
         if err = client.Run(conf.Conf.NsqConf.NsqdAddr, 1); err != nil {
            logx.Errorf("Run err:%v", err)
            return
         }
      }
      logx.Infof("add consumer success, topic:%v", topic)
   })
}

func (c *consumerManager) stop() {
   c.clients.Range(func(key, value any) bool {
      if consumer, ok := value.(*nsqclient.NsqConsumer); ok {
         nsqclient.DestroyNsqConsumer(consumer)
         logx.Infof("try stop consumer, topic : %v", key)
      }
      return true
   })
}

func Init() error {
   return defaultConsumerManager.init()
}

func Stop() {
   defaultConsumerManager.stop()
}