using System; using System.Collections.Concurrent; using System.Collections.ObjectModel; using System.Text; using System.Threading.Tasks; using Windows.Security.Cryptography.Certificates; using Windows.UI.Core; using Windows.UI.Xaml; using MQTTnet.Client; using MQTTnet.Client.Connecting; using MQTTnet.Client.Disconnecting; using MQTTnet.Client.Options; using MQTTnet.Diagnostics; using MQTTnet.Exceptions; using MQTTnet.Extensions.ManagedClient; using MQTTnet.Extensions.Rpc; using MQTTnet.Formatter; using MQTTnet.Implementations; using MQTTnet.Protocol; using MQTTnet.Server; using MQTTnet.Server.Status; using MqttClientConnectedEventArgs = MQTTnet.Client.Connecting.MqttClientConnectedEventArgs; using MqttClientDisconnectedEventArgs = MQTTnet.Client.Disconnecting.MqttClientDisconnectedEventArgs; namespace MQTTnet.TestApp.UniversalWindows { public sealed partial class MainPage { private readonly ConcurrentQueue _traceMessages = new ConcurrentQueue(); private readonly ObservableCollection _sessions = new ObservableCollection(); private IMqttClient _mqttClient; private IManagedMqttClient _managedMqttClient; private IMqttServer _mqttServer; public MainPage() { InitializeComponent(); ClientId.Text = Guid.NewGuid().ToString("D"); MqttNetGlobalLogger.LogMessagePublished += OnTraceMessagePublished; } private async void OnTraceMessagePublished(object sender, MqttNetLogMessagePublishedEventArgs e) { _traceMessages.Enqueue(e.TraceMessage); await UpdateLogAsync(); } private async Task UpdateLogAsync() { while (_traceMessages.Count > 100) { _traceMessages.TryDequeue(out _); } var logText = new StringBuilder(); foreach (var traceMessage in _traceMessages) { logText.AppendFormat( "[{0:yyyy-MM-dd HH:mm:ss.fff}] [{1}] [{2}] [{3}] [{4}]{5}", traceMessage.Timestamp, traceMessage.Level, traceMessage.Source, traceMessage.ThreadId, traceMessage.Message, Environment.NewLine); if (traceMessage.Exception != null) { logText.AppendLine(traceMessage.Exception.ToString()); } } await Trace.Dispatcher.RunAsync(CoreDispatcherPriority.Low, () => { Trace.Text = logText.ToString(); }); } private async void Connect(object sender, RoutedEventArgs e) { var tlsOptions = new MqttClientTlsOptions { UseTls = UseTls.IsChecked == true, IgnoreCertificateChainErrors = true, IgnoreCertificateRevocationErrors = true, AllowUntrustedCertificates = true }; var options = new MqttClientOptions { ClientId = ClientId.Text, ProtocolVersion = MqttProtocolVersion.V500 }; if (UseTcp.IsChecked == true) { options.ChannelOptions = new MqttClientTcpOptions { Server = Server.Text, Port = int.Parse(Port.Text), TlsOptions = tlsOptions }; } if (UseWs.IsChecked == true) { options.ChannelOptions = new MqttClientWebSocketOptions { Uri = Server.Text, TlsOptions = tlsOptions }; } if (options.ChannelOptions == null) { throw new InvalidOperationException(); } options.Credentials = new MqttClientCredentials { Username = User.Text, Password = Password.Text }; options.CleanSession = CleanSession.IsChecked == true; options.KeepAlivePeriod = TimeSpan.FromSeconds(double.Parse(KeepAliveInterval.Text)); if (UseMqtt310.IsChecked == true) { options.ProtocolVersion = MqttProtocolVersion.V310; } else if (UseMqtt311.IsChecked == true) { options.ProtocolVersion = MqttProtocolVersion.V311; } else if (UseMqtt500.IsChecked == true) { options.ProtocolVersion = MqttProtocolVersion.V500; } try { if (_mqttClient != null) { await _mqttClient.DisconnectAsync(); _mqttClient.UseApplicationMessageReceivedHandler(HandleReceivedApplicationMessage); _mqttClient.ConnectedHandler = new MqttClientConnectedHandlerDelegate(x => OnConnected(x)); _mqttClient.DisconnectedHandler = new MqttClientDisconnectedHandlerDelegate(x => OnDisconnected(x)); } var factory = new MqttFactory(); if (UseManagedClient.IsChecked == true) { _managedMqttClient = factory.CreateManagedMqttClient(); _managedMqttClient.UseApplicationMessageReceivedHandler(HandleReceivedApplicationMessage); _managedMqttClient.ConnectedHandler = new MqttClientConnectedHandlerDelegate(x => OnConnected(x)); _managedMqttClient.DisconnectedHandler = new MqttClientDisconnectedHandlerDelegate(x => OnDisconnected(x)); await _managedMqttClient.StartAsync(new ManagedMqttClientOptions { ClientOptions = options }); } else { _mqttClient = factory.CreateMqttClient(); _mqttClient.UseApplicationMessageReceivedHandler(HandleReceivedApplicationMessage); _mqttClient.ConnectedHandler = new MqttClientConnectedHandlerDelegate(x => OnConnected(x)); _mqttClient.DisconnectedHandler = new MqttClientDisconnectedHandlerDelegate(x => OnDisconnected(x)); await _mqttClient.ConnectAsync(options); } } catch (Exception exception) { Trace.Text += exception + Environment.NewLine; } } private void OnDisconnected(MqttClientDisconnectedEventArgs e) { _traceMessages.Enqueue(new MqttNetLogMessage("", DateTime.Now, -1, "", MqttNetLogLevel.Info, "! DISCONNECTED EVENT FIRED", null)); Task.Run(UpdateLogAsync); } private void OnConnected(MqttClientConnectedEventArgs e) { _traceMessages.Enqueue(new MqttNetLogMessage("", DateTime.Now, -1, "", MqttNetLogLevel.Info, "! CONNECTED EVENT FIRED", null)); Task.Run(UpdateLogAsync); } private async Task HandleReceivedApplicationMessage(MqttApplicationMessageReceivedEventArgs eventArgs) { var item = $"Timestamp: {DateTime.Now:O} | Topic: {eventArgs.ApplicationMessage.Topic} | Payload: {Encoding.UTF8.GetString(eventArgs.ApplicationMessage.Payload)} | QoS: {eventArgs.ApplicationMessage.QualityOfServiceLevel}"; await Dispatcher.RunAsync(CoreDispatcherPriority.Low, () => { if (AddReceivedMessagesToList.IsChecked == true) { ReceivedMessages.Items.Add(item); } }); } private async void Publish(object sender, RoutedEventArgs e) { try { var qos = MqttQualityOfServiceLevel.AtMostOnce; if (QoS1.IsChecked == true) { qos = MqttQualityOfServiceLevel.AtLeastOnce; } if (QoS2.IsChecked == true) { qos = MqttQualityOfServiceLevel.ExactlyOnce; } var payload = new byte[0]; if (PlainText.IsChecked == true) { payload = Encoding.UTF8.GetBytes(Payload.Text); } if (Base64.IsChecked == true) { payload = Convert.FromBase64String(Payload.Text); } var message = new MqttApplicationMessageBuilder() .WithContentType(ContentType.Text) .WithResponseTopic(ResponseTopic.Text) .WithTopic(Topic.Text) .WithPayload(payload) .WithQualityOfServiceLevel(qos) .WithRetainFlag(Retain.IsChecked == true) .Build(); if (_mqttClient != null) { await _mqttClient.PublishAsync(message); } if (_managedMqttClient != null) { await _managedMqttClient.PublishAsync(message); } } catch (Exception exception) { Trace.Text += exception + Environment.NewLine; } } private async void Disconnect(object sender, RoutedEventArgs e) { try { if (_mqttClient != null) { await _mqttClient.DisconnectAsync(); _mqttClient.Dispose(); _mqttClient = null; } if (_managedMqttClient != null) { await _managedMqttClient.StopAsync(); _managedMqttClient.Dispose(); _managedMqttClient = null; } } catch (Exception exception) { Trace.Text += exception + Environment.NewLine; } } private void ClearLog(object sender, RoutedEventArgs e) { while (_traceMessages.Count > 0) { _traceMessages.TryDequeue(out _); } Trace.Text = string.Empty; } private async void Subscribe(object sender, RoutedEventArgs e) { try { var qos = MqttQualityOfServiceLevel.AtMostOnce; if (SubscribeQoS1.IsChecked == true) { qos = MqttQualityOfServiceLevel.AtLeastOnce; } if (SubscribeQoS2.IsChecked == true) { qos = MqttQualityOfServiceLevel.ExactlyOnce; } var topicFilter = new TopicFilter { Topic = SubscribeTopic.Text, QualityOfServiceLevel = qos }; if (_mqttClient != null) { await _mqttClient.SubscribeAsync(topicFilter); } if (_managedMqttClient != null) { await _managedMqttClient.SubscribeAsync(topicFilter); } } catch (Exception exception) { Trace.Text += exception + Environment.NewLine; } } private async void Unsubscribe(object sender, RoutedEventArgs e) { try { if (_mqttClient != null) { await _mqttClient.UnsubscribeAsync(SubscribeTopic.Text); } if (_managedMqttClient != null) { await _managedMqttClient.UnsubscribeAsync(SubscribeTopic.Text); } } catch (Exception exception) { Trace.Text += exception + Environment.NewLine; } } // This code is for the Wiki at GitHub! // ReSharper disable once UnusedMember.Local private async void StartServer(object sender, RoutedEventArgs e) { if (_mqttServer != null) { return; } JsonServerStorage storage = null; if (ServerPersistRetainedMessages.IsChecked == true) { storage = new JsonServerStorage(); if (ServerClearRetainedMessages.IsChecked == true) { storage.Clear(); } } _mqttServer = new MqttFactory().CreateMqttServer(); var options = new MqttServerOptions(); options.DefaultEndpointOptions.Port = int.Parse(ServerPort.Text); options.Storage = storage; options.EnablePersistentSessions = ServerAllowPersistentSessions.IsChecked == true; await _mqttServer.StartAsync(options); } private async void StopServer(object sender, RoutedEventArgs e) { if (_mqttServer == null) { return; } await _mqttServer.StopAsync(); _mqttServer = null; } private void ClearReceivedMessages(object sender, RoutedEventArgs e) { ReceivedMessages.Items.Clear(); } private async void ExecuteRpc(object sender, RoutedEventArgs e) { var qos = MqttQualityOfServiceLevel.AtMostOnce; if (RpcQoS1.IsChecked == true) { qos = MqttQualityOfServiceLevel.AtLeastOnce; } if (RpcQoS2.IsChecked == true) { qos = MqttQualityOfServiceLevel.ExactlyOnce; } var payload = new byte[0]; if (RpcText.IsChecked == true) { payload = Encoding.UTF8.GetBytes(RpcPayload.Text); } if (RpcBase64.IsChecked == true) { payload = Convert.FromBase64String(RpcPayload.Text); } try { var rpcClient = new MqttRpcClient(_mqttClient); var response = await rpcClient.ExecuteAsync(TimeSpan.FromSeconds(5), RpcMethod.Text, payload, qos); RpcResponses.Items.Add(RpcMethod.Text + " >>> " + Encoding.UTF8.GetString(response)); } catch (MqttCommunicationTimedOutException) { RpcResponses.Items.Add(RpcMethod.Text + " >>> [TIMEOUT]"); } catch (Exception exception) { RpcResponses.Items.Add(RpcMethod.Text + " >>> [EXCEPTION (" + exception.Message + ")]"); } } private void ClearRpcResponses(object sender, RoutedEventArgs e) { RpcResponses.Items.Clear(); } private void ClearSessions(object sender, RoutedEventArgs e) { _sessions.Clear(); } private void RefreshSessions(object sender, RoutedEventArgs e) { if (_mqttServer == null) { return; } var sessions = _mqttServer.GetClientStatusAsync().GetAwaiter().GetResult(); _sessions.Clear(); foreach (var session in sessions) { _sessions.Add(session); } ListViewSessions.DataContext = _sessions; } #region Wiki Code private async Task WikiCode() { { // Use a custom identifier for the trace messages. var clientOptions = new MqttClientOptionsBuilder() .Build(); } { // Create a new MQTT client. var factory = new MqttFactory(); var client = factory.CreateMqttClient(); // Create TCP based options using the builder. var options = new MqttClientOptionsBuilder() .WithClientId("Client1") .WithTcpServer("broker.hivemq.com") .WithCredentials("bud", "%spencer%") .WithTls() .WithCleanSession() .Build(); await client.ConnectAsync(options); // Reconnecting client.UseDisconnectedHandler(async e => { Console.WriteLine("### DISCONNECTED FROM SERVER ###"); await Task.Delay(TimeSpan.FromSeconds(5)); try { await client.ConnectAsync(options); } catch { Console.WriteLine("### RECONNECTING FAILED ###"); } }); // Consuming messages client.UseApplicationMessageReceivedHandler(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(); }); // Subscribe after connect client.UseConnectedHandler(async e => { Console.WriteLine("### CONNECTED WITH SERVER ###"); // Subscribe to a topic await client.SubscribeAsync(new TopicFilterBuilder().WithTopic("my/topic").Build()); Console.WriteLine("### SUBSCRIBED ###"); }); // Subscribe to a topic await client.SubscribeAsync(new TopicFilterBuilder().WithTopic("my/topic").Build()); // Unsubscribe from a topic await client.UnsubscribeAsync("my/topic"); // Publish an application message var applicationMessage = new MqttApplicationMessageBuilder() .WithTopic("A/B/C") .WithPayload("Hello World") .WithAtLeastOnceQoS() .Build(); await client.PublishAsync(applicationMessage); } { { // Use TCP connection. var options = new MqttClientOptionsBuilder() .WithTcpServer("broker.hivemq.com", 1883) // Port is optional .Build(); } { // Use secure TCP connection. var options = new MqttClientOptionsBuilder() .WithTcpServer("broker.hivemq.com") .WithTls() .Build(); } { // Use WebSocket connection. var options = new MqttClientOptionsBuilder() .WithWebSocketServer("broker.hivemq.com:8000/mqtt") .Build(); } { // Create TCP based options manually var options = new MqttClientOptions { ClientId = "Client1", Credentials = new MqttClientCredentials { Username = "bud", Password = "%spencer%" }, ChannelOptions = new MqttClientTcpOptions { Server = "broker.hivemq.org", TlsOptions = new MqttClientTlsOptions { UseTls = true } }, }; } } // ---------------------------------- { var options = new MqttServerOptions(); options.ConnectionValidator = new MqttServerConnectionValidatorDelegate(c => { if (c.ClientId.Length < 10) { c.ReturnCode = MqttConnectReturnCode.ConnectionRefusedIdentifierRejected; return; } if (c.Username != "mySecretUser") { c.ReturnCode = MqttConnectReturnCode.ConnectionRefusedBadUsernameOrPassword; return; } if (c.Password != "mySecretPassword") { c.ReturnCode = MqttConnectReturnCode.ConnectionRefusedBadUsernameOrPassword; return; } c.ReturnCode = MqttConnectReturnCode.ConnectionAccepted; }); var factory = new MqttFactory(); var mqttServer = factory.CreateMqttServer(); await mqttServer.StartAsync(options); Console.WriteLine("Press any key to exit."); Console.ReadLine(); await mqttServer.StopAsync(); } // ---------------------------------- // For UWP apps: MqttTcpChannel.CustomIgnorableServerCertificateErrorsResolver = o => { if (o.Server == "server_with_revoked_cert") { return new[] { ChainValidationResult.Revoked }; } return new ChainValidationResult[0]; }; { // Start a MQTT server. var mqttServer = new MqttFactory().CreateMqttServer(); await mqttServer.StartAsync(new MqttServerOptions()); Console.WriteLine("Press any key to exit."); Console.ReadLine(); await mqttServer.StopAsync(); } { // Configure MQTT server. var optionsBuilder = new MqttServerOptionsBuilder() .WithConnectionBacklog(100) .WithDefaultEndpointPort(1884); var options = new MqttServerOptions { }; options.ConnectionValidator = new MqttServerConnectionValidatorDelegate(c => { if (c.ClientId != "Highlander") { c.ReturnCode = MqttConnectReturnCode.ConnectionRefusedIdentifierRejected; return; } c.ReturnCode = MqttConnectReturnCode.ConnectionAccepted; }); var mqttServer = new MqttFactory().CreateMqttServer(); await mqttServer.StartAsync(optionsBuilder.Build()); } { // Setup client validator. var options = new MqttServerOptions { ConnectionValidator = new MqttServerConnectionValidatorDelegate(c => { if (c.ClientId.Length < 10) { c.ReturnCode = MqttConnectReturnCode.ConnectionRefusedIdentifierRejected; return; } if (c.Username != "mySecretUser") { c.ReturnCode = MqttConnectReturnCode.ConnectionRefusedBadUsernameOrPassword; return; } if (c.Password != "mySecretPassword") { c.ReturnCode = MqttConnectReturnCode.ConnectionRefusedBadUsernameOrPassword; return; } c.ReturnCode = MqttConnectReturnCode.ConnectionAccepted; }) }; } { // Create a new MQTT server. var mqttServer = new MqttFactory().CreateMqttServer(); } { // Setup and start a managed MQTT client. var options = new ManagedMqttClientOptionsBuilder() .WithAutoReconnectDelay(TimeSpan.FromSeconds(5)) .WithClientOptions(new MqttClientOptionsBuilder() .WithClientId("Client1") .WithTcpServer("broker.hivemq.com") .WithTls().Build()) .Build(); var mqttClient = new MqttFactory().CreateManagedMqttClient(); await mqttClient.SubscribeAsync(new TopicFilterBuilder().WithTopic("my/topic").Build()); await mqttClient.StartAsync(options); } { // Use a custom log ID for the logger. var factory = new MqttFactory(); var client = factory.CreateMqttClient(new MqttNetLogger("MyCustomId")); } { var client = new MqttFactory().CreateMqttClient(); var message = new MqttApplicationMessageBuilder() .WithTopic("MyTopic") .WithPayload("Hello World") .WithExactlyOnceQoS() .WithRetainFlag() .Build(); await client.PublishAsync(message); } { // Write all trace messages to the console window. MqttNetGlobalLogger.LogMessagePublished += (s, e) => { var trace = $">> [{e.TraceMessage.Timestamp:O}] [{e.TraceMessage.ThreadId}] [{e.TraceMessage.Source}] [{e.TraceMessage.Level}]: {e.TraceMessage.Message}"; if (e.TraceMessage.Exception != null) { trace += Environment.NewLine + e.TraceMessage.Exception.ToString(); } Console.WriteLine(trace); }; } } #endregion } }