mqttagent

commit 2e9d5a773fce033a3d2ac7e7b173c057e0d15a4f
parent 102e0ae37e6c06cf2827c22d96f99bd6c910c344
Author: Natasha Kerensikova <natgh@instinctive.eu>
Date:   Sat,  4 Jan 2025 14:56:16 +0000

MQTT connection options are widened
Diffstat:
M
go.mod
 | 
2
++
M
mqttagent.go
 | 
72
+++++++++++++++++++++++++++++++++++++++++++++++++++++++++++-------------

2 files changed, 61 insertions(+), 13 deletions(-)
diff --git a/go.mod b/go.mod
@@ -6,6 +6,7 @@ require (
 	github.com/glebarez/go-sqlite v1.22.0
 	github.com/go-mqtt/mqtt v0.0.0-20210702165922-b33ea0451b0b
 	github.com/layeh/gopher-json v0.0.0-20201124131017-552bb3c4c3bf
+	github.com/yuin/gluamapper v0.0.0-20150323120927-d836955830e7
 	github.com/yuin/gopher-lua v1.1.1
 )
 
@@ -13,6 +14,7 @@ require (
 	github.com/dustin/go-humanize v1.0.1 // indirect
 	github.com/google/uuid v1.5.0 // indirect
 	github.com/mattn/go-isatty v0.0.20 // indirect
+	github.com/mitchellh/mapstructure v1.5.0 // indirect
 	github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect
 	golang.org/x/sys v0.15.0 // indirect
 	modernc.org/libc v1.37.6 // indirect
diff --git a/mqttagent.go b/mqttagent.go
@@ -25,6 +25,7 @@ import (
 	"time"
 
 	"github.com/go-mqtt/mqtt"
+	"github.com/yuin/gluamapper"
 	"github.com/yuin/gopher-lua"
 )
 
@@ -134,22 +135,67 @@ func stateCnxTable(L *lua.LState) *lua.LTable {
 	return stateValue(L, keyCnxTable).(*lua.LTable)
 }
 
+type mqttConfig struct {
+	Connection     string
+	PauseTimeout   string
+	AtLeastOnceMax int
+	ExactlyOnceMax int
+	UserName       string
+	Password       []byte
+	Will           struct {
+		Topic       string
+		Message     []byte
+		Retain      bool
+		AtLeastOnce bool
+		ExactlyOnce bool
+	}
+	KeepAlive    uint16
+	CleanSession bool
+}
+
+func newClient(L *lua.LState, id string) (*mqtt.Client, error) {
+	var config mqttConfig
+	if err := gluamapper.Map(L.CheckTable(1), &config); err != nil {
+		return nil, err
+	}
+
+	pto, err := time.ParseDuration(config.PauseTimeout)
+	if err != nil {
+		pto = time.Second
+	}
+
+	processed_cfg := mqtt.Config{
+		Dialer:         mqtt.NewDialer("tcp", config.Connection),
+		PauseTimeout:   pto,
+		AtLeastOnceMax: config.AtLeastOnceMax,
+		ExactlyOnceMax: config.ExactlyOnceMax,
+		UserName:       config.UserName,
+		Password:       config.Password,
+		Will: struct {
+			Topic       string
+			Message     []byte
+			Retain      bool
+			AtLeastOnce bool
+			ExactlyOnce bool
+		}{
+			Topic:       config.Will.Topic,
+			Message:     config.Will.Message,
+			Retain:      config.Will.Retain,
+			AtLeastOnce: config.Will.AtLeastOnce,
+			ExactlyOnce: config.Will.ExactlyOnce,
+		},
+		KeepAlive:    config.KeepAlive,
+		CleanSession: config.CleanSession,
+	}
+
+	return mqtt.VolatileSession(id, &processed_cfg)
+}
+
 func newMqttClient(L *lua.LState) int {
-	server := L.CheckString(1)
-	user := L.CheckString(2)
-	pass := L.CheckString(3)
-	to := time.Duration(L.OptNumber(4, lua.LNumber(1.0))) * time.Second
-	mt := L.GetTypeMetatable(luaMqttClientTypeName)
 	id := stateCnxTable(L).Len() + 1
-
 	idString := fmt.Sprintf("%s-%d", stateClientPrefix(L), id)
 
-	client, err := mqtt.VolatileSession(idString, &mqtt.Config{
-		Dialer:       mqtt.NewDialer("tcp", server),
-		PauseTimeout: to,
-		UserName:     user,
-		Password:     []byte(pass),
-	})
+	client, err := newClient(L, idString)
 	if err != nil {
 		log.Println(err)
 		L.Push(lua.LNil)
@@ -164,7 +210,7 @@ func newMqttClient(L *lua.LState) int {
 	res := L.NewTable()
 	L.RawSetInt(res, keyClient, ud)
 	L.RawSetInt(res, keySubTable, L.NewTable())
-	L.SetMetatable(res, mt)
+	L.SetMetatable(res, L.GetTypeMetatable(luaMqttClientTypeName))
 	L.RawSetInt(stateCnxTable(L), id, res)
 	L.Push(res)
 	return 1