I have a Paho MQTT GO client that listens to the broker. After around 2400-2500 messages, it stops listening to the new messages. Interestingly, it does not get disconnected with the broker. Seems that it is still active and listening but new messages does not appear anymore.

Below is my code -

package main

import (
    "flag"
    "fmt"
    "log"
    "os"
    "os/signal"
    "syscall"

    mqtt "github.com/eclipse/paho.mqtt.golang"
)

var messagePubHandler mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
    fmt.Printf("Received message: %s from topic: %s\n", msg.Payload(), msg.Topic())
}

var connectHandler mqtt.OnConnectHandler = func(client mqtt.Client) {
    fmt.Println("Connected")
}

var connectLostHandler mqtt.ConnectionLostHandler = func(client mqtt.Client, err error) {
    fmt.Printf("Connect lost: %v", err)
}


func createClient() mqtt.Client {
    var broker = "*******"
    var port = 1883
    opts := mqtt.NewClientOptions()
    opts.AddBroker(fmt.Sprintf("tcp://%s:%d", broker, port))
    opts.SetClientID("go_mqtt_client_test1")
    opts.SetDefaultPublishHandler(messagePubHandler)
    opts.OnConnect = connectHandler
    opts.OnConnectionLost = connectLostHandler
    return mqtt.NewClient(opts)
}

func main() {

    var client = createClient()
    if token := client.Connect(); token.Wait() && token.Error() != nil {
        panic(token.Error())
    }

    // load command line arguments if any
    name := flag.String("name", "world", "name to print")
    flag.Parse()

    log.Printf("Starting sleepservice for %s", *name)

    // setup signal catching
    sigs := make(chan os.Signal, 1)

    // catch all signals since not explicitly listing
    //signal.Notify(sigs)
    signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM, syscall.SIGQUIT)

    // method invoked upon seeing signal
    go func() {
        s := <-sigs
        log.Printf("RECEIVED SIGNAL: %s", s)

        switch s {
        case syscall.SIGINT:
            AppCleanup(client)
            os.Exit(1)
        case syscall.SIGTERM:
            AppCleanup(client)
            os.Exit(1)
        case syscall.SIGQUIT:
            AppCleanup(client)
            os.Exit(1)
        default:
            log.Printf("not supported Signal")
        }

    }()

    sub(client)

    for { /* Endless Loop */
    }

}

func AppCleanup(client mqtt.Client) {
    client.Disconnect(250)
    log.Println("CLEANUP APP BEFORE EXIT!!!")
}

func sub(client mqtt.Client) {
    topic := "test/topic"
    token := client.Subscribe(topic, 1, nil)
    token.Wait()
    fmt.Printf("Subscribed to topic: %s", topic)
}

here, I have hidden the IP of broker

your help is much appreciated. Thanks in advance.

CodePudding user response：

The code looks fine.

Are you connecting with the same client ID in some other place?

Can you try replacing the endless for loop with a latch to avoid spinning cpu?

CodePudding user response：

Finally I found the problem, It was that last endless for loop which was causing heck. I removed it and now problem seems to be solved. Thanks @asad and @Brits