Allow for multiple sources & sinks

AllMyLights.Test/AllMyLights.Test.csproj

@@ -1,4 +1,4 @@
-<Project Sdk="Microsoft.NET.Sdk">
+<Project Sdk="Microsoft.NET.Sdk">
 
   <PropertyGroup>
     <TargetFramework>netcoreapp3.1</TargetFramework>

AllMyLights.Test/BrokerTest.cs

@@ -0,0 +1,55 @@
+using AllMyLights.Connectors.Sinks;
+using AllMyLights.Connectors.Sources;
+using AllMyLights.Models;
+using Microsoft.Reactive.Testing;
+using Moq;
+using NUnit.Framework;
+using OpenRGB.NET;
+using OpenRGB.NET.Models;
+using Unmockable;
+
+namespace AllMyLights.Test
+{
+    public class BrokerTest : ReactiveTest
+    {
+        Mock<ISource> MqttSource = new Mock<ISource>();
+        Mock<ISource> SSESource = new Mock<ISource>();
+
+        Mock<ISink> OpenRGBSink = new Mock<ISink>();
+        Mock<ISink> WallpaperSink = new Mock<ISink>();
+
+        [Test]
+        public void Should_invoke_sinks_with_colors_emitted_from_sources()
+        {
+            var red = System.Drawing.Color.FromName("red");
+            var black = System.Drawing.Color.FromName("black");
+            var green = System.Drawing.Color.FromName("green");
+
+            var broker = new Broker()
+                .RegisterSources(MqttSource.Object, SSESource.Object)
+                .RegisterSinks(OpenRGBSink.Object, WallpaperSink.Object);
+
+            var scheduler = new TestScheduler();
+
+            MqttSource.Setup(it => it.Get()).Returns(() =>
+            {
+                return scheduler.CreateHotObservable(OnNext(150, red));
+            });
+
+            SSESource.Setup(it => it.Get()).Returns(() =>
+            {
+                return scheduler.CreateHotObservable(OnNext(100, black), OnNext(300, green));
+            });
+
+            broker.Listen();
+
+            scheduler.AdvanceBy(200);
+
+            OpenRGBSink.Verify((it) => it.Consume(black));
+            OpenRGBSink.Verify((it) => it.Consume(red));
+
+            WallpaperSink.Verify((it) => it.Consume(black));
+            WallpaperSink.Verify((it) => it.Consume(red));
+        }
+    }
+}

AllMyLights.Test/ColorSubjectTest.cs → AllMyLights.Test/Connectors/MqttSourceTest.cs

@@ -1,7 +1,6 @@
 using System;
 using System.Text;
 using System.Threading;
-using System.Threading.Tasks;
 using AllMyLights.Models;
 using Moq;
 using MQTTnet;
@@ -18,12 +17,14 @@
 using MQTTnet.Client.Connecting;
 using MQTTnet.Client.Subscribing;
 using MQTTnet.Client.Disconnecting;
+using AllMyLights.Connectors.Sources;
+using AllMyLights.Models.Mqtt;
 
 namespace AllMyLights.Test
 {
-    public class ColorSubjectTest : ReactiveTest
+    public class MqttSourceTest : ReactiveTest
     {
-        Configuration Config;
+        MqttSourceParams Options;
 
         public IMqttClientOptions MqttClientOptions { get; private set; }
         public MqttClientTcpOptions MqttClientTcpOptions { get; private set; }
@@ -33,29 +34,26 @@ public class ColorSubjectTest : ReactiveTest
         [SetUp]
         public void Setup()
         {
-            Config = new Configuration
+            Options = new MqttSourceParams
             {
-                Mqtt = new MqttConfiguration
+                Server = "wayne-foundation.com",
+                Port = 1863,
+                Password = "bruce-admires-robin",
+                Username = "bwayne",
+                Topics = new Topics
                 {
-                    Server = "wayne-foundation.com",
-                    Port = 1863,
-                    Password = "bruce-admires-robin",
-                    Username = "bwayne",
-                    Topics = new Topics
+                    Command = "cmnd/tasmota-dimmer/color",
+                    Result = new Topic
                     {
-                        Command = "cmnd/tasmota-dimmer/color",
-                        Result = new Topic
-                        {
-                            Path = "stat/sonoff-1144-dimmer-5/RESULT",
-                            ValuePath = "$.Color"
-                        }
+                        Path = "stat/sonoff-1144-dimmer-5/RESULT",
+                        ValuePath = "$.Color"
                     }
                 }
             };
 
             MqttClientOptions = new MqttClientOptionsBuilder()
-                .WithTcpServer(Config.Mqtt.Server, Config.Mqtt.Port)
-                .WithCredentials(Config.Mqtt.Username, Config.Mqtt.Password)
+                .WithTcpServer(Options.Server, Options.Port)
+                .WithCredentials(Options.Username, Options.Password)
                 .Build();
 
             MqttClientTcpOptions = MqttClientOptions.ChannelOptions as MqttClientTcpOptions;
@@ -67,7 +65,7 @@ public void Should_initialize_MQTTClient()
             var args = new List<MqttClientOptions>();
             MqttClientMock.Setup(m => m.ConnectAsync(Capture.In(args), CancellationToken.None));
 
-            var subject = new ColorSubject(Config, MqttClientMock.Object);
+            var subject = new MqttSource(Options, MqttClientMock.Object);
             var actualOptions = args.First();
             MqttClientTcpOptions actualChannelOptions = actualOptions.ChannelOptions as MqttClientTcpOptions;
 
@@ -85,25 +83,25 @@ public void Should_request_color_via_command_topic()
             var args = new List<MqttApplicationMessage>();
             MqttClientMock.Setup(it => it.PublishAsync(Capture.In(args), CancellationToken.None));
 
-            var subject = new ColorSubject(Config, MqttClientMock.Object);
+            var subject = new MqttSource(Options, MqttClientMock.Object);
 
             MqttClientMock.Verify(it => it.PublishAsync(It.IsAny<MqttApplicationMessage>(), CancellationToken.None));
-            Assert.AreEqual(Config.Mqtt.Topics.Command, args.First().Topic);
+            Assert.AreEqual(Options.Topics.Command, args.First().Topic);
         }
 
         [Test]
         public void Should_subscribe_to_provided_topic()
         {
             MqttClientMock.SetupAllProperties();
-            var subject = new ColorSubject(Config, MqttClientMock.Object);
+            var subject = new MqttSource(Options, MqttClientMock.Object);
 
             var args = new List<MqttClientSubscribeOptions>();
             MqttClientMock.Setup(it => it.SubscribeAsync(Capture.In(args), CancellationToken.None));
             MqttClientMock.Object.ConnectedHandler.HandleConnectedAsync(new MqttClientConnectedEventArgs(new Mock<MqttClientAuthenticateResult>().Object));
             var filter = args.First().TopicFilters.First();
 
             MqttClientMock.Verify(it => it.SubscribeAsync(It.IsAny<MqttClientSubscribeOptions>(), CancellationToken.None));
-            Assert.AreEqual(Config.Mqtt.Topics.Result.Path, filter.Topic);
+            Assert.AreEqual(Options.Topics.Result.Path, filter.Topic);
         }
 
         [Test]
@@ -114,7 +112,7 @@ public void Should_reconnect_after_disconnecting()
             var args = new List<MqttClientOptions>();
             MqttClientMock.Setup(it => it.ConnectAsync(Capture.In(args), CancellationToken.None));
 
-            var subject = new ColorSubject(Config, MqttClientMock.Object);
+            var subject = new MqttSource(Options, MqttClientMock.Object);
             MqttClientMock.Object.DisconnectedHandler.HandleDisconnectedAsync(new MqttClientDisconnectedEventArgs(
                 true,
                 new Exception("Networ error"),
@@ -146,7 +144,7 @@ public void Should_consume_message_and_emit_color()
 
             MqttClientMock.SetupAllProperties();
 
-            var subject = new ColorSubject(Config, MqttClientMock.Object);
+            var subject = new MqttSource(Options, MqttClientMock.Object);
 
             scheduler.Schedule(TimeSpan.FromTicks(20), () =>
             {

AllMyLights.Test/OpenRGBClientTest.cs → AllMyLights.Test/Connectors/OpenRGBSinkTest.cs

@@ -1,31 +1,30 @@
 using System.Collections.Generic;
 using System.Net.Sockets;
+using AllMyLights.Connectors.Sinks;
 using AllMyLights.Models;
+using AllMyLights.Models.OpenRGB;
 using NUnit.Framework;
 using OpenRGB.NET.Models;
 using Unmockable;
 
 namespace AllMyLights.Test
 {
-    public class OpenRGBClientTest
+    public class OpenRGBSinkTest
     {
-        Configuration Config = new Configuration
-            {
-                OpenRgb = new OpenRGBConfiguration()
-                {
-                    Overrides = new Dictionary<string, DeviceOverride>()
+        OpenRGBSinkParams Options = new OpenRGBSinkParams()
+        {
+            Overrides = new Dictionary<string, DeviceOverride>()
                     {
                         { "MSI X570 Tomahawk", new DeviceOverride() {  ChannelLayout = "RBG"} },
                         { "MSI B450 Gaming Pro", new DeviceOverride() {  Ignore = true } },
-                        { "MSI X570 Unify", new DeviceOverride() {  
+                        { "MSI X570 Unify", new DeviceOverride() {
                             Zones = new Dictionary<string, ZoneOverride>
                             {
                                 { "JRGB2",  new ZoneOverride() { ChannelLayout = "RBG" } }
                             }
                         } },
                     }
-                }
-            };
+        };
 
         [Test]
         public void Should_update_devices_with_color()
@@ -58,8 +57,8 @@ public void Should_update_devices_with_color()
                 ));
 
 
-            var client = new OpenRGBClient(openRGBClientMock, Config);
-            client.UpdateAll(targetColor);
+            var client = new OpenRGBSink(Options, openRGBClientMock);
+            client.Consume(targetColor);
 
             openRGBClientMock.Verify();
         }
@@ -89,8 +88,8 @@ public void Should_update_devices_with_channel_layout_defined_in_override()
                 ));
 
 
-            var client = new OpenRGBClient(openRGBClientMock, Config);
-            client.UpdateAll(targetColor);
+            var client = new OpenRGBSink(Options, openRGBClientMock);
+            client.Consume(targetColor);
 
             openRGBClientMock.Verify();
         }
@@ -113,8 +112,8 @@ public void Should_not_update_devices_with_ignore_true()
                 .Returns(new Device[] { gamingPro });
 
 
-            var client = new OpenRGBClient(openRGBClientMock, Config);
-            client.UpdateAll(targetColor);
+            var client = new OpenRGBSink(Options, openRGBClientMock);
+            client.Consume(targetColor);
 
             openRGBClientMock.Verify();
         }
@@ -156,8 +155,8 @@ public void Should_update_zones_with_channel_layout_defined_in_override()
                 ));
 
 
-            var client = new OpenRGBClient(openRGBClientMock, Config);
-            client.UpdateAll(targetColor);
+            var client = new OpenRGBSink(Options, openRGBClientMock);
+            client.Consume(targetColor);
 
             openRGBClientMock.Verify();
         }

AllMyLights/AllMyLights.csproj

@@ -4,7 +4,7 @@
         <OutputType>Exe</OutputType>
         <TargetFramework>netcoreapp3.1</TargetFramework>
         <ApplicationIcon>app.ico</ApplicationIcon>
-        <IsWindows Condition="$(RuntimeIdentifier.StartsWith('win'))">true</IsWindows>
+        <IsWindows Condition="$(RuntimeIdentifier.StartsWith('win')) Or ('$([System.Runtime.InteropServices.RuntimeInformation]::IsOSPlatform($([System.Runtime.InteropServices.OSPlatform]::Windows)))' == 'true' And $(RuntimeIdentifier.Length) == '0' )">true</IsWindows>
     </PropertyGroup>
     <PropertyGroup Condition="'$(IsWindows)'=='true'">
         <DefineConstants>Windows</DefineConstants>

AllMyLights/Broker.cs

@@ -0,0 +1,39 @@
+using AllMyLights.Connectors.Sinks;
+using AllMyLights.Connectors.Sources;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Reactive.Linq;
+
+namespace AllMyLights
+{
+    public class Broker
+    {
+
+        private List<ISink> Sinks { get; } = new List<ISink>();
+        private List<ISource> Sources { get; } = new List<ISource>();
+
+        public Broker RegisterSources(params ISource[] sources)
+        {
+            Sources.AddRange(sources);
+            return this;
+        }
+
+        public Broker RegisterSinks(params ISink[] sinks)
+        {
+            Sinks.AddRange(sinks);
+            return this;
+        }
+
+        public void Listen()
+        {
+            Observable
+                .Merge(Sources.Select((it) => it.Get()))
+                .Subscribe((color) =>
+            {
+                Sinks.ForEach((sink) => sink.Consume(color));
+            });
+
+        }
+    }
+}