feat(lmxproxy): phase 5 — client core (ILmxProxyClient, connection, read/write/subscribe)

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-03-22 00:22:29 -04:00
parent 9eb81180c0
commit 8ba75b50e8
19 changed files with 1819 additions and 0 deletions
--- a/lmxproxy/tests/ZB.MOM.WW.LmxProxy.Client.Tests/LmxProxyClientSubscriptionTests.cs
+++ b/lmxproxy/tests/ZB.MOM.WW.LmxProxy.Client.Tests/LmxProxyClientSubscriptionTests.cs
@@ -0,0 +1,100 @@
+using FluentAssertions;
+using Xunit;
+using ZB.MOM.WW.LmxProxy.Client.Domain;
+using ZB.MOM.WW.LmxProxy.Client.Tests.Fakes;
+
+namespace ZB.MOM.WW.LmxProxy.Client.Tests;
+
+public class LmxProxyClientSubscriptionTests
+{
+    [Fact]
+    public async Task SubscribeAsync_InvokesCallbackForEachUpdate()
+    {
+        var (client, fake) = TestableClient.CreateConnected();
+        fake.SubscriptionMessages =
+        [
+            new VtqMessage
+            {
+                Tag = "Tag1",
+                Value = new TypedValue { DoubleValue = 1.0 },
+                TimestampUtcTicks = DateTime.UtcNow.Ticks,
+                Quality = new QualityCode { StatusCode = 0x00000000 }
+            },
+            new VtqMessage
+            {
+                Tag = "Tag2",
+                Value = new TypedValue { Int32Value = 42 },
+                TimestampUtcTicks = DateTime.UtcNow.Ticks,
+                Quality = new QualityCode { StatusCode = 0x00000000 }
+            }
+        ];
+
+        var updates = new List<(string Tag, Vtq Vtq)>();
+        var subscription = await client.SubscribeAsync(
+            ["Tag1", "Tag2"],
+            (tag, vtq) => updates.Add((tag, vtq)));
+
+        // Wait for processing to complete (fake yields all then stops)
+        await Task.Delay(500);
+
+        updates.Should().HaveCount(2);
+        updates[0].Tag.Should().Be("Tag1");
+        updates[0].Vtq.Value.Should().Be(1.0);
+        updates[1].Tag.Should().Be("Tag2");
+        updates[1].Vtq.Value.Should().Be(42);
+
+        subscription.Dispose();
+        client.Dispose();
+    }
+
+    [Fact]
+    public async Task SubscribeAsync_InvokesStreamErrorOnFailure()
+    {
+        var (client, fake) = TestableClient.CreateConnected();
+        fake.SubscriptionException = new InvalidOperationException("Stream broke");
+
+        Exception? capturedError = null;
+        var subscription = await client.SubscribeAsync(
+            ["Tag1"],
+            (_, _) => { },
+            ex => capturedError = ex);
+
+        // Wait for error to propagate
+        await Task.Delay(500);
+
+        capturedError.Should().NotBeNull();
+        capturedError.Should().BeOfType<InvalidOperationException>();
+        capturedError!.Message.Should().Be("Stream broke");
+
+        subscription.Dispose();
+        client.Dispose();
+    }
+
+    [Fact]
+    public async Task SubscribeAsync_DisposeStopsProcessing()
+    {
+        var (client, fake) = TestableClient.CreateConnected();
+        // Provide many messages but we'll dispose early
+        fake.SubscriptionMessages =
+        [
+            new VtqMessage
+            {
+                Tag = "Tag1",
+                Value = new TypedValue { DoubleValue = 1.0 },
+                TimestampUtcTicks = DateTime.UtcNow.Ticks,
+                Quality = new QualityCode { StatusCode = 0x00000000 }
+            }
+        ];
+
+        var updates = new List<(string Tag, Vtq Vtq)>();
+        var subscription = await client.SubscribeAsync(
+            ["Tag1"],
+            (tag, vtq) => updates.Add((tag, vtq)));
+
+        // Dispose immediately
+        subscription.Dispose();
+
+        // Should not throw
+        client.Dispose();
+    }
+}