dotnet / dotnet/MQTTnet

Client Missing Received Messages

Open
#1,862 1 comment 0 reactions 0 assignees View on GitHub
bug
Dominant language
C#
Stars
5.1k
Forks
1.2k
PR merge metrics
No merged PRs in 30d

Description

### Describe the bug
It appears we are sometimes missing messages in high volume. Using the `ManagedMqttClient` I'm subscribing to a topic, sending a message to that topic, reading that message back from the subscription (or in the cases where we miss messages I'm waiting 10 seconds and then continuing) and then unsubscribing. I'm doing this in a loop for 200 messages and every time it misses at least 1 message, sometimes even more. When I switch over to the regular MqttClient, this does not occur. I've verified that this happens (maybe even worse) in release mode. I've verified that the messages are actually coming through using mosquitto_sub but the `ApplicationMessageReceivedAsync` is just not firing off using ManagedMqttClient for these few messages. I am also using the latest version, 4.3.1.873.

### Which component is your bug related to?
- ManagedClient

### To Reproduce
Steps to reproduce the behavior:
1. Have a broker setup
2. Update the code attached to use your broker, topic, and message
3. Run the code
4. Check the number of messages received at the end, if it is not 200 then a message was missed

### Expected behavior
All 200 response messages should be seen and handled

### Screenshots
![image](https://github.com/dotnet/MQTTnet/assets/28354400/f00ed438-61ce-43cf-82fe-b9e727ee0c9c)

### Code example
[MqttNetMissingMessagesExample.zip](https://github.com/dotnet/MQTTnet/files/13056412/MqttNetMissingMessagesExample.zip)

```csharp
using MQTTnet;
using MQTTnet.Client;
using MQTTnet.Extensions.ManagedClient;
using MQTTnet.Protocol;
using System.Collections.Concurrent;

internal class Program
{
private static readonly ConcurrentDictionary> _DeviceMessageDict = new();

private static async Task WaitForMessage(string topic, DateTime earliestMessageTime, TimeSpan timeout, TimeSpan timeBetweenChecks)
{
if (string.IsNullOrWhiteSpace(topic))
{
throw new ArgumentNullException(nameof(topic));
}

var startTime = DateTime.UtcNow;
while (DateTime.UtcNow - startTime < timeout)
{
if (_DeviceMessageDict.TryGetValue(topic, out var messages))
{
var messageInfo =
messages
.Where(messageInfo => messageInfo.timeOfMessage > earliestMessageTime)
.OrderByDescending(messageInfo => messageInfo.sequence)
.Select(messageInfo => new { Message = messageInfo.message })
.FirstOrDefault();

if (messageInfo is not null)
{
var message = messageInfo.Message;
_DeviceMessageDict.TryUpdate(topic, messages.Where(m => m.message != message).ToList(), messages);
return message;
}
}

await Task.Delay(timeBetweenChecks).ConfigureAwait(false);
}
return $"No message received from {topic} after timeout of {timeout}";
}

private static async Task Main()
{
var mqttFactory = new MqttFactory();

using (var mqttClient = mqttFactory.CreateManagedMqttClient())
{
var mqttClientOptions = new MqttClientOptionsBuilder()
.WithTcpServer("YOUR TCP SERVER")
.WithWillQualityOfServiceLevel(MqttQualityOfServiceLevel.AtMostOnce)
.WithWillRetain(false)
.WithCredentials("YOUR USERNAME", "YOUR PASSWORD")
.Build();

var managedMqttClientOptions = new ManagedMqttClientOptionsBuilder()
.WithAutoReconnectDelay(TimeSpan.FromSeconds(5))
.WithClientOptions(mqttClientOptions)
.Build();

await mqttClient.StartAsync(managedMqttClientOptions);

int counter = 0;
mqttClient.ApplicationMessageReceivedAsync += e =>
{
var count = Interlocked.Increment(ref counter);
var message = e.ApplicationMessage.ConvertPayloadToString();
var topic = e.ApplicationMessage.Topic;

Console.WriteLine($"Received {count} message {message} on Topic {topic}");

_DeviceMessageDict.AddOrUpdate(
topic,
new List<(string, int, DateTime)>() { (message, 1, DateTime.UtcNow) },
(k, v) => v.Append((message, v.Count() + 1, DateTime.UtcNow))
);

return Task.CompletedTask;
};

for (int i = 0; i < 200; i++)
{
var topic = "YOUR TOPIC";
var filter = new[]
{
new MqttTopicFilterBuilder().WithTopic(topic).Build()
};
await mqttClient.SubscribeAsync(filter).ConfigureAwait(false);

var startTime = DateTime.Now;
var message = "YOUR MESSAGE";
await mqttClient.EnqueueAsync(topic, message);
Console.WriteLine($"{message} sent to topic {topic}");

var result = await WaitForMessage(topic, startTime, TimeSpan.FromSeconds(10), TimeSpan.FromMilliseconds(100));
Console.WriteLine($"Result: {result}");
await mqttClient.UnsubscribeAsync(topic).ConfigureAwait(false);
}

Console.WriteLine($"Total messages: {counter}");
}
}
}
```

Contributor guide

No contributing guide indexed for this repository

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.