fix(mqtt): fix mqtt session config (#3864)

Signed-off-by: Song Gao <disxiaofei@163.com>

Author: Yisaer

internal/io/mqtt/v4client/client.go

@@ -77,7 +77,7 @@ func Provision(ctx api.StreamContext, props map[string]any, onConnect client.Con
 	}
 
 	cli := pahoMqtt.NewClient(opts)
-	return &Client{cli: cli}, nil
+	return &Client{cli: cli, EnableClientSession: c.EnableClientSession}, nil
 }
 
 func (c *Client) Connect(_ api.StreamContext) error {
@@ -136,7 +136,7 @@ func (c *Client) Unsubscribe(_ api.StreamContext, topic string) error {
 }
 
 func (c *Client) Disconnect(_ api.StreamContext) {
-	c.cli.Disconnect(1)
+	c.cli.Disconnect(1000)
 }
 
 func ValidateConfig(ctx api.StreamContext, props map[string]any) (*ConnectionConfig, error) {

internal/io/mqtt/v5client/client.go

@@ -15,6 +15,7 @@
 package v5client
 
 import (
+	"context"
 	"crypto/tls"
 	"errors"
 	"fmt"
@@ -47,14 +48,15 @@ type Client struct {
 }
 
 type ConnectionConfig struct {
-	Server              string `json:"server"`
-	ClientId            string `json:"clientid"`
-	Uname               string `json:"username"`
-	Password            string `json:"password"`
-	EnableClientSession bool   `json:"enableClientSession"`
-	ClientStatePath     string `json:"clientStatePath"`
-	serverUrl           *url.URL
-	tls                 *tls.Config
+	Server                       string `json:"server"`
+	ClientId                     string `json:"clientid"`
+	Uname                        string `json:"username"`
+	Password                     string `json:"password"`
+	EnableClientSession          bool   `json:"enableClientSession"`
+	ClientStatePath              string `json:"clientStatePath"`
+	SessionExpiryIntervalSeconds int    `json:"sessionExpiryIntervalSeconds"`
+	serverUrl                    *url.URL
+	tls                          *tls.Config
 }
 
 func Provision(ctx api.StreamContext, props map[string]any, onConnect client.ConnectHandler, onConnectLost client.ConnectErrorHandler, _ client.ConnectHandler) (*Client, error) {
@@ -79,7 +81,7 @@ func Provision(ctx api.StreamContext, props map[string]any, onConnect client.Con
 		// It is important to set this because otherwise, any queued messages will be lost if the connection drops and
 		// the server will not queue messages while it is down. The specific setting will depend upon your needs
 		// (60 = 1 minute, 3600 = 1 hour, 86400 = one day, 0xFFFFFFFE = 136 years, 0xFFFFFFFF = don't expire)
-		SessionExpiryInterval: 60,
+		SessionExpiryInterval: uint32(cc.SessionExpiryIntervalSeconds),
 		OnConnectionUp: func(cm *autopaho.ConnectionManager, connAck *paho.Connack) {
 			onConnect(ctx)
 		},
@@ -235,7 +237,9 @@ func (c *Client) Unsubscribe(ctx api.StreamContext, topic string) error {
 }
 
 func (c *Client) Disconnect(ctx api.StreamContext) {
-	err := c.cm.Disconnect(ctx)
+	dctx, dcancel := context.WithTimeout(context.Background(), time.Second*5)
+	defer dcancel()
+	err := c.cm.Disconnect(dctx)
 	if err != nil {
 		ctx.GetLogger().Warnf("disconnect error: %s", err)
 	}
@@ -263,7 +267,9 @@ func (c *Client) ParseMsg(ctx api.StreamContext, msg any) ([]byte, map[string]an
 }
 
 func ValidateConfig(ctx api.StreamContext, props map[string]any) (*ConnectionConfig, error) {
-	c := &ConnectionConfig{}
+	c := &ConnectionConfig{
+		SessionExpiryIntervalSeconds: 60,
+	}
 	err := cast.MapToStruct(props, c)
 	if err != nil {
 		return nil, err