Hi! Would you be interested in a contribution to enable more fluent MQTT message handling?

Eg:

client.On["{room}/sensors/{sensor}"] = x =>
{
	Log("Data received from {0} (in {1}): {2}", x.sensor, x.room, x.Message);
};

I built something like this years ago in a library I named Charlotte, but neither it nor the MQTT library it depends on supports .Net Core. The API is inspired by Nancy. I'd be happy to contribute and maintain such an API for this library.

Not sure if I'm doing something wrong, but the Namespace MQTTnet disappears when I upgrade the NuGet from v2.14 to v2.15?

MQTTLens is a popular testing / debugging tool for mqtt traffic. When I test it with MQTTnet 2.1.5 the exception at https://github.com/chkr1011/MQTTnet/blob/0172606747e7e3aecca999a798f837c02b74356d/MQTTnet.Core/Serializer/DefaultMqttV311PacketSerializer.cs#L263 is triggered.

I really like the async API of the MQTT client :)

Any strong named assemblies planned?

I've been trying to find simple way to subscribe to few topics, but there's no example shared on wiki.

I've setup ApplicationMessageReceived but it's not getting any messages. I guess I need to subscribe to a topic, but there I'm stuck :/

Any pointers how to do it?

Hi again,

is there a way to send the data via MQTTS (Port 8883?).
It would be gread to send encrypted messages.
Thanks.

Best regards,
Zuendelmeister

In my app I have to instantiate more than one client. Issue is, as long the MqttNetTrace is a static class, I’m not able to differentiate messages between the different client or server instances in my app. I defined an interface based on the MqttNetTrace and added the interface as property in the MqttServerOptions / MqttClientOptions and reimplement the MqttNetTrace based in the Interface but not static.
I’m not sure how I can share the changes, but from my view that helps to go away from a predefined static trace model.

Regards
Hans

Hello,
The MQTT protocol has 3 publish type. The last two (whose name are in the object), can inform the client that the message has been received with 100% of reliability.
Would be nice to have this information published, i.e. when using ::
public static Task PublishAsync(this IMqttClient client, params MqttApplicationMessage[] applicationMessages);

You should return some object result with SENT or not SENT information

JJ

The exception at https://github.com/chkr1011/MQTTnet/blob/0172606747e7e3aecca999a798f837c02b74356d/MQTTnet.Core/Server/MqttClientSession.cs#L175 is always thrown even though the different QoS levels are handled

For server to client data flow MQTT is anyhow a good choice.
But how to handle client =>server commands ?
When I have already a MQTT communication, why should I use another link for commands ?
Commands can/should have a response and in C# I would like to write such functions RPC-like.

I am just now playing with a prototype of MqttRpc. What do you think ?

By the way what is the essential difference between SignalR and the MQTT AspNet Core integration ?

Good morning,

I've been planning to use this library in one of my projects, but I have noticed an omission: SSL/TLS support on port 8883. See this for information.

Is it possible to add this feature to the library? Cheers!

Hi it's me again,

I wanted to start my program (.NET Core) on a raspberry pi and got the following error:

Unhandled Exception: System.IO.FileNotFoundException: Could not load file or assembly 'Windows.Foundation.UniversalApiContract, Version=3.0.0.0, Cultre=neutral, PublicKeyToken=null, ContentType=WindowsRuntime'. The system cannot find the file specified.
at MQTTnet.Implementations.MqttTcpChannel..ctor()
at MQTTnet.MqttClientFactory.CreateMqttClient(MqttClientOptions options)
at MqttClientModule.MqttClientModule.d__8.MoveNext()
--- End of stack trace from previous location where exception was thrown ---
at System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()
at System.Runtime.CompilerServices.AsyncMethodBuilderCore.<>c.b__6_1(Object state)
at System.Threading.QueueUserWorkItemCallbackDefaultContext.WaitCallback_Context(Object state)
at System.Threading.ExecutionContext.Run(ExecutionContext executionContext, ContextCallback callback, Object state)
at System.Threading.QueueUserWorkItemCallbackDefaultContext.System.Threading.IThreadPoolWorkItem.ExecuteWorkItem()
at System.Threading.ThreadPoolWorkQueue.Dispatch()
at System.Threading._ThreadPoolWaitCallback.PerformWaitCallback()

This error occurs in this line:
MqttClient client = new MqttClientFactory().CreateMqttClient(options);

It seems like my program uses the wrong framework. I found three frameworks:
MQTTnet.NetCoreApp
MQTTnet.NetFramework
and MQTTnet.UniversalWindows

Obviously I need the MQTTnet.NetCoreApp framework and it seems like it's using MQTTnet.UniversalWindows.

How do I tell my framework to use the right one?

Thanks again
Best regards,
Zuendelmeister

Supporting construction of web site

Version 2.1.3 has a connection bug it seems.

A request to send or receive data was disallowed because the socket is not connected and (when sending on a datagram socket using a sendto call) no address was supplied
at MQTTnet.Implementations.MqttTcpChannel.WriteAsync(Byte[] buffer)
at MQTTnet.Core.Serializer.MqttPacketWriter.WriteToAsync(IMqttCommunicationChannel destination)
at MQTTnet.Core.Serializer.DefaultMqttV311PacketSerializer.SerializeAsync(MqttConnectPacket packet, IMqttCommunicationChannel destination)
at MQTTnet.Core.Serializer.DefaultMqttV311PacketSerializer.SerializeAsync(MqttBasePacket packet, IMqttCommunicationChannel destination)
at MQTTnet.Core.Adapter.MqttChannelCommunicationAdapter.d__5.MoveNext()

This occurs with and without TLS enabled. The only thing I can gather so far is that the connection fails (via MqttClient.cs line 61) before the packet is sent (at MqttClient.cs line 74).

Can anyone else confirm this issue?

For the record, my MqttClientOptions are:

MqttClientOptions opts = new MqttClientOptions
{
    ClientId = Guid.NewGuid().ToString(),

    UserName = "flight|eedovmDCY48yEZtlXPdJwfeayNK5",
    Password = "...",

    Server = "ssl://mqtt-prod.airmap.io",
    Port = 8883,

    CleanSession = true,
    KeepAlivePeriod = TimeSpan.FromSeconds(15),
    DefaultCommunicationTimeout = TimeSpan.FromSeconds(2)
};

I just got an update earlier and now its failing my build. Here are the errors:

Cannot implicitly convert type 'MQTTnet.Core.Client.IMqttClient' to 'MQTTnet.Core.Client.MqttClient'. An explicit conversion exists (are you missing a cast?)

Cannot implicitly convert type 'MQTTnet.Core.Server.IMqttServer' to 'MQTTnet.Core.Server.MqttServer'. An explicit conversion exists (are you missing a cast?)

Probably an updated doc will help?

Test Name: MqttServer_PublishSimple_ExactlyOnce
Test FullName: MQTTnet.Core.Tests.MqttServerTests.MqttServer_PublishSimple_ExactlyOnce
Test Source: C:\Users\Zazzmatazz\Documents\Visual Studio 2017\Projects\MQTTnet-master\Tests\MQTTnet.Core.Tests\MqttServerTests.cs : line 40
Test Outcome: Failed
Test Duration: 0:00:02.632876

Result StackTrace:
at MQTTnet.Core.Tests.MqttServerTests.d__5.MoveNext() in C:\Users\Zazzmatazz\Documents\Visual Studio 2017\Projects\MQTTnet-master\Tests\MQTTnet.Core.Tests\MqttServerTests.cs:line 110
--- End of stack trace from previous location where exception was thrown ---
at System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task)
at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
at MQTTnet.Core.Tests.MqttServerTests.<MqttServer_PublishSimple_ExactlyOnce>d__2.MoveNext() in C:\Users\Zazzmatazz\Documents\Visual Studio 2017\Projects\MQTTnet-master\Tests\MQTTnet.Core.Tests\MqttServerTests.cs:line 40
--- End of stack trace from previous location where exception was thrown ---
at System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task)
at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
at System.Runtime.CompilerServices.TaskAwaiter.GetResult()
Result Message: Assert.AreEqual failed. Expected:<1>. Actual:<0>.

Hi again,

is there a maximum message size? If I send a message with 500 character everything works fine. The broker dies if I send a message with 1000 character though.
Any ideas?
Thanks as always =)

Best regards,
Zuendelmeister

I suggest to implement a new feature. Now the password seems to be static and must be given during the client creation. It is very useful to modify the psw during connection (dynamic psw) and not during client creation.
For example it's possible to use a TOTP psw during reconnection.

I will propose a PR.
JJ

Any way to make the IgnorableServerCertificateErrors configurable. I needed a way to add one but couldn't do so.

Right now it's harded coded with the following:

                _socket.Control.IgnorableServerCertificateErrors.Add(ChainValidationResult.IncompleteChain);
                _socket.Control.IgnorableServerCertificateErrors.Add(ChainValidationResult.RevocationInformationMissing);

while looking at the code i found an issue with MqttTcpChannel.

ReadAsync does not handle the returncode of _socket.ReceiveAsync. so if calling code wants to read 1000 bytes and method returns 500 just 500 bytes are read not the desired 1000. (byte array has correct size but half of it is unitialized)

hi two issue around client connected event:

1. if a client connected event is triggered Server.GetConnectedClients() does not contain the new client.
   id expect that the client is in the list of connected cliensts if an event signals this.

2. the identifier in the event is the remote endpoint of that socket or something else for websockets.
   id expect the client id of the connect packet in the event.

id like to submit a pr either change the existing behavior or add another event like session created.

@chkr1011 thoughts?

I'm struggling with the RecievePackets piece of you code on UWP. The issue only arise in Release, not in Debug mode.

Things seem to works on Windows 10, but not on Windows Mobile 10. I can replicate the issue both on a physical Windows 10 Mobile device as well as the Windows 10 Emulator. I even see the issue on rare occations on Windows 10 too actually.

I've been mulling over this piece .

I've narrowed it down to here. The behaviors is really strange, which leads me to thinking that this is a Race Condition sort of issue.

The Exception logged is:

The I/O operation has been aborted because of either a thread exit or an application request.

... and the exception originates from here.

I've been trying to figure out exactly what you are doing with the three threads you seem to spin up here, but it is not all that clear to me.

I always tend to get concerned when I see Task.Run(async () => ...) as a lot of care must be taken when going from one thread to multiple.

First of, I think that you can just make StartProcessReceivedPacket Async. Like this:

private async Task StartProcessReceivedPacketAsync(MqttBasePacket packet, CancellationToken cancellationToken)
{
       await ProcessReceivedPacketAsync(packet).ConfigureAwait(false);
}

I've tried this, and it seems to work. Well, it doesn't solve my issue in Windows Mobile 10, but it doesn't introduce any new issues either and with this change we are down to only two thread.

Any chance that you (or we) could take a closer look at this central piece of your code where packets are received? For one, I'd like to see the code here to become more explicit and easier to follow . Is this some sort of packet receiver pump? Are you handling the order of which things are being send and received somehow?

Personally, I would have used use Reactive Extensions and maybe even some state machine here to keep track of if Connection is made etc, as this is exactly the sort of areas where Rx excels and tight state management matters.

May you add a wiki example?
I cannot get it working.

Thank you
JJ

Hi,

can you please write an example of how to use the certificate for TLS encryption?
I couldn't find the way how to do it.
Thanks =)

Best regards,
Zuendelmeister

Sorry, what i mean when it can support websocket

When I specify "MqttQualityOfServiceLevel.AtMostOnce)" then the exception is thrown in MqttClientSession. A look in the code shows, in case of AtMostOnce the method doesn't has a return statement and falls through to the exception.

What action has to be executed instead?
Or is this QoS not supported?

I found an issue using TLS and MONO during connection.
To allow secured connections to work in MONO, we need to manage the RemoteCertificateValidationCallback of the SSLStream. So into MqttTcpChannel :

_sslStream = new SslStream(new NetworkStream(_socket, true), false, this.userCertificateValidationCallback, this.userCertificateSelectionCallback);

The two callbaack should be exposed to the calling application and the userCertificateValidationCallback should be assigned to a function returning true:

private bool remoteValidation(object sender, X509Certificate certificate, X509Chain chain, System.Net.Security.SslPolicyErrors sslPolicyErrors) { return true; }

this way we can use the library with TLS in MONO.

JJ

Hi there im currently evaluating your library.

I try to send as much packages as possible but the server seems to be only capable of handling 500 msgs/ sec. cpu is at 100% and most time is wasted in mqttpacketreader according to redgate perf profiler.

is this expected perf? i first thought i need a better json serializer library but json.net is just at less than 1 % of cpu.

Hi,

I just want to write a simple messaging app to test MQTTnet. I want to subscribe 2 or more clients. Client_1 should send to Client_2, Client_2 to Client_3 and Client_3 to Client_1. How do I do this?
Also for logging purposes: How do I log (or write to the console) incoming messages in the Server program?

Server Program:

var options = new MqttServerOptions
{
ConnectionValidator = p =>
{
Console.WriteLine("New");
Console.WriteLine(p.ClientId);
//if (p.ClientId == "SpecialClient")
//{
// if (p.Username != "USER" || p.Password != "PASS")
//{
// return MqttConnectReturnCode.ConnectionRefusedBadUsernameOrPassword;
// }
//}
return MqttConnectReturnCode.ConnectionAccepted;
}
};
var mqttServer = new MqttServerFactory().CreateMqttServer(options);
mqttServer.Start();
Console.WriteLine("Press any key to exit.");
Console.ReadLine();
mqttServer.Stop();

ClientProgram:

var options = new MqttClientOptions
{
Server = "localhost",
ClientId = "Client_1"
};
var client = new MqttClientFactory().CreateMqttClient(options);
client.ApplicationMessageReceived += (s, e) =>
{
Console.WriteLine("### RECEIVED APPLICATION MESSAGE ###");
Console.WriteLine($"+ Topic = {e.ApplicationMessage.Topic}");
Console.WriteLine($"+ Payload = {Encoding.UTF8.GetString(e.ApplicationMessage.Payload)}");
Console.WriteLine($"+ QoS = {e.ApplicationMessage.QualityOfServiceLevel}");
Console.WriteLine($"+ Retain = {e.ApplicationMessage.Retain}");
Console.WriteLine();
};
client.Connected += async (s,e) =>
{
Console.WriteLine("### CONNECTED WITH SERVER ###");
await client.SubscribeAsync(new List
{
new TopicFilter("bla1", MqttQualityOfServiceLevel.ExactlyOnce)
});
Console.WriteLine("### SUBSCRIBED ###");
};
client.Disconnected += async (s, e) =>
{
Console.WriteLine("### DISCONNECTED FROM SERVER ###");
await Task.Delay(TimeSpan.FromSeconds(5));
try
{
await client.ConnectAsync();
}
catch
{
Console.WriteLine("### RECONNECTING FAILED ###");
}
};
try
{
client.ConnectAsync();
}
catch
{
Console.WriteLine("### CONNECTING FAILED ###");
}
Console.WriteLine("### WAITING FOR APPLICATION MESSAGES ###");
while (true)
{
Console.ReadLine();
var applicationMessage = new MqttApplicationMessage(
"bla2",
Encoding.UTF8.GetBytes("Hello World"),
MqttQualityOfServiceLevel.AtLeastOnce,
false
);
Console.WriteLine("Sending");
client.PublishAsync(applicationMessage);
}

Thanks for helping me =)
Best regards,
Zuendelmeister

It seems that when using MQTTnet with this popular mqtt npm package, a packet type "0" is sent.

This causes exceptions in MQTTnet.

Hi,

I´m working in a Univeral Windows project. When I check you code it seems like I only can have one certificate. The server I´m accessing needs One certificate, one private key and a ca certificate. Create an PFX-certificate and I´m done?

/Raymond

i would like to submit a pr that deletes .net framework projects and adds net45 or net46 as targetframework to the .net.sdk based projects.

net45 would mean same compatibility but 3 ifdefs
net46 requires higher dependency but no ifdef required

@chkr1011 thoughts?

Hi again,

I have a problem with the reconnect. I'm sending data and sometimes the connection does not work anymore. My client code trys to reconnect:

client.Disconnected += async (s, e) =>
{
Console.WriteLine("### DISCONNECTED FROM SERVER ###");
sendMessages = false;
while (!sendMessages)
{
await Task.Delay(TimeSpan.FromSeconds(5));
Console.WriteLine("### Try to connect ###");
try
{
await client.ConnectAsync();
sendMessages = true;
}
catch (Exception ex)
{
Console.WriteLine("### RECONNECTING FAILED ### " + ex);
}
}
};

The reconnection never works though. This is what I get:

### Try to connect ###
### RECONNECTING FAILED ### MQTTnet.Core.Exceptions.MqttProtocolViolationException: It is not allowed to connect with a server after the connection is established.
at MQTTnet.Core.Client.MqttClient.d__20.MoveNext()
--- End of stack trace from previous location where exception was thrown ---
at System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()
at System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task)
at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
at System.Runtime.CompilerServices.TaskAwaiter.GetResult()
at MqttClientModule.MqttClientModule.<>c__DisplayClass12_0.<b__2>d.MoveNext()
### Try to connect ###
### RECONNECTING FAILED ### MQTTnet.Core.Exceptions.MqttProtocolViolationException: It is not allowed to connect with a server after the connection is established.
at MQTTnet.Core.Client.MqttClient.d__20.MoveNext()
--- End of stack trace from previous location where exception was thrown ---
at System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()
at System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task)
at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
at System.Runtime.CompilerServices.TaskAwaiter.GetResult()
at MqttClientModule.MqttClientModule.<>c__DisplayClass12_0.<b__2>d.MoveNext()
...

In this case it does not work to restart the client or the broker. I have to kill both and restart them or a reconnection is impossible.
How do I make shure that my reconnect works properly?
And how can I destroy the client object? There is no dispose.
Thanks =)

Best regards,
Zuendelmeister

When MqttApplicationMessage.MqttQualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce that client can not FireApplicationMessageReceivedEvent,but MqttApplicationMessage.MqttQualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce can FireApplicationMessageReceivedEvent ,why?

Is there a reason that you did not implement the DubFlag as part of Publish?

Or, maybe I'm just overlooking something?

On DubFlag.

I ran into a issue when trying to connect via SSL with a Self-Signed Cert. I needed to override the validation function to always return true for self-signed certs. In previous versions of .NET I could use

ServicePointManager.ServerCertificateValidationCallback = (sender, certificate, chain, errors) => { return true; };

In dotnet CORE this is ignored. My only remaining option is to call the SslStream ctor overload that accepts a validation function

public SslStream(Stream innerStream, bool leaveInnerStreamOpen, RemoteCertificateValidationCallback userCertificateValidationCallback)

I implemented a custom MqttClientFactory that in turn called public MqttTcpChannel(Socket socket, SslStream sslStream)

This allowed me to pass in the custom validation logic. However when I ran the code the certificate was still being flagged as invalid. Upon digging into the code I noticed

public async Task ConnectAsync(MqttClientOptions options)

Ignores this value... by always overwriting it with _sslStream = new SslStream(new NetworkStream(_socket, true));

if (options.TlsOptions.UseTls) { _sslStream = new SslStream(new NetworkStream(_socket, true)); await _sslStream.AuthenticateAsClientAsync(options.Server, LoadCertificates(options), SslProtocols.Tls12, options.TlsOptions.CheckCertificateRevocation); }

This does not appear to be 100% correct, I see that one overload is intended for SERVER opposed to CLIENT channel. However I don't think having ConnectAsync stomp on the value set in the ctor is correct. Its also not intuitive that only a specific overload must be used depending on which end of the channel is communicating.

I would be happy to submit a pull request. I'm thinking that Client/Server need separate implementations of IMqttCommunicationChannel OR marking one as internal (this may not work without the ugly InternalVisibleToAttribute)

Another approach is to just expose the validation delegate as a property on MqttClientOptions. This seems hacky and does nothing to address the greater issue (specific ctor for Client vs. Server)

Thoughts?

How to Use With WebSocket Server,Could you Write a Demo With this?

Great library so far, How do I ignore tls certificate errors using your library.

Dear’s

First of all, thank a lot for the great solution.

I did some test by using the MQTTserver in a UWP app and identified an issue. I used the example code from the Readme and set the Port to 1883 in the DefaultEndpointOptions.
The Server runs into a Nullpointer Exception, because the ReceiveStream in the MqttTcpChannel is not initialized. I see that the initialization is done in the ConnectAsync method, but it looks the method is used in MQQTClient context but not if the Channel is used in Server context.

My fix in MqttTcpChannel (MQTTnet.UniversalWindows), please validate ;-)
…..

public MqttTcpChannel(StreamSocket socket)
{
	_socket = socket ?? throw new ArgumentNullException(nameof(socket));
	CreateStreams(socket);
}

private void CreateStreams(StreamSocket socket)
{
	SendStream = socket.OutputStream.AsStreamForWrite();
	ReceiveStream = socket.InputStream.AsStreamForRead();
	RawReceiveStream = ReceiveStream;
}

public async Task ConnectAsync(MqttClientOptions options)
{
	if (options == null) throw new ArgumentNullException(nameof(options));
	if (_socket == null)
	{
		_socket = new StreamSocket();
	}
	if (!options.TlsOptions.UseTls)
	{
		await _socket.ConnectAsync(new HostName(options.Server), options.GetPort().ToString());
	}
	else
	{
		_socket.Control.ClientCertificate = LoadCertificate(options);

		if (!options.TlsOptions.CheckCertificateRevocation)
		{
			_socket.Control.IgnorableServerCertificateErrors.Add(ChainValidationResult.IncompleteChain);
			_socket.Control.IgnorableServerCertificateErrors.Add(ChainValidationResult.RevocationInformationMissing);
		}

		await _socket.ConnectAsync(new HostName(options.Server), options.GetPort().ToString(), SocketProtectionLevel.Tls12);
	}

	CreateStreams(_socket);
}

Cheers

I notice your improvements in the developer branch. Great.
What do you think about making IApplicationMessageReceiver and IApplicationMessagePublisher a common interface for client and server ?
HTTPnet also runs on "client platforms", I think it would be an advantage for MQTTnet based components to provide the same abstraction for receiving and sending on both sides.

Maybe I'm doing something wrong, however while creating a Rx Wrapper for MQTTnet I notices that I was unable to set TlsOptions.

Did I miss something or...?

Cannot use a pfx-certificate. Get "An unknown error occurred while processing the certificate".

Should I use code:
tlsOptions.Certificates = new List<byte[]>();
byte[] byteArray = cert.GetRawCertData();
tlsOptions.Certificates.Add(byteArray);

I tested the library on MONO (last version), and it works very well.
I will do much more intensive test this month. But you can assume:
Supported frameworks:
NET Standard 1.3+
.NET Core 1.1+
.NET Core App 1.1+
.NET Framework 4.5.2+ (x86, x64, AnyCPU)
Universal Windows (UWP) 10.0.10240+ (x86, x64, ARM, AnyCPU)
MONO

I found a small issue in the Server implementation. I tested the server in a “rough environment”, means sometimes the MQTT clients go down without an unsubscription by the server and are off for minutes and reconnect again.

My understanding of the code is, the SendPendingPublishPacketsAsync Methods sends the pakages to the client. In case the remote host is not available, the SendPacketsAsync runs into an exception and the packages are restored back to the send queue until a cancelation is triggered from outside. The result is, a loop and the code never returns and give up to send packages to a not connectable client.

I fixed that by using a trails counter and exit after 10 trails with an exception. On a first view, that works fine, but after some time the server crashes and disconnects other clients. I’m investigating on that.. a bit hard because some documentation is missing ;-)

private async Task SendPendingPublishPacketsAsync(CancellationToken cancellationToken, IMqttCommunicationAdapter adapter)
        {
            var consumable = _pendingPublishPackets.GetConsumingEnumerable();
            int trials = 0;
            while (!cancellationToken.IsCancellationRequested)
            {
                if (_pendingPublishPackets.Count == 0)
                {
                    await Task.Delay(TimeSpan.FromMilliseconds(5), cancellationToken).ConfigureAwait(false);
                    continue;
                }

                var packets = consumable.Take(_pendingPublishPackets.Count).ToList();
                try
                {
                    await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, packets).ConfigureAwait(false);
                    trials = 0;
                }
                catch (MqttCommunicationTimedOutException exception)
                {
                    // remote host is not reachable
                    MqttTrace.Warning(nameof(MqttClientMessageQueue), exception, "Sending publish packet failed.");
                    foreach (var publishPacket in packets)
                    {
                        publishPacket.Dup = true;
                        _pendingPublishPackets.Add(publishPacket, cancellationToken);
                        trials++;
                    }
                    await Task.Delay(TimeSpan.FromMilliseconds(20), cancellationToken).ConfigureAwait(false);
                }
                catch (MqttCommunicationException exception)
                {
                    MqttTrace.Warning(nameof(MqttClientMessageQueue), exception, "Sending publish packet failed.");
                    foreach (var publishPacket in packets)
                    {
                        publishPacket.Dup = true;
                        _pendingPublishPackets.Add(publishPacket, cancellationToken);
                        trials++;
                    }
                }                
                catch (Exception exception)
                {
                    MqttTrace.Error(nameof(MqttClientMessageQueue), exception, "Sending publish packet failed.");
                    foreach (var publishPacket in packets)
                    {
                        publishPacket.Dup = true;
                        _pendingPublishPackets.Add(publishPacket, cancellationToken);
                        trials++;
                    }
                }

                if (trials > 10)
                {
                    // give up
                    MqttTrace.Warning(nameof(MqttClientMessageQueue), "Sending publish packet failed. Give up");
                    throw new MqttCommunicationException($"Give up to try to send packages, receiver is lost. Trials {trials}");
                }
            }
        }

Regards
Hans

in file MqttTcpChannel.cs

public Task DisconnectAsync()
{
try
{
//not judge if _sslStream is null
_sslStream.Dispose();

            _socket.Dispose();
            return Task.FromResult(0);
        }
        catch (SocketException exception)
        {
            throw new MqttCommunicationException(exception);
        }
    }

The documentation is unclear and I couldn't find anything in the source code about Web Socket support in this library.
If it does support WS connections please tell me how to use it because I can't connect to the server with web sockets but I can using pure TCP connections.

Thanks.

Hi!
It looks like the disconnected event of the client is called when a ConnectAsync fails.
Wouldn't it be better to call the event only if the client was really connected before?

BR, Rainer.

I identified issues with my client connections and ask you about your experience or testing’s in your environments.
My configuration is as following:

Server as UWP on a Win10 Workstation
Client as UWP (subscriber) on Win10 Iot Raspi
Fhem as Client (subscriber) on Raspi
Easy ESP as client (publisher) on ESP8266

The problem is, the server closes the connections to the client after random timeout exceptions. I did the following tests:

a.) Use a Raspi with fhem (client), Easy ESP(client) and mosquito (broker), fhem doesn’t log any issues.
b.) Use Mqttnet as Server, Easy ESP (publisher). Fhem and Mqttnet do not report any issues
c.) Use Mqttnet as Server, fhem (subscriber) without publisher. Fhem and Mqttnet do not report any issues (see log: Log_OnlyPublisher.txt)
d.) Use Mqttnet as Server, fhem (subscriber) and EasyESP(publisher). Fhem reports broken connections and Mqttnet several Exceptions (see log Log_ESP_FhemClient.txt)
e.) Use Mqttnet as Server, Mqttnet Client (subscriber) and EasyESP(publisher). Mqttnet Client reports broken connections and Mqttnet several Exceptions (see log Log_ESP_UWPClient.txt)

I did also tests running the MqttServer on a Raspi and use MqttBox as client and remember the issues with lost connections.

At the beginning, I assume an performance issue in fhem or my UWP client, but with issues also with MqttBox, I assume another issue.

To explain the test and logs

The publisher sends two issues: temperature and pressure. In several cases, the subscriber receives all fine, but sometimes the server runs into an Timeout Exception.

Any ideas ?

Log_ESP_FhemClient.zip
Log_ESP_UWPClient.zip
Log_OnlyPublisher.zip

I integrated MQTTnet in an AspNetCore web app. WebSocketStream can be used for all websocket client and server implementations when you change the parameter to class WebSocket.

Hi again,

I finally found the time to test you ApplicationMessageReceived event for the mqtt broker.
This is what I did:

` MqttServer mqttServer = new MqttServerFactory().CreateMqttServer(options);

        mqttServer.ApplicationMessageReceived += (s, e) =>

        {

            string messageString = Encoding.UTF8.GetString(e.ApplicationMessage.Payload);

            try

            {

                Dictionary<string, string> messageProperty = new Dictionary<string, string>();

                Console.WriteLine(e.ClientId + " sent " + e.ApplicationMessage.Payload);

            }

            catch (Exception ex)

            {

                Console.WriteLine(ex);

            }

        };

`
It looks all like I wanted it. Unfortunately VS somehow cannot "find" the dll, even though I added it in my project:

Next I tried to use the NugetPackage. With it everything works gread on windows. Unfortunately it does not work on my raspberry pi.

What did I do wrong?
Also I cannot find the "MQTTnet.NetCoreApp" folder under "Frameworks":

Thanks again.

Best regards,
Zuendelmeister