using MQTTnet.Core; using MQTTnet.Core.Client; using MQTTnet.Core.Diagnostics; using MQTTnet.Core.Packets; using MQTTnet.Core.Protocol; using MQTTnet.Core.Server; using System; using System.Collections.Generic; using System.Text; using System.Threading; using System.Threading.Tasks; namespace MQTTnet.TestApp.NetCore { public static class Program { public static void Main(string[] args) { Console.WriteLine("MQTTnet - TestApp.NetFramework"); Console.WriteLine("1 = Start client"); Console.WriteLine("2 = Start server"); var pressedKey = Console.ReadKey(true); if (pressedKey.Key == ConsoleKey.D1) { Task.Run(() => RunClientAsync(args)); Thread.Sleep(Timeout.Infinite); } else if (pressedKey.Key == ConsoleKey.D2) { Task.Run(() => RunServerAsync(args)); Thread.Sleep(Timeout.Infinite); } } private static async Task RunClientAsync(string[] arguments) { MqttTrace.TraceMessagePublished += (s, e) => { Console.WriteLine($">> [{e.ThreadId}] [{e.Source}] [{e.Level}]: {e.Message}"); if (e.Exception != null) { Console.WriteLine(e.Exception); } }; try { var options = new MqttClientOptions { Server = "localhost", ClientId = "XYZ", CleanSession = true, ConnectionType = MqttConnectionType.Ws }; 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("#", MqttQualityOfServiceLevel.AtMostOnce) }); 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 { await client.ConnectAsync(); } catch (Exception exception) { Console.WriteLine("### CONNECTING FAILED ###" + Environment.NewLine + exception); } Console.WriteLine("### WAITING FOR APPLICATION MESSAGES ###"); while (true) { Console.ReadLine(); var applicationMessage = new MqttApplicationMessage( "A/B/C", Encoding.UTF8.GetBytes("Hello World"), MqttQualityOfServiceLevel.AtLeastOnce, false ); await client.PublishAsync(applicationMessage); } } catch (Exception exception) { Console.WriteLine(exception); } } private static void RunServerAsync(string[] arguments) { MqttTrace.TraceMessagePublished += (s, e) => { Console.WriteLine($">> [{e.ThreadId}] [{e.Source}] [{e.Level}]: {e.Message}"); if (e.Exception != null) { Console.WriteLine(e.Exception); } }; try { var options = new MqttServerOptions { ConnectionValidator = p => { 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(); } catch (Exception e) { Console.WriteLine(e); } Console.ReadLine(); } } }