Messages not published on Disconnect

[IlSocio]

Hi, I created a unit-test to publish a single message:

• connect
• publish
• disconnect and I noticed that sometime the message is not published. It happens because the message queue doesn't get flushed during the disconnect, and if the client gets collected by the garbage-collector, the the messages which are still in the queue will get lost.

[Invisibleman1002]

I created an ASP.NET C# AJAX service. This service publishes a message, but without a thread.sleep in there, it fails every time. Of course, the sleep causes other issues.
Hopefully there is a fix for this. Maybe a connect event, so on connect, i can send my message, then an on message sent event, i can close the connection?

    void PublishMessage(string mcid)
    {
        uPLibrary.Networking.M2Mqtt.MqttClient clientSub;
        clientSub = new  uPLibrary.Networking.M2Mqtt.MqttClient("10.2.3.5");
        clientSub.Connect("WebAppTest_subasd");

        System.Threading.Thread.Sleep(1000);

        clientSub.Publish("ptest/"+mcid, Encoding.UTF8.GetBytes("2"));

        System.Threading.Thread.Sleep(5000);

        clientSub.Disconnect();
    }

[Invisibleman1002]

instead of System.Threading.Thread.Sleep(1000); I used

        while (!clientSub.IsConnected)
        {
            System.Threading.Thread.Sleep(1);
        }

Seems to fix part of the problem. My Publish event currently only fires once, more requests to the Ajax service fires nothing after the first one.

[JTrotta]

Did you try using Connection method override with parameter cleanSession to false? byte result = _client.Connect("uniquename", UserName,Password, false, 500);

In this way when the connection rise up again the queue still have the unset message reay to be sent again.

[mjedrzejas]

@IlSocio it would be nice to have a safe disconnect option. Now in order to be sure that a message was published you need to subscribe for MqttMsgPublished events:

client.MqttMsgPublished += MqttMsgPublished;

Then you can get a msgId from the Publish() method

msgId = client.Publish(topic, Encoding.UTF8.GetBytes(message), qos, retained)

And wait for the confirmation event, you need to compare the msgId. You can use AutoResetEvent for synchronization.

if (e.IsPublished && e.MessageId == msgId) { IsMqttMsgPublished.Set(); }

@Invisibleman1002 it may help you, you could remove the second sleep.

[Invisibleman1002]

Thank you @mjedrzejas. I solved the problem by putting the connect and disconnect inside my global.asax file. Not where I would have preferred to do it, but it worked.

        public static uPLibrary.Networking.M2Mqtt.MqttClient clientSub;

        protected void Application_Start(object sender, EventArgs e)
        {
            Random random = new Random();
            clientSub = new uPLibrary.Networking.M2Mqtt.MqttClient("10.2.3.5");
            clientSub.Connect("WebAppTest_subasd" + random.Next(0, 1000).ToString());
        }
        protected void Application_End(object sender, EventArgs e)
        {
            clientSub.Disconnect();
        }

I also had a page caching problem and I think some of my other methods may have worked without using the Global.asax. This was for a proof of concept that worked enough to get some people interested. I just need to wait for the go ahead to work on it, then I'll revisit some better methods. I did try something like this in a publish event, but it did not work as expected, which again, might be because of my cache problem. That was fixed by adding Response.Expires = -1; to my ajax page. The code below is one thing I tried before moving it into a Global.asax.

    static void client_MqttMsgPublish(object sender, uPLibrary.Networking.M2Mqtt.Messages.MqttMsgPublishedEventArgs e)
    {
        // handle message received 
        clientSub.Disconnect();
    }

[chriswue]

@mjedrzejas One downside is that the published event does not get raised for QoS level 0. OTOH I guess you could argue that if you publish with level 0 you actually don't care whether or not your message made it out the door.

[mjedrzejas]

@chriswue you are absolutely correct with both things. Using QoS 0 you should be more focused on sending next message that verifying if the current one was sent.

[jayathuam]

hi @mjedrzejas, your answer is really helpful. But the "client.MqttMsgPublished" was not working for me. I have done all the steps you mentioned above. But the "MqttMsgPublished" event not fired after a success publish.

[chriswue]

@jayathuam: Which QoS level did you use?

[shpall]

Hi, @IlSocio. Is there any information?

@Invisibleman1002 we added a simple workaround to MqttClient.cs:

        public void DisconnectGracefully()
        {
            var disconnect = new MqttMsgDisconnect {QosLevel = MqttMsgBase.QOS_LEVEL_AT_LEAST_ONCE};
            this.EnqueueInflight(disconnect, MqttMsgFlow.ToPublish);
        }

and use this instead of Disconnect(). Not tested so far, but it seems like this is what you want.

[Invisibleman1002]

@shpall Thank you. That looks like a nice fix. When I have a moment, I'll resurrect the code and give it a try.

Thank you for the update!

-Trey

[dpsenner]

@shpall @Invisibleman1002 Has the "DisconnectGracefully"-fix ever been released?

[shpall]

@shpall @Invisibleman1002 Has the "DisconnectGracefully"-fix ever been released?

No, that is a "workaround"

[dpsenner]

Using the library in a powershell script, we were able to work around this with a finally block and a Start-Sleep command, i.e.:

$mqttclient = [uPLibrary.Networking.M2Mqtt.MqttClient]($broker)
try
{
    $mqttclient.Connect([guid]::NewGuid())
    [...]
}
finally
{
    Start-Sleep -Milliseconds 500
    $mqttclient.Disconnect()
}

[Niko-O]

In case anyone else runs into this problem (and in case I run into it again in the future): This is my quick and dirty workaround: In MqttClient.cs:

        public void Disconnect()
        {
            while (inflightQueue.Count > 0) // Add this loop to wait for pending messages
            {
                System.Threading.Thread.Sleep(1);
            }
            //....

Could be wrong, could be subtly broken, but preliminary testing seems to indicate that it works.