Separate transformation from transport logic

AllMyLights.Test/AllMyLights.Test.csproj

@@ -25,4 +25,8 @@
   <ItemGroup>
     <ProjectReference Include="..\AllMyLights\AllMyLights.csproj" />
   </ItemGroup>
+  <ItemGroup>
+    <Folder Include="Extensions\" />
+    <Folder Include="Transformations\" />
+  </ItemGroup>
 </Project>

AllMyLights.Test/BrokerTest.cs

@@ -1,12 +1,8 @@
 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
 {
@@ -21,9 +17,9 @@ public class BrokerTest : ReactiveTest
         [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 red = System.Drawing.Color.FromName("red") as object;
+            var black = System.Drawing.Color.FromName("black") as object;
+            var green = System.Drawing.Color.FromName("green") as object;
 
             var broker = new Broker()
                 .RegisterSources(MqttSource.Object, SSESource.Object)

AllMyLights.Test/Connectors/MqttSourceTest.cs

@@ -1,7 +1,6 @@
 using System;
 using System.Text;
 using System.Threading;
-using AllMyLights.Models;
 using Moq;
 using MQTTnet;
 using MQTTnet.Client;
@@ -19,12 +18,14 @@
 using MQTTnet.Client.Disconnecting;
 using AllMyLights.Connectors.Sources;
 using AllMyLights.Models.Mqtt;
+using AllMyLights.Models.Transformations;
+using AllMyLights.Common;
 
 namespace AllMyLights.Test
 {
-    public class MqttSourceTest : ReactiveTest
+    public class MqttSourceTest: ReactiveTest
     {
-        MqttSourceParams Options;
+        MqttSourceOptions Options;
 
         public IMqttClientOptions MqttClientOptions { get; private set; }
         public MqttClientTcpOptions MqttClientTcpOptions { get; private set; }
@@ -34,7 +35,7 @@ public class MqttSourceTest : ReactiveTest
         [SetUp]
         public void Setup()
         {
-            Options = new MqttSourceParams
+            Options = new MqttSourceOptions
             {
                 Server = "wayne-foundation.com",
                 Port = 1863,
@@ -43,11 +44,7 @@ public void Setup()
                 Topics = new Topics
                 {
                     Command = "cmnd/tasmota-dimmer/color",
-                    Result = new Topic
-                    {
-                        Path = "stat/sonoff-1144-dimmer-5/RESULT",
-                        ValuePath = "$.Color"
-                    }
+                    Result = "stat/sonoff-1144-dimmer-5/RESULT",
                 }
             };
 
@@ -78,7 +75,7 @@ public void Should_initialize_MQTTClient()
         }
 
         [Test]
-        public void Should_request_color_via_command_topic()
+        public void Should_request_value_via_command_topic()
         {
             var args = new List<MqttApplicationMessage>();
             MqttClientMock.Setup(it => it.PublishAsync(Capture.In(args), CancellationToken.None));
@@ -101,7 +98,7 @@ public void Should_subscribe_to_provided_topic()
             var filter = args.First().TopicFilters.First();
 
             MqttClientMock.Verify(it => it.SubscribeAsync(It.IsAny<MqttClientSubscribeOptions>(), CancellationToken.None));
-            Assert.AreEqual(Options.Topics.Result.Path, filter.Topic);
+            Assert.AreEqual(Options.Topics.Result, filter.Topic);
         }
 
         [Test]
@@ -133,12 +130,12 @@ public void Should_reconnect_after_disconnecting()
         }
 
         [Test]
-        public void Should_consume_message_and_emit_color()
+        public void Should_consume_message_and_emit_payload()
         {
             var black = "#000000";
-            Color expectedColor = Color.FromArgb(255, 0, 0, 0);
 
-            var message = new MqttApplicationMessage { Payload = Encoding.UTF8.GetBytes($"{{ \"Color\": \"{black}\" }}") };
+            string payload = $"{{ \"Color\": \"{black}\" }}";
+            var message = new MqttApplicationMessage { Payload = Encoding.UTF8.GetBytes(payload) };
             var eventArgs = new MqttApplicationMessageReceivedEventArgs("", message);
             var scheduler = new TestScheduler();
 
@@ -158,8 +155,49 @@ public void Should_consume_message_and_emit_color()
               disposed: 100
             );
 
-            var expected = new Recorded<Notification<Color>>[] {
-                OnNext(20, expectedColor)
+            var expected = new Recorded<Notification<object>>[] {
+                OnNext(20, (object)payload)
+            };
+
+            ReactiveAssert.AreElementsEqual(expected, actual.Messages);
+        }
+
+
+        [Test]
+        public void Should_consume_message_and_apply_transformations()
+        {
+            var red = "#ff0000";
+            Color expectedColor = Color.FromArgb(255, 0, 255, 0);
+
+            string payload = $"{{ \"Color\": \"{red}\" }}";
+            var message = new MqttApplicationMessage { Payload = Encoding.UTF8.GetBytes(payload) };
+            var eventArgs = new MqttApplicationMessageReceivedEventArgs("", message);
+            var scheduler = new TestScheduler();
+
+            MqttClientMock.SetupAllProperties();
+
+            Options.Transformations = new List<TransformationOptions>()
+            {
+                new JsonPathTransformationOptions() { Expression = "$.Color" },
+                new ColorTransformationOptions() { ChannelLayout = "GRB" },
+            }.ToArray();
+
+            var subject = new MqttSource(Options, MqttClientMock.Object);
+
+            scheduler.Schedule(TimeSpan.FromTicks(20), () =>
+            {
+                MqttClientMock.Object.ApplicationMessageReceivedHandler.HandleApplicationMessageReceivedAsync(eventArgs);
+            });
+
+            var actual = scheduler.Start(
+              () => subject.Get(),
+              created: 0,
+              subscribed: 10,
+              disposed: 100
+            );
+
+            var expected = new Recorded<Notification<object>>[] {
+                OnNext(20, (object)new Ref<Color>(expectedColor))
             };
 
             ReactiveAssert.AreElementsEqual(expected, actual.Messages);

AllMyLights.Test/Connectors/OpenRGBSinkTest.cs

@@ -1,4 +1,5 @@
 using System.Collections.Generic;
+using AllMyLights.Common;
 using AllMyLights.Connectors.Sinks;
 using AllMyLights.Models.OpenRGB;
 using Moq;
@@ -10,7 +11,7 @@ namespace AllMyLights.Test
 {
     public class OpenRGBSinkTest
     {
-        OpenRGBSinkParams Options = new OpenRGBSinkParams()
+        OpenRGBSinkOptions Options = new OpenRGBSinkOptions()
         {
             Overrides = new Dictionary<string, DeviceOverride>()
                     {
@@ -54,7 +55,7 @@ public void Should_update_devices_with_color()
 
 
             var client = new OpenRGBSink(Options, openRGBClientMock.Object);
-            client.Consume(targetColor);
+            client.Consume(new Ref<System.Drawing.Color>(targetColor));
 
             openRGBClientMock.Verify();
         }
@@ -83,7 +84,7 @@ public void Should_update_devices_with_channel_layout_defined_in_override()
 
 
             var client = new OpenRGBSink(Options, openRGBClientMock.Object);
-            client.Consume(targetColor);
+            client.Consume(new Ref<System.Drawing.Color>(targetColor));
 
             openRGBClientMock.Verify();
         }
@@ -104,7 +105,7 @@ public void Should_not_update_devices_with_ignore_true()
 
 
             var client = new OpenRGBSink(Options, openRGBClientMock.Object);
-            client.Consume(targetColor);
+            client.Consume(new Ref<System.Drawing.Color>(targetColor));
 
             openRGBClientMock.Verify();
         }
@@ -144,7 +145,7 @@ public void Should_update_zones_with_channel_layout_defined_in_override()
 
 
             var client = new OpenRGBSink(Options, openRGBClientMock.Object);
-            client.Consume(targetColor);
+            client.Consume(new Ref<System.Drawing.Color>(targetColor));
 
             openRGBClientMock.Verify();
         }

AllMyLights.Test/Extensions/ObservableExtensionsTest.cs

@@ -0,0 +1,9 @@
+using Microsoft.Reactive.Testing;
+
+namespace AllMyLights.Test
+{
+    public class ObservableExtensionsTest : ReactiveTest
+    {
+
+    }
+}

AllMyLights.Test/Transformations/ColorTransformationTest.cs

@@ -0,0 +1,97 @@
+using System;
+using System.Drawing;
+using System.Linq;
+using System.Reactive;
+using System.Reactive.Linq;
+using AllMyLights.Common;
+using AllMyLights.Models.Transformations;
+using AllMyLights.Transformations;
+using Microsoft.Reactive.Testing;
+using NUnit.Framework;
+
+namespace AllMyLights.Test
+{
+    public class ColorTransformationTest : ReactiveTest
+    {
+
+        [Test]
+        public void Should_transform_color()
+        {
+            var black = "#000000";
+            Ref<Color> expectedColor = new Ref<Color>(Color.FromArgb(255, 0, 0, 0));
+            var source = Observable.Return(black);
+
+            var transformation = new ColorTransformation(new ColorTransformationOptions());
+
+            var scheduler = new TestScheduler();
+
+
+            var actual = scheduler.Start(
+              () => transformation.GetOperator()(source),
+              created: 0,
+              subscribed: 10,
+              disposed: 100
+            );
+
+            var expected = new Recorded<Notification<Ref<Color>>>[] {
+                OnNext(10, expectedColor),
+                OnCompleted<Ref<Color>>(10)
+            };
+
+            ReactiveAssert.AreElementsEqual(expected, actual.Messages);
+        }
+
+        [Test]
+        public void Should_transform_color_using_given_channel_layout()
+        {
+            var black = "#FF0000";
+            var expectedColor = new Ref<Color>(Color.FromArgb(255, 0, 255, 0));
+            var source = Observable.Return(black);
+
+            var transformation = new ColorTransformation(new ColorTransformationOptions()
+            {
+                ChannelLayout = "GRB"
+            });
+
+            var scheduler = new TestScheduler();
+
+
+            var actual = scheduler.Start(
+              () => transformation.GetOperator()(source),
+              created: 0,
+              subscribed: 10,
+              disposed: 100
+            );
+
+            var expected = new Recorded<Notification<Ref<Color>>>[] {
+                OnNext(10, expectedColor),
+                OnCompleted<Ref<Color>>(10)
+            };
+
+            ReactiveAssert.AreElementsEqual(expected, actual.Messages);
+        }
+
+        [Test]
+        public void Should_throw_argument_exception_for_invalid_input()
+        {
+            var source = Observable.Return(new object());
+
+            var transformation = new ColorTransformation(new ColorTransformationOptions());
+
+            var scheduler = new TestScheduler();
+
+
+            var actual = scheduler.Start(
+              () => transformation.GetOperator()(source),
+              created: 0,
+              subscribed: 10,
+              disposed: 100
+            );
+
+
+            Recorded<Notification<Ref<Color>>> first = actual.Messages.First();
+            Equals(NotificationKind.OnError, first.Value.Kind);
+            Equals(typeof(ArgumentException), first.Value.Exception.GetType());
+        }
+    }
+}

AllMyLights.Test/Transformations/JsonPathTransformationTest.cs

@@ -0,0 +1,67 @@
+using System;
+using System.Linq;
+using System.Reactive;
+using System.Reactive.Linq;
+using AllMyLights.Models.Transformations;
+using AllMyLights.Transformations;
+using Microsoft.Reactive.Testing;
+using NUnit.Framework;
+
+namespace AllMyLights.Test
+{
+    public class JsonPathTransformationTest : ReactiveTest
+    {
+
+        [Test]
+        public void Should_extract_deeply_nested_value()
+        {
+            var json = "{\"data\":{\"livingRoom\":{\"color\": \"red\"}}}";
+            var source = Observable.Return(json);
+
+            var options = new JsonPathTransformationOptions() { Expression = "$.data.livingRoom.color" };
+            var transformation = new JsonPathTransformation<string>(options);
+
+            var scheduler = new TestScheduler();
+
+
+            var actual = scheduler.Start(
+              () => transformation.GetOperator()(source),
+              created: 0,
+              subscribed: 10,
+              disposed: 100
+            );
+
+            var expected = new Recorded<Notification<string>>[] {
+                OnNext(10, "red"),
+                OnCompleted<string>(10)
+            };
+
+            ReactiveAssert.AreElementsEqual(expected, actual.Messages);
+        }
+
+        [Test]
+        public void Should_throw_argument_exception_for_invalid_input()
+        {
+            var source = Observable.Return(new object());
+
+            var options = new JsonPathTransformationOptions() { Expression = "$.data.livingRoom.color" };
+            var transformation = new JsonPathTransformation<string>(options);
+
+            var scheduler = new TestScheduler();
+
+
+            var actual = scheduler.Start(
+              () => transformation.GetOperator()(source),
+              created: 0,
+              subscribed: 0,
+              disposed: 100
+            );
+
+
+
+            Recorded<Notification<string>> first = actual.Messages.First();
+            Equals(NotificationKind.OnError, first.Value.Kind);
+            Equals(typeof(ArgumentException), first.Value.Exception.GetType());
+        }
+    }
+}