I publish 1000 messages from an MQTT Server when a client subscribes to a topic. Some of the messages are not received by the MQTT Client. Is this a bug or am I implementing something wrong?
Here are the server and client configuration examples.
using MQTTnet;
using MQTTnet.Server;
using System;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
namespace MqttNetServerExample
{
class Program
{
private static IMqttServer _mqttServer;
static void Main(string[] args)
{
// Configure MQTT server.
var optionsBuilder = new MqttServerOptionsBuilder().WithConnectionBacklog(100)
.WithDefaultEndpointPort(1884);
// Define a mqttServer
_mqttServer = new MqttFactory().CreateMqttServer();
// Message arrived configuration
_mqttServer.UseApplicationMessageReceivedHandler(async e =>
{
if (Encoding.UTF8.GetString(e.ApplicationMessage.Payload)== "Test Message")
{
Console.WriteLine("subscription message received");
Console.WriteLine("Simulating messages...");
await Simulate();
}
});
// When a new client connected
_mqttServer.UseClientConnectedHandler(e =>
{
Console.WriteLine("***** CLIENT CONNECTED : " + e.ClientId + " *******");
});
// Start the mqtt server
_mqttServer.StartAsync(optionsBuilder.Build());
Console.ReadLine();
}
private static async Task PublishMessage(string message)
{
// Create mqttMessage
var mqttMessage = new MqttApplicationMessageBuilder()
.WithTopic("mqttServerTopic")
.WithPayload(message)
.WithAtLeastOnceQoS()
.WithRetainFlag(false)
.WithDupFlag(false)
.Build();
// Publish the message asynchronously
var result = await _mqttServer.PublishAsync(mqttMessage, CancellationToken.None);
if(result.ReasonCode == MQTTnet.Client.Publishing.MqttClientPublishReasonCode.Success)
Console.WriteLine("Message published : " + message);
}
private static async Task Simulate()
{
for (int i = 0; i < 1000; i++)
{
var message = "This is a message from server " + i.ToString();
await PublishMessage(message);
}
}
}
}
using MQTTnet;
using MQTTnet.Client;
using MQTTnet.Client.Options;
using MQTTnet.Client.Subscribing;
using System;
using System.Text;
using System.Threading;
namespace MqttNetClientExample
{
class Program
{
private static IMqttClient _mqttClient;
static void Main(string[] args)
{
// Create client
_mqttClient = new MqttFactory().CreateMqttClient();
var options = new MqttClientOptionsBuilder().WithClientId("MqttClient")
.WithTcpServer("localhost", 1884)
.Build();
// When client connected to the server
_mqttClient.UseConnectedHandler(async e =>
{
// Subscribe to a topic
MqttClientSubscribeResult subResult = await _mqttClient.SubscribeAsync(new MqttClientSubscribeOptionsBuilder()
.WithTopicFilter("mqttServerTopic")
.Build());
// Sen a test message to the server
PublishMessage("Test Message");
});
// When client received a message from server
_mqttClient.UseApplicationMessageReceivedHandler(e =>
{
Console.WriteLine($"+ Payload = {Encoding.UTF8.GetString(e.ApplicationMessage.Payload)}");
});
// Connect ot server
_mqttClient.ConnectAsync(options, CancellationToken.None);
Console.Read();
}
private static async void PublishMessage(string message)
{
// Create mqttMessage
var mqttMessage = new MqttApplicationMessageBuilder()
.WithTopic("mqttServerTopic")
.WithPayload(message)
.WithExactlyOnceQoS()
.Build();
// Publish the message asynchronously
await _mqttClient.PublishAsync(mqttMessage, CancellationToken.None);
}
}
}
This is a sample output from client app:
+ Payload = Test Message
+ Payload = This is a message from server 0
+ Payload = This is a message from server 1
+ Payload = This is a message from server 2
+ Payload = This is a message from server 3
+ Payload = This is a message from server 4
+ Payload = This is a message from server 5
+ Payload = This is a message from server 6
+ Payload = This is a message from server 7
+ Payload = This is a message from server 8
+ Payload = This is a message from server 9
+ Payload = This is a message from server 10
+ Payload = This is a message from server 11
+ Payload = This is a message from server 13
+ Payload = This is a message from server 14
+ Payload = This is a message from server 23
+ Payload = This is a message from server 53
+ Payload = This is a message from server 54
+ Payload = This is a message from server 55
+ Payload = This is a message from server 56
+ Payload = This is a message from server 57
+ Payload = This is a message from server 83
+ Payload = This is a message from server 105
+ Payload = This is a message from server 120
+ Payload = This is a message from server 138
+ Payload = This is a message from server 139
+ Payload = This is a message from server 140
+ Payload = This is a message from server 141
+ Payload = This is a message from server 172
+ Payload = This is a message from server 192
+ Payload = This is a message from server 207
+ Payload = This is a message from server 218
+ Payload = This is a message from server 236
+ Payload = This is a message from server 258
+ Payload = This is a message from server 278
+ Payload = This is a message from server 302
.
.
.
.
+ Payload = This is a message from server 999
As you see, some messages are lost. How can I get all messages without any loss? Thanks.
As per the comments the issue was with the configuration of the MqttNet Broker.
The setting MaxPendingMessagesPerClient defaults to 250 meaning that when there are 250 messages awaiting delivery for a specific client, and another message arrives, the first message in the queue will be dropped.
Your application is sending messages faster than the receiver can process them so the queue builds quickly and messages are lost. The last 250 messages arrive without issue because nothing new is being sent (so the broker has no need to drop further messages).
To fix this you can use WithMaxPendingMessagesPerClient e.g.
var optionsBuilder = new MqttServerOptionsBuilder()
.WithMaxPendingMessagesPerClient(1000);
Note: The value used for this setting in production will depend upon your specific requirements. You need to ensure that the subscriber can consume messages at the rate they are produced (otherwise the brokers queue will grow continuously).