From eba4eb850f0ecfb5839395aa125955ceaa2a454f Mon Sep 17 00:00:00 2001
From: zhangqian <zhangqian@123.com>
Date: 星期四, 19 十月 2023 16:39:07 +0800
Subject: [PATCH] Merge branch 'feat-serf'

---
 nsq/nsq.go |  115 +++++++++++++++++++++++++++++++++------------------------
 1 files changed, 67 insertions(+), 48 deletions(-)

diff --git a/nsq/nsq.go b/nsq/nsq.go
index 0c293db..888cf08 100644
--- a/nsq/nsq.go
+++ b/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()
+}

--
Gitblit v1.8.0