refactor(device): extract the USB stream-interface routing decision into a collaborator (part of #344)

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

@@ -0,0 +1,208 @@
+using Daqifi.Core.Device;
+using Daqifi.Core.Device.Internal;
+using System;
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+using Xunit;
+
+namespace Daqifi.Core.Tests.Device.Internal;
+
+/// <summary>
+/// Unit tests for <see cref="UsbStreamInterfaceInitializer"/>, which decides whether a device being
+/// initialized must have its stream re-routed to USB and how hard to retry a transient rejection.
+/// These pin the decision itself; <c>DaqifiDeviceInitializeTests</c> pins what the device does with
+/// it end to end.
+/// </summary>
+public class UsbStreamInterfaceInitializerTests
+{
+    private const string ScpiError = "**ERROR: -200, \"Execution error\"";
+
+    /// <summary>Bound on every wait so a policy regression fails the run instead of parking it.</summary>
+    private static readonly TimeSpan RouteTimeout = TimeSpan.FromSeconds(5);
+
+    /// <summary>
+    /// Records every attempt and answers each one from a queued script, so a test states the
+    /// device's answers rather than the sender's mechanics.
+    /// </summary>
+    private sealed class ScriptedSender
+    {
+        private readonly Queue<IReadOnlyList<string>> _responses = new();
+
+        public int AttemptCount { get; private set; }
+
+        public List<CancellationToken> ObservedTokens { get; } = new();
+
+        public ScriptedSender Answers(params string[] lines)
+        {
+            _responses.Enqueue(lines);
+            return this;
+        }
+
+        public Task<IReadOnlyList<string>> SendAsync(CancellationToken cancellationToken)
+        {
+            AttemptCount++;
+            ObservedTokens.Add(cancellationToken);
+
+            // An unscripted attempt is a test bug, not a silent success: surface it.
+            Assert.True(_responses.Count > 0, "The sender was invoked more times than the test scripted.");
+            return Task.FromResult(_responses.Dequeue());
+        }
+    }
+
+    private static Task RouteAsync(
+        ScriptedSender sender,
+        bool isUsbConnection = true,
+        bool preserveActiveStream = false,
+        CancellationToken cancellationToken = default)
+        => UsbStreamInterfaceInitializer
+            .RouteStreamToUsbAsync(isUsbConnection, preserveActiveStream, sender.SendAsync, cancellationToken)
+            .WaitAsync(RouteTimeout);
+
+    [Theory]
+    [InlineData(true, false, true)]
+    [InlineData(true, true, false)]
+    [InlineData(false, false, false)]
+    [InlineData(false, true, false)]
+    public void OnlyAUsbSessionThatOwnsTheStream_Routes(
+        bool isUsbConnection,
+        bool preserveActiveStream,
+        bool expected)
+    {
+        // A non-USB connection has nothing to route, and an observe-only session must not steal the
+        // stream from the session already receiving it (#385).
+        Assert.Equal(
+            expected,
+            UsbStreamInterfaceInitializer.ShouldRouteStreamToUsb(isUsbConnection, preserveActiveStream));
+    }
+
+    [Fact]
+    public async Task ANonUsbConnection_SendsNothing()
+    {
+        var sender = new ScriptedSender();
+
+        await RouteAsync(sender, isUsbConnection: false);
+
+        Assert.Equal(0, sender.AttemptCount);
+    }
+
+    [Fact]
+    public async Task AnObserveOnlySession_SendsNothing()
+    {
+        var sender = new ScriptedSender();
+
+        await RouteAsync(sender, preserveActiveStream: true);
+
+        Assert.Equal(0, sender.AttemptCount);
+    }
+
+    [Fact]
+    public async Task ASkippedRoute_DoesNotObserveCancellation()
+    {
+        // Deliberate: there is no work to abandon, and the caller re-checks cancellation before it
+        // marks the device ready, so a token cancelled here is still honored — just not by throwing
+        // out of a step that did nothing.
+        using var cts = new CancellationTokenSource();
+        cts.Cancel();
+        var sender = new ScriptedSender();
+
+        await RouteAsync(sender, preserveActiveStream: true, cancellationToken: cts.Token);
+
+        Assert.Equal(0, sender.AttemptCount);
+    }
+
+    [Fact]
+    public async Task ACleanResponse_RoutesOnce()
+    {
+        var sender = new ScriptedSender().Answers("0");
+
+        await RouteAsync(sender);
+
+        Assert.Equal(1, sender.AttemptCount);
+    }
+
+    [Fact]
+    public async Task AnEmptyResponse_IsNotTreatedAsAnError()
+    {
+        // The firmware answers this command with nothing at all on the happy path, so "no lines"
+        // must mean success rather than an unclassifiable failure.
+        var sender = new ScriptedSender().Answers();
+
+        await RouteAsync(sender);
+
+        Assert.Equal(1, sender.AttemptCount);
+    }
+
+    [Fact]
+    public async Task ATransientRejection_IsRetriedAndSucceeds()
+    {
+        // The firmware persists the last-used stream interface across sessions and can reject the
+        // command right after connect; that is the case retrying exists for (#310).
+        var sender = new ScriptedSender()
+            .Answers(ScpiError)
+            .Answers("0");
+
+        await RouteAsync(sender);
+
+        Assert.Equal(2, sender.AttemptCount);
+    }
+
+    [Fact]
+    public async Task APersistentRejection_FailsAfterExhaustingRetries()
+    {
+        var sender = new ScriptedSender()
+            .Answers(ScpiError)
+            .Answers(ScpiError);
+
+        var ex = await Assert.ThrowsAsync<ScpiInitializationErrorException>(() => RouteAsync(sender));
+
+        // Literal rather than MaxRetries + 1: deriving it from the constant would make this test
+        // agree with any retry budget, including one silently reduced to zero.
+        Assert.Equal(2, sender.AttemptCount);
+        Assert.Equal(ScpiError, ex.LastScpiError);
+        Assert.Equal(new[] { ScpiError }, ex.RawDeviceResponse);
+    }
+
+    [Fact]
+    public async Task TheReportedFailure_IsTheLastAttemptsResponse()
+    {
+        // Reporting the first attempt's response would describe an error the device has already
+        // moved past — the caller needs the answer that actually ended the routing.
+        var sender = new ScriptedSender()
+            .Answers("ERROR: -113, \"Undefined header\"")
+            .Answers("noise", "  " + ScpiError + "  ");
+
+        var ex = await Assert.ThrowsAsync<ScpiInitializationErrorException>(() => RouteAsync(sender));
+
+        Assert.Equal(ScpiError, ex.LastScpiError);
+        Assert.Equal(new[] { "noise", "  " + ScpiError + "  " }, ex.RawDeviceResponse);
+    }
+
+    [Fact]
+    public async Task CancellationBetweenAttempts_StopsBeforeTheRetry()
+    {
+        // The settle delay is cancellable, so a token cancelled while waiting must abandon the retry
+        // rather than send a command into a session that is being torn down.
+        using var cts = new CancellationTokenSource();
+        var sender = new ScriptedSender().Answers(ScpiError);
+        cts.CancelAfter(TimeSpan.FromMilliseconds(10));
+
+        await Assert.ThrowsAnyAsync<OperationCanceledException>(
+            () => RouteAsync(sender, cancellationToken: cts.Token));
+
+        Assert.Equal(1, sender.AttemptCount);
+    }
+
+    [Fact]
+    public async Task TheCallersToken_ReachesEveryAttempt()
+    {
+        using var cts = new CancellationTokenSource();
+        var sender = new ScriptedSender()
+            .Answers(ScpiError)
+            .Answers("0");
+
+        await RouteAsync(sender, cancellationToken: cts.Token);
+
+        Assert.Equal(new[] { cts.Token, cts.Token }, sender.ObservedTokens);
+    }
+}

src/Daqifi.Core/Device/DaqifiStreamingDevice.cs

@@ -29,17 +29,10 @@ namespace Daqifi.Core.Device
     public class DaqifiStreamingDevice : DaqifiDevice, IStreamingDevice, INetworkConfigurable, ISdCardOperations, ILanChipInfoProvider, IDeviceDiagnostics, IDeviceOperationHost
     {
         /// <summary>
-        /// Maximum number of retry attempts for the USB stream-interface command sent during
-        /// <see cref="OnDeviceInitializingAsync"/> when the device returns a transient SCPI error
-        /// (e.g. because the firmware still has the interface set from a prior WiFi session).
+        /// Response window allowed for the USB stream-interface command sent during
+        /// <see cref="OnDeviceInitializingAsync"/>.
         /// </summary>
-        private const int UsbStreamInterfaceMaxRetries = 1;
-
-        /// <summary>
-        /// Delay in milliseconds before retrying the USB stream-interface command after a
-        /// transient SCPI error.
-        /// </summary>
-        private const int UsbStreamInterfaceRetryDelayMs = 150;
+        private const int UsbStreamInterfaceResponseTimeoutMs = 500;
 
         /// <summary>
         /// Raised when a stream frame was withheld from consumers because the device should not have
@@ -226,20 +219,17 @@ private void InitializeStreamingDevice()
         /// <see cref="DaqifiDevice.InitializeAsync"/> after the standard SCPI sequence.
         /// </summary>
         /// <remarks>
-        /// The DAQiFi firmware persists the last configured stream interface across sessions.
-        /// If the device was previously set to stream to WiFi (<c>SYSTem:STReam:INTerface 1</c>),
-        /// it will continue sending data over WiFi even when connected via USB — causing the serial
-        /// consumer to receive nothing. Sending <c>SYSTem:STReam:INTerface 0</c> during USB
-        /// initialization ensures data flows to the serial port.
+        /// Whether to route at all, and how hard to retry a transient rejection, is
+        /// <see cref="UsbStreamInterfaceInitializer"/>'s decision; this hook supplies the effect —
+        /// the actual command send — and the connection facts the decision needs.
+        ///
+        /// The send goes through <c>DaqifiDevice.ExecuteTextCommandAsync</c> so the command is sent
+        /// in text mode (protobuf consumer temporarily stopped) and any SCPI error response is
+        /// captured rather than garbling the protobuf stream.
         ///
         /// This runs inside the base <see cref="DaqifiDevice.InitializeAsync"/> exception handling
         /// (before the device is marked initialized/ready), so a cancellation or SCPI error here
         /// leaves the device in a consistent state and re-initializable, rather than falsely Ready.
-        ///
-        /// The routing command is global device state: it takes the stream away from whatever
-        /// interface it was going to, so a second session running it steals another session's data
-        /// (#385). It is therefore skipped entirely when <paramref name="preserveActiveStream"/>
-        /// is set.
         /// </remarks>
         /// <param name="preserveActiveStream">
         /// When <c>true</c>, this initialization must leave a stream another session is already
@@ -253,59 +243,17 @@ private void InitializeStreamingDevice()
         /// command because it still has the interface set from a prior WiFi-streaming session,
         /// within the tight response window right after connect.
         /// </exception>
-        protected override async Task OnDeviceInitializingAsync(
+        protected override Task OnDeviceInitializingAsync(
             bool preserveActiveStream,
-            CancellationToken cancellationToken)
-        {
-            if (!IsUsbConnection)
-            {
-                return;
-            }
-
-            // An observe-only session must not re-route the device's single global stream: doing so
-            // would take the data away from the session that is already receiving it (#385). The
-            // interface is left exactly as the owning session configured it.
-            //
-            // Returning without observing the token is deliberate: there is no work to abandon, and
-            // InitializeAsync re-checks cancellation before it marks the device Ready, so a token
-            // cancelled during this hook is still honored.
-            if (preserveActiveStream)
-            {
-                return;
-            }
-
-            // Direct streaming to the USB interface. Uses ExecuteTextCommandAsync so the
-            // command is sent in text mode (protobuf consumer temporarily stopped) and any
-            // SCPI error response is captured rather than garbling the protobuf stream.
-            //
-            // The firmware persists the last-used stream interface across sessions, so this can
-            // transiently reject with a -200 "Execution error" right after connect. Retry with a
-            // settle delay before treating it as a hard failure (mirrors the SD card retry).
-            IReadOnlyList<string> lines = Array.Empty<string>();
-            for (var attempt = 0; attempt <= UsbStreamInterfaceMaxRetries; attempt++)
-            {
-                if (attempt > 0)
-                {
-                    await Task.Delay(UsbStreamInterfaceRetryDelayMs, cancellationToken).ConfigureAwait(false);
-                }
-
-                lines = await ExecuteTextCommandAsync(
+            CancellationToken cancellationToken) =>
+            UsbStreamInterfaceInitializer.RouteStreamToUsbAsync(
+                IsUsbConnection,
+                preserveActiveStream,
+                ct => ExecuteTextCommandAsync(
                     () => Send(ScpiMessageProducer.SetStreamInterface(StreamInterface.Usb)),
-                    responseTimeoutMs: 500,
-                    cancellationToken: cancellationToken).ConfigureAwait(false);
-
-                if (!ScpiResponseClassifier.ContainsScpiError(lines))
-                {
-                    return;
-                }
-            }
-
-            var lastScpiError = lines.LastOrDefault(ScpiResponseClassifier.IsScpiErrorLine)?.Trim();
-            throw new ScpiInitializationErrorException(
-                "Device returned a SCPI error while setting stream interface to USB.",
-                lines,
-                lastScpiError);
-        }
+                    responseTimeoutMs: UsbStreamInterfaceResponseTimeoutMs,
+                    cancellationToken: ct),
+                cancellationToken);
 
         /// <summary>
         /// Starts streaming data from the device at the configured frequency.