Add a relayOn option to FW messages to another XMR instance, tone dow…

…n default logging (#41)

Program.cs

@@ -18,9 +18,6 @@
  * You should have received a copy of the GNU Affero General Public License
  * along with Xibo.  If not, see <http://www.gnu.org/licenses/>.
  */
-using Microsoft.Extensions.Configuration;
-using Microsoft.Extensions.DependencyInjection;
-using Microsoft.Extensions.Hosting;
 using xibo_xmr;
 
 IHost host = Host.CreateDefaultBuilder(args)

Worker.cs

@@ -19,11 +19,8 @@
  * along with Xibo.  If not, see <http://www.gnu.org/licenses/>.
  */
 using System.Collections.Concurrent;
-using System.Linq.Expressions;
 using System.Text;
 using ConcurrentPriorityQueue.Core;
-using Microsoft.Extensions.Hosting;
-using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
 using NetMQ;
 using NetMQ.Sockets;
@@ -38,13 +35,16 @@ public class Worker : BackgroundService
 
     private readonly ConcurrentPriorityQueue<ZmqMessage, int> _queue;
 
+    private readonly BlockingCollection<string> _relayQueue;
+
     private int _sentCount = 0;
 
     public Worker(ILogger<Worker> logger, IOptions<ZmqSettings> settings)
     {
         _logger = logger;
         _settings = settings.Value;
         _queue = new();
+        _relayQueue = new();
     }
 
     protected override async Task ExecuteAsync(CancellationToken stoppingToken)
@@ -69,12 +69,23 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken)
         //    messages arriving from the CMS and adds them to the queue with the right QoS
         // 2. Set up a Publisher (PUB) socket bound to `pubOn` which processes the queue
         // 3. Set up a periodic timer which sends a heartbeat message (H) every 30 seconds
+        // 4. Handle relay if set
         // -------
-        await Task.WhenAll(
+        List<Task> tasks = new()
+        {
             Task.Factory.StartNew(() => { new NetMQRuntime().Run(stoppingToken, ResponderAsync(stoppingToken)); }, stoppingToken, TaskCreationOptions.LongRunning, TaskScheduler.Default),
             Task.Factory.StartNew(() => { new NetMQRuntime().Run(stoppingToken, PublisherAsync(stoppingToken)); }, stoppingToken, TaskCreationOptions.LongRunning, TaskScheduler.Default),
             Task.Factory.StartNew(() => { new NetMQRuntime().Run(stoppingToken, HeartbeatAsync(stoppingToken)); }, stoppingToken, TaskCreationOptions.LongRunning, TaskScheduler.Default)
-        );
+        };
+
+        // Do we relay?
+        if (!string.IsNullOrEmpty(_settings.relayOn))
+        {
+            tasks.Add(Task.Factory.StartNew(() => { new NetMQRuntime().Run(stoppingToken, RelayAsync(stoppingToken)); }, stoppingToken, TaskCreationOptions.LongRunning, TaskScheduler.Default));
+        }
+
+        // Await all
+        await Task.WhenAll(tasks);
 
         // Must call clean up at the end
         NetMQConfig.Cleanup();
@@ -111,6 +122,16 @@ async Task ResponderAsync(CancellationToken stoppingToken)
             }
             else
             {
+                // Relay
+                if (!string.IsNullOrEmpty(_settings.relayOn))
+                {
+                    bool relayResult = _relayQueue.TryAdd(message);
+                    if (!relayResult)
+                    {
+                        _logger.LogError("Failed to add message to the relay queue");
+                    }
+                }
+
                 // Decode the message
                 try
                 {
@@ -235,6 +256,27 @@ async Task HeartbeatAsync(CancellationToken stoppingToken)
         }
     }
 
+    async Task RelayAsync(CancellationToken stoppingToken)
+    {
+        using var relaySocket = new RequestSocket(_settings.relayOn);
+
+        while (!stoppingToken.IsCancellationRequested)
+        {
+            await Task.Run(() => {
+                bool result = _relayQueue.TryTake(out string message, -1, stoppingToken);
+                if (result && !string.IsNullOrEmpty(message))
+                {
+                    _logger.LogDebug("Relay message");
+                    bool sendResult = relaySocket.TrySendFrame(message);
+                    if (!sendResult)
+                    {
+                        _logger.LogError("Unable to relay message");
+                    }
+                }
+            }, stoppingToken);
+        }
+    }
+
     private static Dictionary<string, int> NewStats()
     {
         return new()

ZmqSettings.cs

@@ -28,4 +28,5 @@ public class ZmqSettings
     public bool ipv6RespSupport { get; set; }
     public bool ipv6PubSupport { get; set; }
     public int? pubSendTimeoutMs { get; set; }
+    public string? relayOn {get; set; }
 }

appsettings.json

@@ -1,7 +1,7 @@
 {
   "Logging": {
     "LogLevel": {
-      "Default": "Debug",
+      "Default": "Information",
       "Microsoft.Hosting.Lifetime": "Information"
     },
     "Console": {
@@ -20,6 +20,7 @@
     "queueSize": 10,
     "ipv6RespSupport": false,
     "ipv6PubSupport": false,
-    "pubSendTimeoutMs": 500
+    "pubSendTimeoutMs": 500,
+    "relayOn": ""
   }
 }