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 e267c5c..cf91135 100644 --- a/src/Daqifi.Core/Device/DaqifiStreamingDevice.cs +++ b/src/Daqifi.Core/Device/DaqifiStreamingDevice.cs @@ -29,17 +29,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 @@ -226,20 +219,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 @@ -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. /// - 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); + } + } +}