Add communication with local MQTT

TestApp/Program.cs

@@ -1,12 +1,19 @@
-using NModbus;
+using MQTTnet;
+using MQTTnet.Client;
+using NModbus;
 using System.Net.Sockets;
 
 namespace TestApp
 {
     internal class Program
     {
-        static void Main(string[] args)
+        private static IMqttClient _MqttClient;
+        private static Dictionary<UInt64, UInt64> _CurrentValues = new Dictionary<ulong, ulong>();
+
+        static async Task Main(string[] args)
         {
+            await ConnectMQTT();
+
             while (true)
             {
                 try
@@ -32,10 +39,11 @@ namespace TestApp
             {
                 try
                 {
+                    var currentWirkleistung = _CurrentValues.GetValueOrDefault(1099528667391UL) / 1000; 
+
                     ushort input = master.ReadHoldingRegisters(1, 1000, 1)[0];
-                    ushort output = (ushort)(input + 50);
+                    ushort output = (ushort)(input + currentWirkleistung);
                     master.WriteMultipleRegisters(1, 1100, new ushort[] { output });
-                    Console.WriteLine($"Register values updated to {input} + 50 = {output}");
                     Thread.Sleep(100);
                 }
                 catch (Exception e)
@@ -46,5 +54,56 @@ namespace TestApp
             }
         }
 
+        private static async Task ConnectMQTT() {
+            var factory = new MqttFactory();
+
+            _MqttClient = factory.CreateMqttClient();
+
+            var clientOptions = new MqttClientOptionsBuilder()
+                .WithTcpServer("localhost")
+                .Build();
+
+            Console.WriteLine("Connecting to local MQTT ...");
+            await _MqttClient.ConnectAsync(clientOptions);
+            Console.WriteLine("Connected to local MQTT!");
+
+            _MqttClient.ApplicationMessageReceivedAsync += Client_ApplicationMessageReceivedAsync;
+
+            var subscribeOptions = factory.CreateSubscribeOptionsBuilder()
+                .WithTopicFilter(
+                    f => {
+                        f.WithTopic("gdr/local/values/smart-meter");
+                    }
+                )
+                .Build();
+
+            Console.WriteLine("Subscribing to smart-meter values ...");
+            var response = await _MqttClient.SubscribeAsync(subscribeOptions);
+            Console.WriteLine("Subscribed to smart-meter values!");
+        }
+
+        private static Task Client_ApplicationMessageReceivedAsync(MqttApplicationMessageReceivedEventArgs arg)
+        {
+            switch (arg.ApplicationMessage.Topic) {
+                case "gdr/local/values/smart-meter":
+                    ParseGDRs(arg.ApplicationMessage.Payload);
+                    break;
+            }
+
+            return Task.CompletedTask;
+        }
+
+        private static void ParseGDRs(byte[] payload)
+        {
+            var gdrs = GDRs.Parser.ParseFrom(payload);
+
+            if (!gdrs.GDRs_.TryGetValue("smart-meter", out GDR smartMeterGDR)) {
+                return;
+            }
+
+            foreach (var value in smartMeterGDR.Values) {
+                _CurrentValues[value.Key] = value.Value;
+            }
+        }
     }
 }