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);
+ }
+ }
+}