refactor(device): extract the session snapshot/restore decision into a collaborator (part of #344)

src/Daqifi.Core.Tests/Device/Internal/StreamingSessionSnapshotTests.cs

@@ -0,0 +1,186 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using Daqifi.Core.Channel;
+using Daqifi.Core.Device;
+using Daqifi.Core.Device.Internal;
+using Xunit;
+
+namespace Daqifi.Core.Tests.Device.Internal;
+
+/// <summary>
+/// Unit tests for <see cref="StreamingSessionSnapshot"/>, which records what a streaming session
+/// looked like at a drop and decides what putting it back implies (issue #379). These pin the
+/// decision itself; <c>DeviceReconnectTests</c> pins the effects the device applies from it.
+/// </summary>
+public class StreamingSessionSnapshotTests
+{
+    private static AnalogChannel Analog(int number, bool enabled)
+    {
+        var channel = new AnalogChannel(number);
+        channel.IsEnabled = enabled;
+        return channel;
+    }
+
+    private static DigitalChannel Digital(int number, bool enabled)
+    {
+        var channel = new DigitalChannel(number);
+        channel.IsEnabled = enabled;
+        return channel;
+    }
+
+    private static ReconnectOptions Policy(bool resumeStreaming = true) =>
+        new() { Enabled = true, ResumeStreaming = resumeStreaming };
+
+    [Fact]
+    public void Capture_RecordsOnlyTheEnabledChannels()
+    {
+        var channels = new IChannel[]
+        {
+            Analog(0, enabled: true),
+            Analog(1, enabled: false),
+            Digital(2, enabled: true),
+        };
+
+        var snapshot = StreamingSessionSnapshot.Capture(channels, isStreaming: true);
+
+        Assert.Equal(2, snapshot.EnabledChannelCount);
+        Assert.True(snapshot.WasStreaming);
+    }
+
+    [Fact]
+    public void Capture_TakesACopy_SoLaterChannelMutationCannotRewriteTheSession()
+    {
+        // This is the whole reason a snapshot exists: re-initialization after a reconnect goes on
+        // mutating these same channel objects (analog IsEnabled is resynced from the device's own
+        // reported mask on every status frame, #409), so a snapshot that held them by reference
+        // would describe the reconnected device rather than the session that was lost.
+        var enabled = Analog(0, enabled: true);
+        var disabled = Analog(1, enabled: false);
+
+        var snapshot = StreamingSessionSnapshot.Capture(new IChannel[] { enabled, disabled }, isStreaming: true);
+
+        enabled.IsEnabled = false;
+        disabled.IsEnabled = true;
+
+        var plan = snapshot.PlanRestore(new IChannel[] { enabled, disabled }, Policy());
+
+        Assert.Equal(new[] { 0 }, plan.ChannelsToEnable.Select(c => c.ChannelNumber));
+    }
+
+    [Fact]
+    public void Capture_WithNoChannelsEnabled_PlansNothingToEnable()
+    {
+        var snapshot = StreamingSessionSnapshot.Capture(
+            new IChannel[] { Analog(0, enabled: false) }, isStreaming: false);
+
+        var plan = snapshot.PlanRestore(new IChannel[] { Analog(0, enabled: false) }, Policy());
+
+        Assert.Equal(0, snapshot.EnabledChannelCount);
+        Assert.Empty(plan.ChannelsToEnable);
+        Assert.False(plan.ResumeStreaming);
+    }
+
+    [Fact]
+    public void PlanRestore_ReturnsTheReconnectedDevicesOwnChannelObjects_NotTheCapturedOnes()
+    {
+        // A reconnect can replace the channel objects wholesale, so matching has to be by identity
+        // (type + number) and the plan has to hand back the objects the device has now — enabling
+        // the old ones would set a flag on garbage.
+        var beforeDrop = Analog(3, enabled: true);
+        var snapshot = StreamingSessionSnapshot.Capture(new IChannel[] { beforeDrop }, isStreaming: true);
+
+        var afterReconnect = Analog(3, enabled: false);
+        var plan = snapshot.PlanRestore(new IChannel[] { afterReconnect }, Policy());
+
+        Assert.Same(afterReconnect, Assert.Single(plan.ChannelsToEnable));
+    }
+
+    [Fact]
+    public void PlanRestore_MatchesOnChannelTypeAsWellAsNumber()
+    {
+        // Analog 0 and digital 0 share a number and are different channels; a number-only match
+        // would enable the wrong one.
+        var snapshot = StreamingSessionSnapshot.Capture(
+            new IChannel[] { Analog(0, enabled: true), Digital(0, enabled: false) }, isStreaming: true);
+
+        var plan = snapshot.PlanRestore(
+            new IChannel[] { Analog(0, enabled: false), Digital(0, enabled: false) }, Policy());
+
+        var restored = Assert.Single(plan.ChannelsToEnable);
+        Assert.Equal(ChannelType.Analog, restored.Type);
+    }
+
+    [Fact]
+    public void PlanRestore_WhenTheDeviceComesBackSmaller_RestoresTheIntersection()
+    {
+        var snapshot = StreamingSessionSnapshot.Capture(
+            new IChannel[] { Analog(0, enabled: true), Analog(1, enabled: true) }, isStreaming: true);
+
+        var plan = snapshot.PlanRestore(new IChannel[] { Analog(0, enabled: false) }, Policy());
+
+        Assert.Equal(new[] { 0 }, plan.ChannelsToEnable.Select(c => c.ChannelNumber));
+    }
+
+    [Fact]
+    public void PlanRestore_IgnoresChannelsTheSessionNeverHad()
+    {
+        var snapshot = StreamingSessionSnapshot.Capture(
+            new IChannel[] { Analog(0, enabled: true) }, isStreaming: false);
+
+        var plan = snapshot.PlanRestore(
+            new IChannel[] { Analog(0, enabled: false), Analog(7, enabled: true) }, Policy());
+
+        Assert.Equal(new[] { 0 }, plan.ChannelsToEnable.Select(c => c.ChannelNumber));
+    }
+
+    [Theory]
+    [InlineData(true, true, true)]
+    [InlineData(true, false, false)]
+    [InlineData(false, true, false)]
+    [InlineData(false, false, false)]
+    public void PlanRestore_ResumesOnlyWhenTheSessionWasStreamingAndThePolicyAllowsIt(
+        bool wasStreaming, bool resumeStreaming, bool expected)
+    {
+        var snapshot = StreamingSessionSnapshot.Capture(
+            new IChannel[] { Analog(0, enabled: true) }, wasStreaming);
+
+        var plan = snapshot.PlanRestore(
+            new IChannel[] { Analog(0, enabled: false) }, Policy(resumeStreaming));
+
+        Assert.Equal(expected, plan.ResumeStreaming);
+    }
+
+    [Fact]
+    public void PlanRestore_CanBeCalledMoreThanOnce_WithoutConsumingTheSnapshot()
+    {
+        // The device keeps one snapshot field across reconnect attempts, so a failed attempt must
+        // not leave a snapshot that has already given up its contents.
+        var snapshot = StreamingSessionSnapshot.Capture(
+            new IChannel[] { Analog(0, enabled: true) }, isStreaming: true);
+
+        var first = snapshot.PlanRestore(new IChannel[] { Analog(0, enabled: false) }, Policy());
+        var second = snapshot.PlanRestore(new IChannel[] { Analog(0, enabled: false) }, Policy());
+
+        Assert.Single(first.ChannelsToEnable);
+        Assert.Single(second.ChannelsToEnable);
+        Assert.True(second.ResumeStreaming);
+    }
+
+    [Fact]
+    public void Capture_WithNullChannels_Throws()
+    {
+        Assert.Throws<ArgumentNullException>(
+            () => StreamingSessionSnapshot.Capture(null!, isStreaming: false));
+    }
+
+    [Fact]
+    public void PlanRestore_WithNullArguments_Throws()
+    {
+        var snapshot = StreamingSessionSnapshot.Capture(Array.Empty<IChannel>(), isStreaming: false);
+
+        Assert.Throws<ArgumentNullException>(() => snapshot.PlanRestore(null!, Policy()));
+        Assert.Throws<ArgumentNullException>(
+            () => snapshot.PlanRestore(new List<IChannel>(), null!));
+    }
+}

src/Daqifi.Core/Device/DaqifiStreamingDevice.cs

@@ -514,59 +514,19 @@ private void ApplyAdcEnableMask(uint mask)
         /// </summary>
         private volatile StreamingSessionSnapshot? _sessionSnapshot;
 
-        /// <summary>
-        /// The subset of a streaming session that Core owns and can therefore put back: which
-        /// channels were enabled, and whether data was flowing.
-        /// </summary>
-        private sealed class StreamingSessionSnapshot
-        {
-            public StreamingSessionSnapshot(HashSet<(ChannelType Type, int Number)> enabledChannels, bool wasStreaming)
-            {
-                EnabledChannels = enabledChannels;
-                WasStreaming = wasStreaming;
-            }
-
-            /// <summary>
-            /// The enabled channels, held by identity rather than by reference: a reconnect can
-            /// replace the channel objects, and a device that came back with a different channel
-            /// count should restore the intersection rather than fail.
-            /// </summary>
-            public HashSet<(ChannelType Type, int Number)> EnabledChannels { get; }
-
-            public bool WasStreaming { get; }
-        }
-
         /// <inheritdoc />
-        protected override void CaptureSessionSnapshot()
-        {
-            var enabled = new HashSet<(ChannelType, int)>();
-            foreach (var channel in GetChannelsSnapshot())
-            {
-                if (channel.IsEnabled)
-                {
-                    enabled.Add((channel.Type, channel.ChannelNumber));
-                }
-            }
-
-            _sessionSnapshot = new StreamingSessionSnapshot(enabled, IsStreaming);
-        }
+        protected override void CaptureSessionSnapshot() =>
+            _sessionSnapshot = StreamingSessionSnapshot.Capture(GetChannelsSnapshot(), IsStreaming);
 
         /// <summary>
         /// Re-applies the enabled-channel set recorded at the drop and, if the policy says so,
         /// restarts a stream that was interrupted.
         /// </summary>
         /// <remarks>
         /// <para>
-        /// The enable set has to be replayed from the snapshot rather than read back off the
-        /// channel objects: <see cref="DaqifiDevice.PopulateChannelsFromStatus"/> resyncs analog
-        /// <c>IsEnabled</c> from the device's own enabled mask on every status message (#409), so by
-        /// the time re-initialization is done the in-memory view reflects the freshly reconnected
-        /// device, not the session that was lost.
-        /// </para>
-        /// <para>
-        /// The streaming frequency needs no replay — it is a host-side setting that the drop never
-        /// touched — but it does have to reach the device again, which is what the resumed
-        /// <see cref="StartStreaming"/> does.
+        /// What to restore is <see cref="StreamingSessionSnapshot.PlanRestore"/>'s decision; this
+        /// method owns the effects, and their order is the part that matters. See that method for
+        /// why the enable set is replayed from the snapshot rather than read back off the channels.
         /// </para>
         /// <para>
         /// A resumed stream is a genuinely new session: timestamp reconstruction re-anchors and the
@@ -597,32 +557,24 @@ protected override Task<bool> RestoreSessionSnapshotAsync(
             cancellationToken.ThrowIfCancellationRequested();
 
             // Normalize to a known state before re-applying: whatever the device came back with is
-            // not necessarily what it had, and the enable commands are set-replace anyway.
+            // not necessarily what it had, and the enable commands are set-replace anyway. The
+            // channel list is read afterwards so the plan is built against the post-reset objects.
             DisableAllChannels();
 
-            var toEnable = new List<IChannel>();
-            foreach (var channel in GetChannelsSnapshot())
-            {
-                if (snapshot.EnabledChannels.Contains((channel.Type, channel.ChannelNumber)))
-                {
-                    toEnable.Add(channel);
-                }
-            }
-
-            if (toEnable.Count > 0)
+            var plan = snapshot.PlanRestore(GetChannelsSnapshot(), options);
+            if (plan.ChannelsToEnable.Count > 0)
             {
-                EnableChannels(toEnable);
+                EnableChannels(plan.ChannelsToEnable);
             }
 
             cancellationToken.ThrowIfCancellationRequested();
 
-            var resumeStreaming = snapshot.WasStreaming && options.ResumeStreaming;
-            if (resumeStreaming)
+            if (plan.ResumeStreaming)
             {
                 StartStreaming();
             }
 
-            return Task.FromResult(resumeStreaming);
+            return Task.FromResult(plan.ResumeStreaming);
         }
 
         #endregion