diff --git a/README.md b/README.md index 52bcff27..7629e260 100644 --- a/README.md +++ b/README.md @@ -135,6 +135,13 @@ var options = new DeviceConnectionOptions using var device = await DaqifiDeviceFactory.ConnectTcpAsync("192.168.1.100", 9760, options); ``` +> **Connecting takes control of the device.** A DAQiFi unit has a single global acquisition, and the +> default connect sequence stops it — so connecting to a device another session is already streaming +> silently ends that session's data. Use `DeviceConnectionOptions.Observing` for a secondary session +> that only needs to look, and `DaqifiDeviceRegistry` to avoid opening the same unit twice in one +> process. See +> [Connecting stops any stream already running](docs/DEVICE_INTERFACES.md#connecting-stops-any-stream-already-running). + ### Device discovery ```csharp diff --git a/docs/DEVICE_INTERFACES.md b/docs/DEVICE_INTERFACES.md index 28631fc2..5bc4266b 100644 --- a/docs/DEVICE_INTERFACES.md +++ b/docs/DEVICE_INTERFACES.md @@ -113,7 +113,8 @@ var options = new DeviceConnectionOptions MaxAttempts = 3, ConnectionTimeout = TimeSpan.FromSeconds(10) }, - InitializeDevice = true // Run init sequence after connect + InitializeDevice = true, // Run init sequence after connect + PreserveActiveStream = false // Take control of the device (see the warning below) }; ``` @@ -121,6 +122,13 @@ Pre-configured presets: - `DeviceConnectionOptions.Default` - Standard settings - `DeviceConnectionOptions.Fast` - Quick connection, fewer retries - `DeviceConnectionOptions.Resilient` - More retries, longer timeouts +- `DeviceConnectionOptions.Observing` - Connect without disturbing a stream already running + (see [Connecting stops any stream already running](#connecting-stops-any-stream-already-running)) + +> **Connecting stops whatever the device was streaming.** With the default options, initialization +> takes control of the device. If another session — another app, another process, another machine — +> was already streaming from that unit, its data stops, silently. +> See [Connecting stops any stream already running](#connecting-stops-any-stream-already-running). ## Usage Examples @@ -327,6 +335,59 @@ blocks other threads. Two concurrent `ConnectAsync` calls for the *same* physica open a connection — the loser is detected after connecting and disposed — so serialize your own calls if a single connect attempt matters. +### Connecting stops any stream already running + +A DAQiFi device has **one** acquisition: one ADC, one sample rate, one destination interface. There +is no per-connection stream. So the connect-time initialization sequence, which stops streaming, +sets the power state, fixes the stream format, and (over USB) routes the stream to this connection, +acts on the device as a whole — not on your connection. + +That is the right default when your session owns the device: it also clears a stream orphaned by a +previously crashed session, so a stale acquisition never leaks into a fresh one. But when a +**second** session connects to a device that is already streaming, the same sequence ends the first +session's acquisition. Neither side is told. The first app's data simply stops. + +Realistic ways to hit this, all silent to the victim: + +- A desktop app on USB while a script or second app talks to the same unit over WiFi +- Two instances of the same application +- A second viewer attaching to a unit a logger is already streaming + +**If you only need to look, connect as an observer.** `PreserveActiveStream` skips every +initialization command that writes global stream state: + +```csharp +// A secondary session that must not disturb whoever is already streaming. +var device = await DaqifiDeviceFactory.ConnectTcpAsync( + ip, DaqifiDeviceFactory.DefaultTcpDataPort, DeviceConnectionOptions.Observing); + +// Connecting manually? Set it before InitializeAsync. +device.PreserveActiveStream = true; +await device.InitializeAsync(); +``` + +| Initialization step | Default | `PreserveActiveStream` | +|---|---|---| +| `SYSTem:ECHO -1` | sent | sent — text-mode only, no stream state | +| `SYSTem:StopStreamData` | sent | **skipped** — this is what kills the other session | +| `SYSTem:POWer:STATe 1` | sent | **skipped** | +| `SYSTem:STReam:FORmat 0` | sent | **skipped** | +| `SYSTem:STReam:INTerface` (USB routing) | sent | **skipped** — would steal the stream | +| `SYSTem:SYSInfoPB?` + capability query | sent | sent — read-only | + +The observing session is fully usable for status, metadata, and channel inspection, and reaches +`DeviceState.Ready` exactly as a normal connection does. What it is **not** is configured to stream: +the format and destination interface are left as the other session set them, and frames keep going +wherever they were already going. A session that later wants to stream itself has to take control, +which necessarily stops the other one — reconnect with the default options for that. + +**Limits.** This is a courtesy, not arbitration. It stops *this* library from clobbering a stream; +it cannot stop anything else from doing so, and the firmware does not currently reject or announce +a second controlling session. Within one process, prefer +[`DaqifiDeviceRegistry`](#managing-multiple-devices-daqifideviceregistry) — it refuses to open the +same physical unit twice at all, so the conflict never arises. Across processes there is no +protection beyond both sides opting in to `PreserveActiveStream`. + ### Manual Device Connection (Advanced) For cases where you need more control over the connection process: @@ -342,6 +403,10 @@ await transport.ConnectAsync(new ConnectionRetryOptions { MaxAttempts = 3 }); // Create device with transport using var device = new DaqifiDevice("My Device", transport); device.Connect(); + +// InitializeAsync takes control of the device and stops any stream it was already running. +// Set device.PreserveActiveStream = true first if another session may be streaming — see +// "Connecting stops any stream already running" above. await device.InitializeAsync(); // Now ready to send commands diff --git a/src/Daqifi.Core.Tests/Device/DaqifiDeviceFactoryTests.cs b/src/Daqifi.Core.Tests/Device/DaqifiDeviceFactoryTests.cs index 1fc296ba..15f5e757 100644 --- a/src/Daqifi.Core.Tests/Device/DaqifiDeviceFactoryTests.cs +++ b/src/Daqifi.Core.Tests/Device/DaqifiDeviceFactoryTests.cs @@ -20,6 +20,9 @@ public void DeviceConnectionOptions_DefaultValues_AreCorrect() Assert.Equal("DAQiFi Device", options.DeviceName); Assert.Null(options.ConnectionRetry); Assert.True(options.InitializeDevice); + // Connecting takes control of the device (and stops any running stream) unless + // explicitly opted out of — the historical behavior (#385). + Assert.False(options.PreserveActiveStream); } [Fact] @@ -32,6 +35,26 @@ public void DeviceConnectionOptions_Default_ReturnsDefaultOptions() Assert.Equal("DAQiFi Device", options.DeviceName); Assert.Null(options.ConnectionRetry); Assert.True(options.InitializeDevice); + Assert.False(options.PreserveActiveStream); + } + + [Fact] + public void DeviceConnectionOptions_Observing_PreservesActiveStream() + { + // Act + var options = DeviceConnectionOptions.Observing; + + // Assert — the opt-in preset still initializes, it just does so non-disruptively + Assert.True(options.PreserveActiveStream); + Assert.True(options.InitializeDevice); + } + + [Fact] + public void DeviceConnectionOptions_FastAndResilient_DoNotPreserveActiveStream() + { + // Assert — the existing presets keep the historical take-control behavior + Assert.False(DeviceConnectionOptions.Fast.PreserveActiveStream); + Assert.False(DeviceConnectionOptions.Resilient.PreserveActiveStream); } [Fact] diff --git a/src/Daqifi.Core.Tests/Device/DaqifiDeviceInitializeTests.cs b/src/Daqifi.Core.Tests/Device/DaqifiDeviceInitializeTests.cs index 56e4d0b0..195f243e 100644 --- a/src/Daqifi.Core.Tests/Device/DaqifiDeviceInitializeTests.cs +++ b/src/Daqifi.Core.Tests/Device/DaqifiDeviceInitializeTests.cs @@ -32,6 +32,111 @@ public async Task InitializeAsync_SendsAllConfigCommands() Assert.Contains(sentData, d => d.Contains("SYSTem:POWer:STATe 1")); Assert.Contains(sentData, d => d.Contains("SYSTem:STReam:FORmat 0")); Assert.Contains(sentData, d => d.Contains("SYSTem:SYSInfoPB?")); + + // The stream-disruptive commands are the default because this session is assumed to + // own the device; the opt-out below must not change that (#385). + Assert.False(device.PreserveActiveStream); + } + + [Fact] + public async Task InitializeAsync_WithPreserveActiveStream_OmitsStreamDisruptiveCommands() + { + // Arrange — a secondary session attaching to a device another session may already be + // streaming from. + var device = new TestableDaqifiDevice("TestDevice") { PreserveActiveStream = true }; + device.Connect(); + + // Act + await device.InitializeAsync(); + + // Assert — nothing that halts, powers, or reconfigures the device's single global + // stream is sent. + var sentData = device.DirectSentMessages.Select(m => m.Data).ToList(); + + Assert.DoesNotContain(sentData, d => d.Contains("SYSTem:StopStreamData")); + Assert.DoesNotContain(sentData, d => d.Contains("SYSTem:POWer:STATe")); + Assert.DoesNotContain(sentData, d => d.Contains("SYSTem:STReam:FORmat")); + + // ...but the session is still fully set up: echo off so its own replies parse, and the + // identity query that populates channels. + Assert.Contains(sentData, d => d.Contains("SYSTem:ECHO -1")); + Assert.Contains(sentData, d => d.Contains("SYSTem:SYSInfoPB?")); + } + + [Fact] + public async Task InitializeAsync_WithPreserveActiveStream_StillProducesReadyPopulatedDevice() + { + // Arrange + var device = new TestableDaqifiDevice("TestDevice") { PreserveActiveStream = true }; + device.Connect(); + + // Act + await device.InitializeAsync(); + + // Assert — skipping the disruptive commands must not cost the caller a usable session + Assert.Equal(DeviceState.Ready, device.State); + Assert.Equal(4, device.Channels.Count); // 2 analog + 2 digital + } + + [Fact] + public async Task InitializeAsync_WithPreserveActiveStream_StreamingUsb_DoesNotRouteStreamToUsb() + { + // Arrange — over USB the streaming device normally claims the stream with + // SYSTem:STReam:INTerface 0, which would take data away from a session already + // receiving it over WiFi. + var device = new TestableStreamingDevice("TestDevice") { PreserveActiveStream = true }; + device.Connect(); + + // Act + await device.InitializeAsync(); + + // Assert + Assert.DoesNotContain(device.SentData, d => d.Contains("SYSTem:STReam:INTerface")); + Assert.Equal(0, device.UsbStepAttemptCount); + Assert.Equal(DeviceState.Ready, device.State); + Assert.Equal(4, device.Channels.Count); + } + + [Fact] + public async Task InitializeAsync_WhenPreserveActiveStreamChangesMidInitialization_KeepsTheDecisionItStartedWith() + { + // Arrange — an observing initialization whose flag is flipped to false after it has + // begun (a concurrent initialization on the same instance, or a caller mutating the + // property). The decision belongs to the operation, so the USB routing step must still + // be skipped: honoring the late change would steal a stream this session promised not + // to touch. + var device = new TestableStreamingDevice("TestDevice") { PreserveActiveStream = true }; + device.MutateDuringInitialization = () => device.PreserveActiveStream = false; + device.Connect(); + + // Act + await device.InitializeAsync(); + + // Assert + Assert.False(device.PreserveActiveStream); // the mutation really did land + Assert.DoesNotContain(device.SentData, d => d.Contains("SYSTem:STReam:INTerface")); + Assert.Equal(0, device.UsbStepAttemptCount); + Assert.Equal(DeviceState.Ready, device.State); + } + + [Fact] + public async Task InitializeAsync_WhenPreserveActiveStreamIsSetMidInitialization_StillTakesControl() + { + // Arrange — the mirror case: a normal take-control initialization must not be silently + // downgraded to observing by a flag set after it started, which would leave the stream + // routed somewhere this session cannot read. + var device = new TestableStreamingDevice("TestDevice"); + device.MutateDuringInitialization = () => device.PreserveActiveStream = true; + device.Connect(); + + // Act + await device.InitializeAsync(); + + // Assert + Assert.True(device.PreserveActiveStream); // the mutation really did land + Assert.Contains(device.SentData, d => d.Contains("SYSTem:STReam:INTerface 0")); + Assert.Contains(device.SentData, d => d.Contains("SYSTem:StopStreamData")); + Assert.Equal(DeviceState.Ready, device.State); } [Fact] @@ -480,6 +585,217 @@ private enum UsbStepBehavior Cancel } + [Theory] + [InlineData(true)] // observing: the hook returns immediately, no awaitable work + [InlineData(false)] // take-control: the hook does send SCPI + public async Task InitializeAsync_WhenCancelledAfterChannelsPopulate_DoesNotReportReady(bool preserveActiveStream) + { + // Arrange — cancellation lands at the one seam nothing was guaranteed to observe: + // after channels populate, during the capability read. Firmware that does not + // advertise the capability document returns from that read without touching the + // token, and on the observing path the derived hook then returns immediately too, so + // a device whose caller had already cancelled still reached Ready. + using var cts = new CancellationTokenSource(); + var device = new CancelDuringCapabilityReadDevice("TestDevice", cts) + { + PreserveActiveStream = preserveActiveStream + }; + device.Connect(); + + // Act & Assert — a cancelled initialization reports cancellation, not success. + await Assert.ThrowsAsync( + () => device.InitializeAsync(TimeSpan.FromSeconds(5), cts.Token)); + + Assert.NotEqual(DeviceState.Ready, device.State); + Assert.Equal(DeviceState.Connected, device.State); + } + + /// + /// A testable USB streaming device that cancels the supplied token from inside the + /// capability read — i.e. after channels have populated and before the derived + /// initialization hook — and returns without observing the token itself, exactly as the + /// real read does on firmware that does not advertise a capability document. + /// + private class CancelDuringCapabilityReadDevice : DaqifiStreamingDevice + { + private readonly CancellationTokenSource _cts; + + public override bool IsUsbConnection => true; + + public CancelDuringCapabilityReadDevice(string name, CancellationTokenSource cts) + : base(name, (IPAddress?)null) + { + _cts = cts; + } + + public override void Send(IOutboundMessage message) + { + if (message is IOutboundMessage stringMessage && + stringMessage.Data.Contains("SYSInfoPB")) + { + PopulateChannelsFromStatus(new DaqifiOutMessage + { + AnalogInPortNum = 2, + DigitalPortNum = 2 + }); + } + } + + public override Task ReadCapabilityDocumentAsync( + CancellationToken cancellationToken = default) + { + _cts.Cancel(); + return Task.FromResult(null); + } + + protected override async Task> ExecuteTextCommandAsync( + Action setupAction, + int responseTimeoutMs = 1000, + int completionTimeoutMs = 250, + CancellationToken cancellationToken = default, + Func? prepareAsync = null, + Func? finalizeAsync = null) + { + try + { + if (prepareAsync != null) + { + await prepareAsync(cancellationToken).ConfigureAwait(false); + } + + setupAction(); + return Array.Empty(); + } + finally + { + // Honor the exchange's finalize phase the way the real device does: it runs + // however the exchange ended, still inside the exchange (#407). + if (finalizeAsync != null) + { + await finalizeAsync().ConfigureAwait(false); + } + } + } + } + + [Fact] + public async Task InitializeAsync_WhenTwoInitializationsOverlapOnOneDevice_EachKeepsItsOwnDecision() + { + // Arrange — the race reported against the first cut of this change: the observing + // decision was held in an instance field, so a second InitializeAsync starting while + // the first was still in flight overwrote it, and the first initialization's USB + // routing step then acted on the second one's decision. Reproduced by holding an + // observing initialization inside its first SCPI exchange — past the point the + // decision is made, before the routing step — while a take-control initialization + // starts on the same instance. + var device = new OverlappingInitDevice("TestDevice"); + device.Connect(); + + device.PreserveActiveStream = true; + var observing = device.InitializeAsync(); + + // Wait until the observing call is parked inside its first exchange. + await device.FirstExchangeEntered.Task.WaitAsync(TimeSpan.FromSeconds(5)); + + device.PreserveActiveStream = false; + var takingControl = device.InitializeAsync(); + await takingControl.WaitAsync(TimeSpan.FromSeconds(5)); + + // Act — let the observing initialization finish, now that the flag has moved under it. + device.ReleaseFirstExchange(); + await observing.WaitAsync(TimeSpan.FromSeconds(5)); + + // Assert — one initialization decided to observe and one to take control, and each + // reached its own hook with its own decision. A shared field yields {false, false}. + Assert.Equal( + new[] { false, true }, + device.HookDecisions.OrderBy(flag => flag).ToArray()); + } + + /// + /// A testable USB streaming device that records the decision each initialization passes to + /// OnDeviceInitializingAsync, and can park its first text exchange so a second + /// initialization can be started while the first is still in flight. + /// + private class OverlappingInitDevice : DaqifiStreamingDevice + { + private readonly TaskCompletionSource _release = + new(TaskCreationOptions.RunContinuationsAsynchronously); + private int _exchangeCount; + + /// Completes once an initialization has entered its first text exchange. + public TaskCompletionSource FirstExchangeEntered { get; } = + new(TaskCreationOptions.RunContinuationsAsynchronously); + + /// The decision each initialization handed to the derived hook. + public System.Collections.Concurrent.ConcurrentBag HookDecisions { get; } = new(); + + public override bool IsUsbConnection => true; + + public OverlappingInitDevice(string name) : base(name, (IPAddress?)null) { } + + public void ReleaseFirstExchange() => _release.TrySetResult(); + + public override void Send(IOutboundMessage message) + { + if (message is IOutboundMessage stringMessage && + stringMessage.Data.Contains("SYSInfoPB")) + { + PopulateChannelsFromStatus(new DaqifiOutMessage + { + AnalogInPortNum = 2, + DigitalPortNum = 2 + }); + } + } + + protected override Task OnDeviceInitializingAsync( + bool preserveActiveStream, + CancellationToken cancellationToken) + { + HookDecisions.Add(preserveActiveStream); + return Task.CompletedTask; + } + + protected override async Task> ExecuteTextCommandAsync( + Action setupAction, + int responseTimeoutMs = 1000, + int completionTimeoutMs = 250, + CancellationToken cancellationToken = default, + Func? prepareAsync = null, + Func? finalizeAsync = null) + { + try + { + if (prepareAsync != null) + { + await prepareAsync(cancellationToken).ConfigureAwait(false); + } + + setupAction(); + + // Park only the very first exchange — the one belonging to the initialization + // the test starts first. + if (Interlocked.Increment(ref _exchangeCount) == 1) + { + FirstExchangeEntered.TrySetResult(); + await _release.Task.ConfigureAwait(false); + } + + return Array.Empty(); + } + finally + { + // Honor the exchange's finalize phase the way the real device does: it runs + // however the exchange ended, still inside the exchange (#407). + if (finalizeAsync != null) + { + await finalizeAsync().ConfigureAwait(false); + } + } + } + } + /// /// A testable DaqifiStreamingDevice (always USB) whose base init populates channels on /// GetDeviceInfo and whose USB stream-interface step can be made to succeed, return a SCPI @@ -497,6 +813,15 @@ private class TestableStreamingDevice : DaqifiStreamingDevice /// public int UsbStepAttemptCount { get; private set; } + /// + /// Invoked once, from inside the first text exchange of initialization — i.e. after + /// InitializeAsync has captured its PreserveActiveStream decision but before the + /// derived USB step runs. Lets a test mutate device state mid-initialization. + /// + public Action? MutateDuringInitialization { get; set; } + + private bool _mutationApplied; + public override bool IsUsbConnection => true; public TestableStreamingDevice(string name, UsbStepBehavior usbStepBehavior = UsbStepBehavior.Succeed) @@ -538,6 +863,12 @@ protected override async Task> ExecuteTextCommandAsync( await prepareAsync(cancellationToken).ConfigureAwait(false); } + if (!_mutationApplied && MutateDuringInitialization != null) + { + _mutationApplied = true; + MutateDuringInitialization(); + } + var before = _sent.Count; setupAction(); var sentThisCall = _sent.Skip(before).ToList(); diff --git a/src/Daqifi.Core/Device/DaqifiDevice.cs b/src/Daqifi.Core/Device/DaqifiDevice.cs index 5bed4137..133f4d8c 100644 --- a/src/Daqifi.Core/Device/DaqifiDevice.cs +++ b/src/Daqifi.Core/Device/DaqifiDevice.cs @@ -313,6 +313,45 @@ protected void WithChannelsLock(Action action) /// public DeviceState State { get; private set; } = DeviceState.Disconnected; + /// + /// When true, omits the initialization commands that + /// would halt or re-route a stream the device is already running, so connecting does not + /// disturb a session started elsewhere. Default is false — the historical behavior, + /// where connecting takes control of the device. + /// + /// + /// + /// Streaming is a single global device state: one acquisition at one rate, delivered to one + /// interface. The default initialization sequence stops that stream, sets the device's power + /// state, fixes the stream format, and (over USB) routes the stream to this connection. That + /// is correct when this session owns the device — it also clears a stream orphaned by a + /// previously crashed session — but a second session running it silently ends the + /// first session's acquisition, with no error surfaced to either side. + /// + /// + /// With this set, initialization sends only SYSTem:ECHO -1 (so replies to this + /// connection can be parsed) followed by the read-only identity and capability queries. + /// Skipped are SYSTem:StopStreamData, SYSTem:POWer:STATe 1, + /// SYSTem:STReam:FORmat 0, and the USB SYSTem:STReam:INTerface routing step. + /// + /// + /// The resulting session is fully usable for status, metadata, channel inspection, and any + /// command the caller chooses to send. It is not configured to stream: the device's + /// stream format and destination interface are left exactly as the other session left them, + /// and stream frames continue to go wherever they were already going. A session that later + /// wants to stream itself must take control of the device, which necessarily stops whatever + /// the other session was doing. + /// + /// + /// Read once, when runs; changing it afterwards has no effect. + /// This guards only against this library's own connect sequence — it is not device-side + /// arbitration, and two processes can still fight over one unit. Within a single process, + /// prefer , which refuses to open the same physical unit + /// twice. + /// + /// + public bool PreserveActiveStream { get; set; } + private ConnectionStatus _status; /// @@ -1714,6 +1753,13 @@ public void Dispose() /// 4. Set protobuf message format /// 5. Query device info and block until the device reports its channel configuration /// + /// Steps 2-4 write global device state. Streaming is a single global state on a DAQiFi + /// device, so step 2 stops a stream any session started, not just this one: + /// connecting to a device another session is already streaming silently ends that session's + /// acquisition. That is the right default when this session owns the device (it also clears a + /// stream orphaned by a crashed session). Set before + /// calling this to skip steps 2-4 and connect as a non-disruptive observer instead. + /// /// Rather than returning after a fixed delay, the method awaits the first /// event so callers receive a fully populated device. /// Serial/CDC devices can take noticeably longer than the previous fixed wait to send @@ -1774,6 +1820,14 @@ public virtual async Task InitializeAsync( _messageConsumer.MessageReceived += OnInboundMessageReceived; } + // Snapshot into a local, once. Every retry attempt and the derived-class hook + // further down are then handed the same decision explicitly, so it cannot be + // changed out from under an initialization already in flight — neither by a caller + // mutating the property nor by a second concurrent InitializeAsync on this same + // instance. The decision belongs to this operation, so it lives on the stack rather + // than in a field. + var preserveActiveStream = PreserveActiveStream; + // Send the text-mode SCPI setup commands via ExecuteTextCommandAsync so that // any -200 execution error response is captured rather than silently discarded // by the protobuf consumer. The protobuf consumer is stopped for the duration @@ -1795,7 +1849,20 @@ public virtual async Task InitializeAsync( initLines = await ExecuteTextCommandAsync(() => { + // Echo is a per-device text-mode setting, not stream state: this session + // needs it off to parse its own replies, and the value is the same one any + // other Core session already set. Safe to send either way. Send(ScpiMessageProducer.DisableDeviceEcho); + + // Everything below writes global stream state. A secondary "observe" + // session must not touch it — StopStreamData ends another session's + // acquisition outright (#385), and the power-state and stream-format + // commands reconfigure the same single acquisition it is running. + if (preserveActiveStream) + { + return; + } + Thread.Sleep(100); Send(ScpiMessageProducer.StopStreaming); @@ -1852,7 +1919,16 @@ await WaitForChannelsPopulatedAsync( // this try/catch so a failure there leaves the device in a consistent terminal state // rather than a falsely-ready device. _isInitialized is only set after it succeeds, // so a failed init can be safely retried. - await OnDeviceInitializingAsync(cancellationToken).ConfigureAwait(false); + await OnDeviceInitializingAsync(preserveActiveStream, cancellationToken).ConfigureAwait(false); + + // A cancelled initialization must never report Ready. Nothing above is guaranteed + // to observe the token: the capability read returns early on firmware that does not + // advertise the document, channel population can short-circuit when the status + // arrives synchronously, and a derived hook may legitimately have no awaitable work + // (the observing path and the non-USB path both return immediately). So the + // invariant is enforced here, at the one transition that matters, rather than relying + // on every path and every override to check for itself. + cancellationToken.ThrowIfCancellationRequested(); _isInitialized = true; State = DeviceState.Ready; @@ -1883,9 +1959,18 @@ await WaitForChannelsPopulatedAsync( /// state and other faults set — rather than a falsely-ready /// device, and the failed initialization can be retried. The base implementation does nothing. /// + /// + /// The decision for this initialization, passed + /// explicitly rather than read from the device so it cannot change while initialization is in + /// flight. When true, an override must not send any command that writes global stream + /// state — stopping, reconfiguring, or re-routing the stream would disturb a session that is + /// already using the device. + /// /// A cancellation token to observe. /// A task representing the asynchronous operation. - protected virtual Task OnDeviceInitializingAsync(CancellationToken cancellationToken) => Task.CompletedTask; + protected virtual Task OnDeviceInitializingAsync( + bool preserveActiveStream, + CancellationToken cancellationToken) => Task.CompletedTask; /// /// Runs the capability-document read during initialization, absorbing any failure. diff --git a/src/Daqifi.Core/Device/DaqifiDeviceFactory.cs b/src/Daqifi.Core/Device/DaqifiDeviceFactory.cs index 743588a5..9b88b163 100644 --- a/src/Daqifi.Core/Device/DaqifiDeviceFactory.cs +++ b/src/Daqifi.Core/Device/DaqifiDeviceFactory.cs @@ -384,6 +384,7 @@ private static async Task ConnectWiFiDeviceAsync( ConnectionRetry = effectiveOptions.ConnectionRetry, InitializeDevice = effectiveOptions.InitializeDevice, ChannelPopulationTimeout = effectiveOptions.ChannelPopulationTimeout, + PreserveActiveStream = effectiveOptions.PreserveActiveStream, Logger = effectiveOptions.Logger }; @@ -427,6 +428,7 @@ private static async Task ConnectSerialDeviceAsync( ConnectionRetry = effectiveOptions.ConnectionRetry, InitializeDevice = effectiveOptions.InitializeDevice, ChannelPopulationTimeout = effectiveOptions.ChannelPopulationTimeout, + PreserveActiveStream = effectiveOptions.PreserveActiveStream, Logger = effectiveOptions.Logger }; @@ -470,7 +472,12 @@ private static async Task ConnectWithTransportAsync( // Step 2: Create the device with the transport // Note: Once created, the device owns the transport and will dispose it - device = new DaqifiStreamingDevice(options.DeviceName, transport, options.Logger); + device = new DaqifiStreamingDevice(options.DeviceName, transport, options.Logger) + { + // Read by InitializeAsync below; set before Connect so it is never observed + // half-applied. + PreserveActiveStream = options.PreserveActiveStream + }; // Step 3: Connect the device (starts message producers/consumers) device.Connect(); diff --git a/src/Daqifi.Core/Device/DaqifiStreamingDevice.cs b/src/Daqifi.Core/Device/DaqifiStreamingDevice.cs index 95059559..c9ed15ae 100644 --- a/src/Daqifi.Core/Device/DaqifiStreamingDevice.cs +++ b/src/Daqifi.Core/Device/DaqifiStreamingDevice.cs @@ -304,7 +304,16 @@ private void InitializeStreamingDevice() /// 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 + /// running untouched, so the routing command is not sent at all. + /// /// A cancellation token to observe while initializing. /// A task representing the asynchronous initialization operation. /// @@ -313,13 +322,27 @@ 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(CancellationToken cancellationToken) + protected override async 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. diff --git a/src/Daqifi.Core/Device/DeviceConnectionOptions.cs b/src/Daqifi.Core/Device/DeviceConnectionOptions.cs index 3e7fb131..c908e58a 100644 --- a/src/Daqifi.Core/Device/DeviceConnectionOptions.cs +++ b/src/Daqifi.Core/Device/DeviceConnectionOptions.cs @@ -37,6 +37,34 @@ public class DeviceConnectionOptions /// public TimeSpan ChannelPopulationTimeout { get; set; } = TimeSpan.FromSeconds(8); + /// + /// Gets or sets a value indicating whether initialization must leave a stream that is already + /// running on the device untouched. Default is false, which preserves the historical + /// behavior: connecting takes control of the device and stops whatever it was streaming. + /// + /// + /// + /// Streaming is a single global device state — one acquisition, one destination interface — so + /// the default initialization sequence deliberately clears it, which is right for the usual + /// single-session case (it also clears a stream orphaned by a crashed session). When a second + /// session connects to a device another session is already streaming, that same sequence + /// silently ends the first session's acquisition. + /// + /// + /// Set this to true for a secondary "observe" connection. Initialization then omits every + /// command that halts or re-routes the device's stream, and only queries the device for its + /// identity and channel configuration. See + /// for exactly which commands are skipped and + /// what the resulting session can and cannot do. + /// + /// + /// This only protects against this library's connect sequence. Two processes can still + /// fight over one device; within a process, route connections through + /// so the same physical unit is not opened twice at all. + /// + /// + public bool PreserveActiveStream { get; set; } + /// /// Optional logger the constructed device routes its diagnostics through (bad calibration/ /// resolution warnings, SCPI text-exchange timing). When null, the device uses a no-op logger. @@ -65,4 +93,13 @@ public class DeviceConnectionOptions { ConnectionRetry = ConnectionRetryOptions.Resilient }; + + /// + /// Creates a configuration for a secondary session that must not disturb a stream another + /// session may already be running on the device. Sets . + /// + public static DeviceConnectionOptions Observing => new() + { + PreserveActiveStream = true + }; }