Continuously receiving messages (waiting for messages in the background)

[JustTotallyMe]

Hi :)

sorry for this really simple question but I am quiet new to async and multi-threading.
I am trying to receive messages in the background. I implemented the "Handle_Received_Application_Message" from the samples in my application but I cannot get it to work.

I tried it in an really old version (3.08) with the synchronous approach and that worked but with the "ApplicationMessageReceivedAsync" I am not getting any messages.

To run the task, I tried using "Task newTask = HandleMessage();" after the "InitializeComponent();" and also "Task.Run(() => HandleMessage());" inside a button-click event.

If I debug my code, I see that (when clicking the button), the "HandleMessage()" is called and executed but nothing happens afterwards and I don't think, that there is a task being created with the Handle to react to any incoming messages.

I hope this makes sense and someone can help me :)

Kind regards and thanks in advance,
Andreas :)

[ramonsmits]

@JustTotallyMe I recently started looking at MQTTnet again. In the past, I created the following issue:

• Concurrent processing #812

It sort of works as shown in the following sample PR I just created:

• Samples that show concurrent processing #1803

[JustTotallyMe]

@ramonsmits thx for your reply but don't seem to be understanding your message. Are you saying I could try the workaround from #812 you posted?
The thing is, that I just need to get the messages in the background and fire an event, if a message was received.

[ramonsmits]

@JustTotallyMe I think I misinterpreted your question.

I'm unsure what your question is. Maybe add actual code to make it clear.

[JustTotallyMe]

@ramonsmits Well I don't really have code to show because I can't figure out how I would do something like this.
Here again, what I am trying to do:

I am trying to receive messages with MQTT continuously in the background. If a message comes in, an event should be fired.
I tried this with the "ApplicationMessageReceivedAsync"-Method from the samples but I cannot get it to work because I do not know how to start the Task for this.

[ramonsmits]

I am trying to receive messages with MQTT continuously in the background.

How are you "trying" this? Where are you initializing the client?

If a message comes in, an event should be fired. I tried this with the "ApplicationMessageReceivedAsync"-Method from the samples but I cannot get it to work because I do not know how to start the Task for this.

What do you mean by "how to start the Task for this?"

This shows how to subscribe to the event:

MQTTnet/Samples/Client/Client_Subscribe_Samples.cs

Lines 33 to 39 in 8ad8692

	mqttClient.ApplicationMessageReceivedAsync += e =>
	{
	Console.WriteLine("Received application message.");
	e.DumpToConsole();
	return Task.CompletedTask;
	};

Where did you wire this in your code?

[JustTotallyMe]

oh sorry I copied the wrong code.

This is the method that does the subscription (I took the code from the samples).

private async Task HandleMessage()
        {
            var factory = new MqttFactory();

            using (var mqttClient = factory.CreateMqttClient())
            {
                var mqttClientOptions = new MqttClientOptionsBuilder()
                    .WithTcpServer("127.0.0.1", 1884)
                    .WithCredentials("username", "password")
                    .WithCleanSession()
                    .Build();

                mqttClient.ApplicationMessageReceivedAsync += e =>
                {
                    this.Invoke((MethodInvoker)delegate ()
                    {
                        string output = "";
                        foreach (var item in Encoding.UTF8.GetChars(e.ApplicationMessage.Payload))
                        {
                            output += item;
                        }
                        textBox1.Text = output;
                    });

                    return Task.CompletedTask;
                };

                await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None);

                var mqttSubscribeOptions = factory.CreateSubscribeOptionsBuilder()
                    .WithTopicFilter(
                    f =>
                    {
                        f.WithTopic("mqtt/users");
                    }).Build();

                await mqttClient.SubscribeAsync(mqttSubscribeOptions, CancellationToken.None);
            }
        }

And here I call the method / start the task:

private void button1_Click(object sender, EventArgs e)
        {
            Task.Run(() => HandleMessage());
        }

I also tried to start the task like so:

Task newTask = HandleMessage();

But this does not work either.

[ramonsmits]

@JustTotallyMe Again, don't setup the client in your click handler. Setup the client as a singleton in your application at application start for sending.

For receiving during the lifetime of the application also setup a receiving client at start or if this is specific for only a certain window setup that receiving client for that form in the constructor.

Your click handler should ONLY do sending OR potentially Start or Stop the client.

[JustTotallyMe]

Thank you for the info but does the method and the Task.Run(() => HandleMessage()); look correct or do I call the method differently?
Because either way, I do not receive any messages at the moment (if I call the method at the start or after a button click).

[ramonsmits]

@JustTotallyMe No, it does not. Your code immediately disposes of the client (using).

Your click handler should at most do:

async void button1_Click(object sender, EventArgs e)
{
    var applicationMessage = new MqttApplicationMessageBuilder()
                .WithTopic("samples/temperature/living_room")
                .WithPayload("19.5")
                .Build();

    await mqttSenderClient.PublishAsync(applicationMessage, CancellationToken.None);
}

The other setup stuff like creating a client and registering a event handler/callback should be done in the lifetime of the application or window/form/etc lifetime

[ramonsmits]

If you actually share code of your application I can assist you but if you don't its very hard to help out as it seems you are a beginner c#/.net developer which is not a problem at all but hard to explain concepts.

[JustTotallyMe]

I am sorry for all the confusion with the wrong code an such. But yes you are kind of right, I am not new to C# but I never really used the whole Task-stuff with multi-threading and stuff.

Lets start from the beginning:
First I like to create a windows-forms application, which keeps listening for incomming messages of MQTT in the background.
If a message is received, it should be displayed inside a text-box-item.

Here is the code, that I have so far:

This is the method to subscribe to a topic and setup the message-received handle:

private async Task HandleMessage()
        {
            var factory = new MqttFactory();

            using (var mqttClient = factory.CreateMqttClient())
            {
                var mqttClientOptions = new MqttClientOptionsBuilder()
                    .WithTcpServer("192.168.1.94", 1884)
                    .WithCredentials("openhab", "OH2020!")
                    .WithCleanSession()
                    .Build();

                mqttClient.ApplicationMessageReceivedAsync += e =>
                {
                    this.Invoke((MethodInvoker)delegate ()
                    {
                        string output = "";
                        foreach (var item in Encoding.UTF8.GetChars(e.ApplicationMessage.Payload))
                        {
                            output += item;
                        }
                        textBox1.Text = output;
                    });

                    return Task.CompletedTask;
                };

                await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None);

                var mqttSubscribeOptions = factory.CreateSubscribeOptionsBuilder()
                    .WithTopicFilter(
                    f =>
                    {
                        f.WithTopic("mqtt/users");
                    }).Build();

                await mqttClient.SubscribeAsync(mqttSubscribeOptions, CancellationToken.None);
            }
        }

I use the this.Invoke, to come back from the side-thread and change the value of the text-box-item.

Here in the constructor of my form, I call the method like so, after the form was initialized:

public Form1()
        {
            InitializeComponent();
            Task.Run(() => HandleMessage());
        }

If I start the program, nothing happens, if messages are published to the subscribed topic.
How could I change that of where is my code/thinking wrong?

[JustTotallyMe]

Omg I feel so stupid :D
Thanks for the hint with the using. I remove the using and now it works like a charm.
Thanks again for your help and patience.

[CZEMacLeod]

@JustTotallyMe I know it isn't directly related to the question here, but there is an extension method ConvertPayloadToString available on the ApplicationMessage object which does the conversion to string much more easily.

textBox1.Text = e.ApplicationMessage.ConvertPayloadToString();

Just make sure you have a using MQTTnet; in your file.

I have some examples of converting json, and numeric payloads too (if you are using net6.0 or higher) available at https://github.com/CZEMacLeod/Tasmota-Mosquitto-MQTTnet-Sample/blob/main/MQTT%20Client/Program.cs if it helps at all.
I found the async callbacks a bit confusing too when I first started working with this library.