Client HandleNext Behavior

[charles-d-burton]

I ran into a bug with your client.HandleNext() function. When I setup a connection like you have in your examples after about a minute client.HandleNext() starts blasting my terminal with EOF errors. According to the function docs if it has an error it should disconnect and go into a reconnect, but that's not the case. The code in question is:

	go func() {
		for {
			if !client.IsConnected() {
				time.Sleep(time.Second)
				tryconnect()
				continue
			}

			err := client.HandleNext()
			if err != nil {
				slog.Error(err.Error())
			}
		}
	}()

After about a minute it hits that if err != nil block and just spins out with EOF

2024/09/12 13:36:40 ERROR EOF
2024/09/12 13:36:40 ERROR EOF
2024/09/12 13:36:40 ERROR EOF
2024/09/12 13:36:40 ERROR EOF
2024/09/12 13:36:40 ERROR EOF
2024/09/12 13:36:40 ERROR EOF
2024/09/12 13:36:40 ERROR EOF
2024/09/12 13:36:40 ERROR EOF
2024/09/12 13:36:40 ERROR EOF
2024/09/12 13:36:40 ERROR EOF
2024/09/12 13:36:40 ERROR EOF
2024/09/12 13:36:40 ERROR EOF
2024/09/12 13:36:40 ERROR EOF

Comments here indicate that it should fail and try to reconnect

natiu-mqtt/client.go

Line 56 in dc6618a

// HandleNext reads from the wire and decodes MQTT packets.

Here's the full code for reference:

func startMQTT(send chan bool, received chan []byte, topics ...string) {

	client := mqtt.NewClient(mqtt.ClientConfig{
		Decoder: mqtt.DecoderNoAlloc{UserBuffer: make([]byte, 1500)},
		OnPub: func(_ mqtt.Header, _ mqtt.VariablesPublish, r io.Reader) error {
			message, _ := io.ReadAll(r)
			if len(message) > 0 {
				select {
				case received <- message:
				default:
					//Ignores message if buffer full
				}
			}
			println("received message")
			return nil
		},
	})

	var varConn mqtt.VariablesConnect
	varConn.SetDefaultMQTT([]byte("meetmon"))
	varConn.Username = []byte("<redacted>")
	varConn.Password = []byte("<redacted>")

	tryconnect := func() error {
		conn, err := net.Dial("tcp", "192.168.1.121:1883")
		if err != nil {
			return err
		}
		ctx, cancel := context.WithTimeout(context.Background(), 4*time.Second)
		defer cancel()
		return client.Connect(ctx, conn, &varConn)
	}

	err := tryconnect()
	if err != nil {
		slog.Error("connect attempt failed")
		panic(err)
	}
	slog.Info("connected to mqtt broker")

	go func() {
		for {
			if !client.IsConnected() {
				time.Sleep(time.Second)
				tryconnect()
				continue
			}

			err := client.HandleNext()
			if err != nil {
				slog.Error(err.Error())
			}
		}
	}()

	slog.Info("setting config state in broker")
	config, err := (&MeetingConfig{}).GetConfig()
	if err != nil {
		panic(err)
	}
	fmt.Println(string(config))

	configPflags, err := mqtt.NewPublishFlags(mqtt.QoS0, false, true)
	if err != nil {
		panic(err)
	}

	vpub := mqtt.VariablesPublish{
		TopicName: []byte("homeassistant/binary_sensor/meeting/config"),
	}

	vpub.PacketIdentifier = randInt16()
	slog.Info("publishing config")
	err = client.PublishPayload(configPflags, vpub, config)
	if err != nil {
		panic(err)
	}

	go func() {
		pflags, err := mqtt.NewPublishFlags(mqtt.QoS0, false, false)
		if err != nil {
			panic(err)
		}
		vpub.TopicName = []byte("homeassistant/binary_sensor/meeting/state")

		for {
			if !client.IsConnected() {
				slog.Info("client is disconnected, sleeping for 1 second and retrying")
				time.Sleep(time.Second)
				continue
			}

			//TODO: Maybe this should be a select
			msg := <-send
			for {
				if msg {
					slog.Info("setting meeting state to ON")
					vpub.PacketIdentifier = randInt16()
					err := client.PublishPayload(pflags, vpub, []byte("ON"))
					if err == nil {
						break //retry until message is sent
					}
				} else {
          //Default to the not in a meeting state
          slog.Info("setting meeting state to OFF")
          err := client.PublishPayload(pflags, vpub, []byte("OFF"))
          if err == nil {
            break
          }
        }
        time.Sleep(time.Second)
			}
		}
	}()
	ctx := context.Background()

	var vsub mqtt.VariablesSubscribe
	vsub.TopicFilters = []mqtt.SubscribeRequest{mqtt.SubscribeRequest{TopicFilter: []byte("homeassistant/binary_sensor/meeting/state"), QoS: mqtt.QoS0}}
	vsub.PacketIdentifier = randInt16()
	err = client.Subscribe(ctx, vsub)
	if err != nil {
		println(err.Error())
		panic(err)
	}
}

[charles-d-burton]

Maybe this behavior is expected? If I add another goroutine that has a ticker to ping the server i stop receiving error messages. I also added a client.Disconnect() if the handler throws an error. That seems to handle the problem of the disconnect not showing up in the system. All of this is an easy enough fix.

[soypat]

io.EOF and net.ErrClosed are special errors. We should always disconnect on receiving them since there is no more data to be expected after receiving them. Changes in #12. Feel free to test :)