From 8a99fd2e8cc9fca93f45c2092205d1449db4e7ad Mon Sep 17 00:00:00 2001 From: Tyler Kron Date: Wed, 5 Aug 2026 20:13:24 -0600 Subject: [PATCH] refactor(device): extract the USB stream-interface routing decision into a collaborator (part of #344) DaqifiStreamingDevice.OnDeviceInitializingAsync carried the whole USB stream-routing policy inline: whether to route at all, how many times to retry a transient SCPI rejection, how long to settle between attempts, and how to turn a persistent rejection into a typed exception. None of that needed a device to decide. Move the decision into a new host-free collaborator, Device/Internal/UsbStreamInterfaceInitializer. It takes the two connection facts the decision depends on (is this USB, is this an observe-only session) plus a delegate that performs the send, so the policy is now testable without a device, a transport, or a wire. The effect stays on the device: the hook still builds the ExecuteTextCommandAsync call so the command goes out in text mode with the protobuf consumer stopped. No public surface changes and no IDeviceOperationHost additions. The existing DaqifiDeviceInitializeTests are left untouched on purpose -- they are the evidence that the extraction changed nothing end to end. DaqifiStreamingDevice.cs: 1196 -> 1144 lines. Co-Authored-By: Claude Opus 5 --- .../UsbStreamInterfaceInitializerTests.cs | 208 ++++++++++++++++++ .../Device/DaqifiStreamingDevice.cs | 90 ++------ .../Internal/UsbStreamInterfaceInitializer.cs | 127 +++++++++++ 3 files changed, 354 insertions(+), 71 deletions(-) create mode 100644 src/Daqifi.Core.Tests/Device/Internal/UsbStreamInterfaceInitializerTests.cs create mode 100644 src/Daqifi.Core/Device/Internal/UsbStreamInterfaceInitializer.cs diff --git a/src/Daqifi.Core.Tests/Device/Internal/UsbStreamInterfaceInitializerTests.cs b/src/Daqifi.Core.Tests/Device/Internal/UsbStreamInterfaceInitializerTests.cs new file mode 100644 index 0000000..82bc41d --- /dev/null +++ b/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; + +/// +/// Unit tests for , 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; DaqifiDeviceInitializeTests pins what the device does with +/// it end to end. +/// +public class UsbStreamInterfaceInitializerTests +{ + private const string ScpiError = "**ERROR: -200, \"Execution error\""; + + /// Bound on every wait so a policy regression fails the run instead of parking it. + private static readonly TimeSpan RouteTimeout = TimeSpan.FromSeconds(5); + + /// + /// 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. + /// + private sealed class ScriptedSender + { + private readonly Queue> _responses = new(); + + public int AttemptCount { get; private set; } + + public List ObservedTokens { get; } = new(); + + public ScriptedSender Answers(params string[] lines) + { + _responses.Enqueue(lines); + return this; + } + + public Task> 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(() => 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(() => 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( + () => 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); + } +} diff --git a/src/Daqifi.Core/Device/DaqifiStreamingDevice.cs b/src/Daqifi.Core/Device/DaqifiStreamingDevice.cs index cc73557..127ed59 100644 --- a/src/Daqifi.Core/Device/DaqifiStreamingDevice.cs +++ b/src/Daqifi.Core/Device/DaqifiStreamingDevice.cs @@ -31,17 +31,10 @@ namespace Daqifi.Core.Device public class DaqifiStreamingDevice : DaqifiDevice, IStreamingDevice, INetworkConfigurable, ISdCardOperations, ILanChipInfoProvider, IDeviceDiagnostics, IDeviceOperationHost { /// - /// Maximum number of retry attempts for the USB stream-interface command sent during - /// 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 + /// . /// - private const int UsbStreamInterfaceMaxRetries = 1; - - /// - /// Delay in milliseconds before retrying the USB stream-interface command after a - /// transient SCPI error. - /// - private const int UsbStreamInterfaceRetryDelayMs = 150; + private const int UsbStreamInterfaceResponseTimeoutMs = 500; /// /// Raised when a stream frame was withheld from consumers because the device should not have @@ -227,20 +220,17 @@ private void InitializeStreamingDevice() /// after the standard SCPI sequence. /// /// - /// The DAQiFi firmware persists the last configured stream interface across sessions. - /// If the device was previously set to stream to WiFi (SYSTem:STReam:INTerface 1), - /// it will continue sending data over WiFi even when connected via USB — causing the serial - /// consumer to receive nothing. Sending SYSTem:STReam:INTerface 0 during USB - /// initialization ensures data flows to the serial port. + /// Whether to route at all, and how hard to retry a transient rejection, is + /// 's decision; this hook supplies the effect — + /// the actual command send — and the connection facts the decision needs. + /// + /// The send goes through DaqifiDevice.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. /// /// This runs inside the base 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 - /// is set. /// /// /// When true, this initialization must leave a stream another session is already @@ -254,59 +244,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. /// - 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 lines = Array.Empty(); - 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); /// /// Starts streaming data from the device at the configured frequency. diff --git a/src/Daqifi.Core/Device/Internal/UsbStreamInterfaceInitializer.cs b/src/Daqifi.Core/Device/Internal/UsbStreamInterfaceInitializer.cs new file mode 100644 index 0000000..14135c6 --- /dev/null +++ b/src/Daqifi.Core/Device/Internal/UsbStreamInterfaceInitializer.cs @@ -0,0 +1,127 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; + +#nullable enable + +namespace Daqifi.Core.Device.Internal +{ + /// + /// Decides whether a device being initialized must have its stream re-routed to USB, and how + /// hard to try, given only what the caller reports about the connection. + /// + /// + /// + /// The DAQiFi firmware persists the last configured stream interface across sessions. If the + /// device was previously set to stream to WiFi (SYSTem:STReam:INTerface 1), it keeps + /// sending data over WiFi even when connected via USB — so the serial consumer receives nothing + /// until SYSTem:STReam:INTerface 0 is sent. + /// + /// + /// This type owns the decision (route or skip, retry or fail) and nothing else. Sending the + /// command is the caller's effect, supplied as a delegate, so the policy is testable without a + /// device, a transport, or a wire. + /// + /// + internal static class UsbStreamInterfaceInitializer + { + /// + /// Maximum number of retry attempts for the USB stream-interface command when the device + /// returns a transient SCPI error (e.g. because the firmware still has the interface set + /// from a prior WiFi session). + /// + internal const int MaxRetries = 1; + + /// + /// Delay in milliseconds before retrying the USB stream-interface command after a transient + /// SCPI error. + /// + internal const int RetryDelayMs = 150; + + /// + /// Decides whether the stream-routing command should be sent at all. + /// + /// + /// 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). An observe-only session must therefore skip it entirely, and a non-USB connection + /// has nothing to route. + /// + /// Whether the device is connected over USB/serial. + /// + /// Whether this initialization must leave a stream another session is already running + /// untouched. + /// + /// true when the command should be sent; otherwise false. + internal static bool ShouldRouteStreamToUsb(bool isUsbConnection, bool preserveActiveStream) + => isUsbConnection && !preserveActiveStream; + + /// + /// Routes the device's stream to USB, retrying a transient SCPI rejection before treating it + /// as a hard failure. + /// + /// + /// + /// When says to skip, this returns without invoking + /// and without observing + /// . Not observing it is deliberate: there is no work to + /// abandon, and the caller's initialization re-checks cancellation before it marks the device + /// ready, so a token cancelled during this step is still honored. + /// + /// + /// The firmware can transiently reject the command with a -200 "Execution error" right + /// after connect, so a rejection is retried once after a settle delay (mirrors the SD card + /// retry). Only a rejection that persists across every attempt is a failure. + /// + /// + /// Whether the device is connected over USB/serial. + /// + /// Whether this initialization must leave a stream another session is already running + /// untouched. + /// + /// + /// Sends the routing command and returns the device's response lines. Invoked once per + /// attempt. + /// + /// A cancellation token to observe while routing. + /// A task representing the asynchronous routing operation. + /// + /// Thrown when the device returns a SCPI error on every attempt. + /// + internal static async Task RouteStreamToUsbAsync( + bool isUsbConnection, + bool preserveActiveStream, + Func>> setStreamInterfaceToUsbAsync, + CancellationToken cancellationToken) + { + if (!ShouldRouteStreamToUsb(isUsbConnection, preserveActiveStream)) + { + return; + } + + IReadOnlyList lines = Array.Empty(); + for (var attempt = 0; attempt <= MaxRetries; attempt++) + { + if (attempt > 0) + { + await Task.Delay(RetryDelayMs, cancellationToken).ConfigureAwait(false); + } + + lines = await setStreamInterfaceToUsbAsync(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); + } + } +}