paho golang client cannot keep stable connection

[cyclamen]

I use the paho golang client like the codes. The client prints:

2023/09/07 12:36:17 ERROR Connect lost: error=EOF
2023/09/07 12:36:17 INFO Connected
2023/09/07 12:36:27 ERROR Connect lost: error=EOF
2023/09/07 12:36:27 INFO Connected
2023/09/07 12:37:14 ERROR Connect lost: error=EOF
2023/09/07 12:37:14 INFO Connected
2023/09/07 12:37:22 ERROR Connect lost: error=EOF
2023/09/07 12:37:22 INFO Connected
2023/09/07 12:38:58 ERROR Connect lost: error=EOF
2023/09/07 12:38:58 INFO Connected
2023/09/07 12:39:23 ERROR Connect lost: error=EOF
2023/09/07 12:39:23 INFO Connected
2023/09/07 12:39:44 ERROR Connect lost: error=EOF
2023/09/07 12:39:44 INFO Connected
2023/09/07 12:42:52 ERROR Connect lost: error=EOF
2023/09/07 12:42:52 INFO Connected
2023/09/07 12:43:51 ERROR Connect lost: error=EOF
2023/09/07 12:43:51 INFO Connected
2023/09/07 12:43:59 ERROR Connect lost: error=EOF
2023/09/07 12:43:59 INFO Connected
.......

package main

import (
	"fmt"
	"github.com/eclipse/paho.mqtt.golang"
	"log/slog"
	"os"
	"time"
)

var f mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
	fmt.Printf("TOPIC: %s\n", msg.Topic())
	fmt.Printf("MSG: %s\n", msg.Payload())
}

func main() {
	//mqtt.DEBUG = log.New(os.Stdout, "", 0)
	//mqtt.ERROR = log.New(os.Stdout, "", 0)
	opts := mqtt.NewClientOptions().AddBroker("tcp://10.1.0.213:1883").SetClientID("abcdefghi1")
	opts.OnConnect = connectHandler
	opts.OnConnectionLost = connectLostHandler
	//opts.SetDefaultPublishHandler(messagePubHandler)
	opts.SetKeepAlive(2 * time.Second)
	opts.SetDefaultPublishHandler(f)
	opts.SetPingTimeout(1 * time.Second)

	c := mqtt.NewClient(opts)
	if token := c.Connect(); token.Wait() && token.Error() != nil {
		panic(token.Error())
	}

	if token := c.Subscribe("go-mqtt/sample", 0, nil); token.Wait() && token.Error() != nil {
		fmt.Println(token.Error())
		os.Exit(1)
	}

	for i := 0; i < 5; i++ {
		text := fmt.Sprintf("this is msg #%d!", i)
		token := c.Publish("go-mqtt/sample", 0, false, text)
		token.Wait()
	}

	time.Sleep(600 * time.Second)

	if token := c.Unsubscribe("go-mqtt/sample"); token.Wait() && token.Error() != nil {
		fmt.Println(token.Error())
		os.Exit(1)
	}

	c.Disconnect(250)

	time.Sleep(1 * time.Second)
}

var connectHandler mqtt.OnConnectHandler = func(client mqtt.Client) {
	slog.Info("Connected")

}

var connectLostHandler mqtt.ConnectionLostHandler = func(client mqtt.Client, err error) {
	slog.Error("Connect lost:", "error", err)

}

and the server log :

12:35PM INF added hook hook=allow-all-auth
12:35PM INF attached listener address=:1883 id=t1 protocol=tcp
12:35PM INF mochi mqtt starting version=2.3.0
12:35PM INF mochi mqtt server started
12:36PM WRN error=EOF listener=t1
12:36PM WRN error="read tcp 10.1.0.213:1883->10.1.0.73:54389: i/o timeout" listener=t1
12:36PM WRN error="read tcp 10.1.0.213:1883->10.1.0.73:54431: i/o timeout" listener=t1
12:37PM WRN error="read tcp 10.1.0.213:1883->10.1.0.73:54474: i/o timeout" listener=t1
12:37PM WRN error="read tcp 10.1.0.213:1883->10.1.0.73:54643: i/o timeout" listener=t1
12:38PM WRN error="read tcp 10.1.0.213:1883->10.1.0.73:54671: i/o timeout" listener=t1
12:39PM WRN error="read tcp 10.1.0.213:1883->10.1.0.73:55054: i/o timeout" listener=t1
12:39PM WRN error="read tcp 10.1.0.213:1883->10.1.0.73:55139: i/o timeout" listener=t1
12:42PM WRN error="read tcp 10.1.0.213:1883->10.1.0.73:55217: i/o timeout" listener=t1
12:43PM WRN error="read tcp 10.1.0.213:1883->10.1.0.73:55913: i/o timeout" listener=t1
12:43PM WRN error="read tcp 10.1.0.213:1883->10.1.0.73:56168: i/o timeout" listener=t1
12:44PM WRN error="read tcp 10.1.0.213:1883->10.1.0.73:56196: i/o timeout" listener=t1
12:45PM WRN error="read tcp 10.1.0.213:1883->10.1.0.73:56332: i/o timeout" listener=t1
12:45PM WRN error="read tcp 10.1.0.213:1883->10.1.0.73:56539: i/o timeout" listener=t1

so, how to make the client be stable.

[werbenhu]

@cyclamen It's caused by a too short keepalive setting. For more information about keepalive, please refer to MQTT v3 spec 3.1.2.10

opts.SetKeepAlive(60 * time.Second)

[cyclamen]

@werbenhu

It works！Thanks！

[lucasjinreal]

@werbenhu hi, if just set 60s, does it too small?