[NF] AUserBackend hinzugefügt

2017-10-02 22:51:08 +00:00
parent 57856a4d8f
commit 6369fe4b46
10 changed files with 122 additions and 98 deletions
@@ -0,0 +1,121 @@
+using System;
+using System.Collections.Generic;
+using System.Diagnostics;
+using System.Text.RegularExpressions;
+
+namespace BlubbFish.Utils.IoT.Connector.Data {
+  class Mosquitto : ADataBackend, IDisposable {
+    private Process p;
+    private String message;
+
+    public override event MqttMessage MessageIncomming;
+    public override event MqttMessage MessageSending;
+
+    public Mosquitto(Dictionary<String, String> mqtt_settings) {
+      this.settings = mqtt_settings;
+      //mosquitto_sub --cafile ca.pem --cert cert.pem --key cert.key -h swb.broker.flex4grid.eu -p 8883 -t "#" -v -d
+      this.message = "";
+      this.p = new Process();
+      this.p.StartInfo.FileName = "mosquitto_sub";
+      String args = "-h " + this.settings["server"]+" ";
+      if(this.settings.ContainsKey("port")) {
+        args += "-p "+ this.settings["port"]+" ";
+      }
+      if (this.settings.ContainsKey("cafile")) {
+        args += "--cafile " + this.settings["cafile"] + " ";
+      }
+      if (this.settings.ContainsKey("cert")) {
+        args += "--cert " + this.settings["cert"] + " ";
+      }
+      if (this.settings.ContainsKey("key")) {
+        args += "--key " + this.settings["key"] + " ";
+      }
+      this.p.StartInfo.Arguments = args+"-t \"#\" -v -d";
+      this.p.StartInfo.CreateNoWindow = true;
+      this.p.StartInfo.UseShellExecute = false;
+      this.p.StartInfo.RedirectStandardOutput = true;
+      this.p.StartInfo.RedirectStandardError = true;
+      this.p.OutputDataReceived += this.P_OutputDataReceived;
+      this.p.ErrorDataReceived += this.P_ErrorDataReceived;
+      this.p.Start();
+      this.p.BeginOutputReadLine();
+      
+    }
+
+    public override void Send(String topic, String data) {
+      Process send = new Process();
+      send.StartInfo.FileName = "mosquitto_pub";
+      String args = "-h " + this.settings["server"] + " ";
+      if (this.settings.ContainsKey("port")) {
+        args += "-p " + this.settings["port"] + " ";
+      }
+      if (this.settings.ContainsKey("cafile")) {
+        args += "--cafile " + this.settings["cafile"] + " ";
+      }
+      if (this.settings.ContainsKey("cert")) {
+        args += "--cert " + this.settings["cert"] + " ";
+      }
+      if (this.settings.ContainsKey("key")) {
+        args += "--key " + this.settings["key"] + " ";
+      }
+      send.StartInfo.Arguments = args + "-m \""+data.Replace("\"","\\\"")+"\" -t \""+topic+"\" -d";
+      send.StartInfo.CreateNoWindow = true;
+      send.StartInfo.UseShellExecute = false;
+      send.StartInfo.RedirectStandardOutput = true;
+      send.StartInfo.RedirectStandardError = true;
+      send.Start();
+      send.WaitForExit();
+      MessageSending?.Invoke(this, new MqttEventArgs(data, topic));
+    }
+
+    private void P_ErrorDataReceived(Object sender, DataReceivedEventArgs e) {
+      if (e.Data != null) {
+        throw new NotImplementedException(e.Data);
+      }
+    }
+
+    private void P_OutputDataReceived(Object sender, DataReceivedEventArgs e) {
+      if (e.Data != null) {
+        if (e.Data.StartsWith("Client mosqsub")) {
+          if (this.message != "" && this.message.IndexOf(" received PUBLISH ") > 0) {
+            MatchCollection matches = (new Regex("^Client mosqsub[\\|/].*received PUBLISH \\(.*,.*,.*,.*, '(.*)'.*\\)\\)\n[^ ]* (.*)$", RegexOptions.IgnoreCase | RegexOptions.Singleline)).Matches(this.message);
+            String topic = matches[0].Groups[1].Value;
+            String message = matches[0].Groups[2].Value.Trim();
+            this.MessageIncomming?.Invoke(this, new MqttEventArgs(message, topic));
+          }
+          this.message = e.Data + "\n";
+        } else {
+          this.message += e.Data + "\n";
+        }
+      }
+    }
+
+    #region IDisposable Support
+    private Boolean disposedValue = false; // Dient zur Erkennung redundanter Aufrufe.
+    private readonly Dictionary<String, String> settings;
+
+    protected virtual void Dispose(Boolean disposing) {
+      if (!this.disposedValue) {
+        if (disposing) {
+          this.p.CancelOutputRead();
+          if (!this.p.HasExited) {
+            this.p.Kill();
+          }
+          this.p.Close();
+        }
+        this.p = null;
+        this.disposedValue = true;
+      }
+    }
+
+    ~Mosquitto() {
+      Dispose(false);
+    }
+
+    public override void Dispose() {
+      Dispose(true);
+      GC.SuppressFinalize(this);
+    }
+    #endregion
+  }
+}
@@ -0,0 +1,65 @@
+using System;
+using System.Collections.Generic;
+using System.Text;
+using uPLibrary.Networking.M2Mqtt;
+using uPLibrary.Networking.M2Mqtt.Messages;
+
+namespace BlubbFish.Utils.IoT.Connector.Data {
+  class Mqtt : ADataBackend, IDisposable {
+    private MqttClient client;
+
+    public override event MqttMessage MessageIncomming;
+    public override event MqttMessage MessageSending;
+
+    public Mqtt(Dictionary<String, String> settings) {
+      if(settings.ContainsKey("port")) {
+        this.client = new MqttClient(settings["server"], Int32.Parse(settings["port"]), false, null, null, MqttSslProtocols.None);
+      } else {
+        this.client = new MqttClient(settings["server"]);
+      }
+      Connect();
+    }
+
+    private void Connect() {
+      this.client.MqttMsgPublishReceived += this.Client_MqttMsgPublishReceived;
+      this.client.Connect(Guid.NewGuid().ToString());
+      this.client.Subscribe(new String[] { "#" }, new Byte[] { MqttMsgBase.QOS_LEVEL_AT_LEAST_ONCE });
+    }
+
+    private void Client_MqttMsgPublishReceived(Object sender, MqttMsgPublishEventArgs e) {
+      this.MessageIncomming?.Invoke(this, new MqttEventArgs(Encoding.UTF8.GetString(e.Message), e.Topic));
+    }
+
+    public override void Send(String topic, String data) {
+      this.client.Publish(topic, Encoding.UTF8.GetBytes(data));
+      this.MessageSending?.Invoke(this, new MqttEventArgs(data, topic));
+    }
+
+    #region IDisposable Support
+    private Boolean disposedValue = false;
+
+    
+
+    protected virtual void Dispose(Boolean disposing) {
+      if(!this.disposedValue) {
+        if(disposing) {
+          this.client.MqttMsgPublishReceived -= this.Client_MqttMsgPublishReceived;
+          this.client.Unsubscribe(new String[] { "#" });
+          this.client.Disconnect();
+        }
+
+        this.client = null;
+
+        this.disposedValue = true;
+      }
+    }
+    ~Mqtt() {
+      Dispose(false);
+    }
+    public override void Dispose() {
+      Dispose(true);
+      GC.SuppressFinalize(this);
+    }
+    #endregion
+  }
+}