添加参数

mqtt_binding/mqtt_client/mqtt_client.go

@@ -19,29 +19,26 @@ type MqttClientOptions struct {
 	PingTimeoutSec        int64
 	WriteTimeoutSec       int64
 	SubscribeRoutineCount int
+	SubscribeBufferSize   int
 }
 
-func (opt *MqttClientOptions) check() error {
-	if utils.IsStringEmpty(opt.UserName) {
+func (opts *MqttClientOptions) check() error {
+	if utils.IsStringEmpty(opts.UserName) {
 		return errors.New("必须传递用户名")
 	}
 
-	if utils.IsStringEmpty(opt.Password) {
+	if utils.IsStringEmpty(opts.Password) {
 		return errors.New("必须传递密码")
 	}
 
-	if utils.IsStringEmpty(opt.Address) {
+	if utils.IsStringEmpty(opts.Address) {
 		return errors.New("必须传递地址")
 	}
 
-	if utils.IsStringEmpty(opt.ClientID) {
+	if utils.IsStringEmpty(opts.ClientID) {
 		return errors.New("必须传递客户端ID")
 	}
 
-	if opt.SubscribeRoutineCount == 0 {
-		opt.SubscribeRoutineCount = 10
-	}
-
 	return nil
 }
 
@@ -70,10 +67,18 @@ func NewMqttClient(opts *MqttClientOptions) (*MqttClient, error) {
 		return nil, err
 	}
 
+	if opts.SubscribeRoutineCount == 0 {
+		opts.SubscribeRoutineCount = 10
+	}
+
+	if opts.SubscribeBufferSize == 0 {
+		opts.SubscribeRoutineCount = 1024
+	}
+
 	mqttClient := &MqttClient{
 		routersMutex:      &sync.Mutex{},
 		routers:           make([]*router.Router, 0),
-		subscribeChan:     make(chan *subscribePayload, 1024),
+		subscribeChan:     make(chan *subscribePayload, opts.SubscribeRoutineCount),
 		subscribeDoneChan: make(chan any),
 	}