Phase 2 Streams A+B+C feature-complete — real Win32 pump, all 9 IDriver capabilities, end-to-end IPC dispatch. Streams D+E remain (Galaxy MXAccess code lift + parity-debug cycle, plan-budgeted 3-4 weeks). The 494 v1 IntegrationTests still pass — legacy OtOpcUa.Host untouched. StaPump replaces the BlockingCollection placeholder with a real Win32 message pump lifted from v1 StaComThread per CLAUDE.md "Reference Implementation": dedicated STA Thread with SetApartmentState(STA), GetMessage/PostThreadMessage/PeekMessage/TranslateMessage/DispatchMessage/PostQuitMessage P/Invoke, WM_APP=0x8000 for work-item dispatch, WM_APP+1 for graceful-drain → PostQuitMessage, peek-pm-noremove on entry to force the system to create the thread message queue before signalling Started, IsResponsiveAsync probe still no-op-round-trips through PostThreadMessage so the wedge detection works against the real pump. Concurrent ConcurrentQueue<WorkItem> drains on every WM_APP; fault path on dispose drains-and-faults all pending work-item TCSes with InvalidOperationException("STA pump has exited"). All three StaPumpTests pass against the real pump (apartment state STA, healthy probe true, wedged probe false). GalaxyProxyDriver now implements every Phase 2 Stream C capability — IDriver, ITagDiscovery, IReadable, IWritable, ISubscribable, IAlarmSource, IHistoryProvider, IRediscoverable, IHostConnectivityProbe — each forwarding through the matching IPC contract. ReadAsync preserves request order even when the Host returns out-of-order values; WriteAsync MessagePack-serializes the value into ValueBytes; SubscribeAsync wraps SubscriptionId in a GalaxySubscriptionHandle record; UnsubscribeAsync uses the new SendOneWayAsync helper on GalaxyIpcClient (fire-and-forget but still gated through the call-semaphore so it doesn't interleave with CallAsync); AlarmSubscribe is one-way and the Host pushes events back via OnAlarmEvent; ReadProcessedAsync short-circuits to NotSupportedException (Galaxy historian only does raw); IRediscoverable's OnRediscoveryNeeded fires when the Host pushes a deploy-watermark notification; IHostConnectivityProbe.GetHostStatuses() snapshots and OnHostStatusChanged fires on Running↔Stopped/Faulted transitions, with IpcHostConnectivityStatus aliased to disambiguate from the Core.Abstractions namespace's same-named type. Internal RaiseDataChange/RaiseAlarmEvent/RaiseRediscoveryNeeded/OnHostConnectivityUpdate methods are the entry points the IPC client will invoke when push frames arrive. Host side: new Backend/IGalaxyBackend interface defines the seam between IPC dispatch and the live MXAccess code (so the dispatcher is unit-testable against an in-memory mock without needing live Galaxy); Backend/StubGalaxyBackend returns success for OpenSession/CloseSession/Subscribe/Unsubscribe/AlarmSubscribe/AlarmAck/Recycle and a recognizable "stub: MXAccess code lift pending (Phase 2 Task B.1)"-tagged error for Discover/ReadValues/WriteValues/HistoryRead — keeps the IPC end-to-end testable today and gives the parity team a clear seam to slot the real implementation into; Ipc/GalaxyFrameHandler is the new real dispatcher (replaces StubFrameHandler in Program.cs) — switch on MessageKind, deserialize the matching contract, await backend method, write the response (one-way for Unsubscribe/AlarmSubscribe/AlarmAck/CloseSession), heartbeat handled inline so liveness still works if the backend is sick, exceptions caught and surfaced as ErrorResponse with code "handler-exception" so the Proxy raises GalaxyIpcException instead of disconnecting. End-to-end IPC integration test (EndToEndIpcTests) drives every operation through the full stack — Initialize → Read → Write → Subscribe → Unsubscribe → SubscribeAlarms → AlarmAck → ReadRaw → ReadProcessed (short-circuit) — proving the wire protocol, dispatcher, capability forwarding, and one-way semantics agree end-to-end. Skipped on Windows administrator shells per the same PipeAcl-denies-Administrators reasoning the IpcHandshakeIntegrationTests use. Full solution 952 pass / 1 pre-existing Phase 0 baseline. Phase 2 evidence doc updated: status header now reads "Streams A+B+C complete... Streams D+E remain — gated only on the iterative Galaxy code lift + parity-debug cycle"; new Update 2026-04-17 (later) callout enumerates the upgrade with explicit "what's left for the Phase 2 exit gate" — replace StubGalaxyBackend with a MxAccessClient-backed implementation calling on the StaPump, then run the v1 IntegrationTests against the v2 topology and iterate on parity defects until green, then delete legacy OtOpcUa.Host.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Joseph Doherty
@@ -1,25 +1,43 @@
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Proxy.Ipc;
|
||||
using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Shared.Contracts;
|
||||
using IpcHostConnectivityStatus = ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Shared.Contracts.HostConnectivityStatus;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Proxy;
|
||||
|
||||
/// <summary>
|
||||
/// <see cref="IDriver"/> implementation that forwards every capability over the Galaxy IPC
|
||||
/// channel to the out-of-process Host. Implements <see cref="ITagDiscovery"/> as the
|
||||
/// Phase 2 minimum; other capability interfaces (<see cref="IReadable"/>, etc.) will be wired
|
||||
/// in once the Host's MXAccess code lift is complete and end-to-end parity tests run.
|
||||
/// channel to the out-of-process Host. Implements the full Phase 2 capability surface;
|
||||
/// bodies that depend on the deferred Host-side MXAccess code lift will surface
|
||||
/// <see cref="GalaxyIpcException"/> with code <c>not-implemented</c> until the Host's
|
||||
/// <c>IGalaxyBackend</c> is wired to the real <c>MxAccessClient</c>.
|
||||
/// </summary>
|
||||
public sealed class GalaxyProxyDriver(GalaxyProxyOptions options)
|
||||
: IDriver, ITagDiscovery, IDisposable
|
||||
: IDriver,
|
||||
ITagDiscovery,
|
||||
IReadable,
|
||||
IWritable,
|
||||
ISubscribable,
|
||||
IAlarmSource,
|
||||
IHistoryProvider,
|
||||
IRediscoverable,
|
||||
IHostConnectivityProbe,
|
||||
IDisposable
|
||||
{
|
||||
private GalaxyIpcClient? _client;
|
||||
private long _sessionId;
|
||||
private DriverHealth _health = new(DriverState.Unknown, null, null);
|
||||
|
||||
private IReadOnlyList<Core.Abstractions.HostConnectivityStatus> _hostStatuses = [];
|
||||
|
||||
public string DriverInstanceId => options.DriverInstanceId;
|
||||
public string DriverType => "Galaxy";
|
||||
|
||||
public event EventHandler<DataChangeEventArgs>? OnDataChange;
|
||||
public event EventHandler<AlarmEventArgs>? OnAlarmEvent;
|
||||
public event EventHandler<RediscoveryEventArgs>? OnRediscoveryNeeded;
|
||||
public event EventHandler<HostStatusChangedEventArgs>? OnHostStatusChanged;
|
||||
|
||||
public async Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken)
|
||||
{
|
||||
_health = new DriverHealth(DriverState.Initializing, null, null);
|
||||
@@ -59,9 +77,10 @@ public sealed class GalaxyProxyDriver(GalaxyProxyOptions options)
|
||||
|
||||
try
|
||||
{
|
||||
await _client.CallAsync<CloseSessionRequest, ErrorResponse>(
|
||||
MessageKind.CloseSessionRequest, new CloseSessionRequest { SessionId = _sessionId },
|
||||
MessageKind.ErrorResponse, cancellationToken);
|
||||
await _client.SendOneWayAsync(
|
||||
MessageKind.CloseSessionRequest,
|
||||
new CloseSessionRequest { SessionId = _sessionId },
|
||||
cancellationToken);
|
||||
}
|
||||
catch { /* shutdown is best effort */ }
|
||||
|
||||
@@ -71,17 +90,17 @@ public sealed class GalaxyProxyDriver(GalaxyProxyOptions options)
|
||||
}
|
||||
|
||||
public DriverHealth GetHealth() => _health;
|
||||
|
||||
public long GetMemoryFootprint() => 0; // Tier C footprint is reported by the Host over IPC
|
||||
|
||||
public long GetMemoryFootprint() => 0;
|
||||
public Task FlushOptionalCachesAsync(CancellationToken cancellationToken) => Task.CompletedTask;
|
||||
|
||||
// ---- ITagDiscovery ----
|
||||
|
||||
public async Task DiscoverAsync(IAddressSpaceBuilder builder, CancellationToken cancellationToken)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(builder);
|
||||
if (_client is null) throw new InvalidOperationException("Driver not initialized");
|
||||
var client = RequireClient();
|
||||
|
||||
var resp = await _client.CallAsync<DiscoverHierarchyRequest, DiscoverHierarchyResponse>(
|
||||
var resp = await client.CallAsync<DiscoverHierarchyRequest, DiscoverHierarchyResponse>(
|
||||
MessageKind.DiscoverHierarchyRequest,
|
||||
new DiscoverHierarchyRequest { SessionId = _sessionId },
|
||||
MessageKind.DiscoverHierarchyResponse,
|
||||
@@ -109,6 +128,245 @@ public sealed class GalaxyProxyDriver(GalaxyProxyOptions options)
|
||||
}
|
||||
}
|
||||
|
||||
// ---- IReadable ----
|
||||
|
||||
public async Task<IReadOnlyList<DataValueSnapshot>> ReadAsync(
|
||||
IReadOnlyList<string> fullReferences, CancellationToken cancellationToken)
|
||||
{
|
||||
var client = RequireClient();
|
||||
var resp = await client.CallAsync<ReadValuesRequest, ReadValuesResponse>(
|
||||
MessageKind.ReadValuesRequest,
|
||||
new ReadValuesRequest { SessionId = _sessionId, TagReferences = [.. fullReferences] },
|
||||
MessageKind.ReadValuesResponse,
|
||||
cancellationToken);
|
||||
|
||||
if (!resp.Success)
|
||||
throw new InvalidOperationException($"Galaxy.Host ReadValues failed: {resp.Error}");
|
||||
|
||||
var byRef = resp.Values.ToDictionary(v => v.TagReference);
|
||||
var result = new DataValueSnapshot[fullReferences.Count];
|
||||
for (var i = 0; i < fullReferences.Count; i++)
|
||||
{
|
||||
result[i] = byRef.TryGetValue(fullReferences[i], out var v)
|
||||
? ToSnapshot(v)
|
||||
: new DataValueSnapshot(null, StatusBadInternalError, null, DateTime.UtcNow);
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
// ---- IWritable ----
|
||||
|
||||
public async Task<IReadOnlyList<WriteResult>> WriteAsync(
|
||||
IReadOnlyList<WriteRequest> writes, CancellationToken cancellationToken)
|
||||
{
|
||||
var client = RequireClient();
|
||||
var resp = await client.CallAsync<WriteValuesRequest, WriteValuesResponse>(
|
||||
MessageKind.WriteValuesRequest,
|
||||
new WriteValuesRequest
|
||||
{
|
||||
SessionId = _sessionId,
|
||||
Writes = [.. writes.Select(FromWriteRequest)],
|
||||
},
|
||||
MessageKind.WriteValuesResponse,
|
||||
cancellationToken);
|
||||
|
||||
return [.. resp.Results.Select(r => new WriteResult(r.StatusCode))];
|
||||
}
|
||||
|
||||
// ---- ISubscribable ----
|
||||
|
||||
public async Task<ISubscriptionHandle> SubscribeAsync(
|
||||
IReadOnlyList<string> fullReferences, TimeSpan publishingInterval, CancellationToken cancellationToken)
|
||||
{
|
||||
var client = RequireClient();
|
||||
var resp = await client.CallAsync<SubscribeRequest, SubscribeResponse>(
|
||||
MessageKind.SubscribeRequest,
|
||||
new SubscribeRequest
|
||||
{
|
||||
SessionId = _sessionId,
|
||||
TagReferences = [.. fullReferences],
|
||||
RequestedIntervalMs = (int)publishingInterval.TotalMilliseconds,
|
||||
},
|
||||
MessageKind.SubscribeResponse,
|
||||
cancellationToken);
|
||||
|
||||
if (!resp.Success)
|
||||
throw new InvalidOperationException($"Galaxy.Host Subscribe failed: {resp.Error}");
|
||||
|
||||
return new GalaxySubscriptionHandle(resp.SubscriptionId);
|
||||
}
|
||||
|
||||
public async Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken)
|
||||
{
|
||||
var client = RequireClient();
|
||||
var sid = ((GalaxySubscriptionHandle)handle).SubscriptionId;
|
||||
await client.SendOneWayAsync(
|
||||
MessageKind.UnsubscribeRequest,
|
||||
new UnsubscribeRequest { SessionId = _sessionId, SubscriptionId = sid },
|
||||
cancellationToken);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Internal entry point used by the IPC client when the Host pushes an
|
||||
/// <see cref="MessageKind.OnDataChangeNotification"/> frame. Surfaces it as a managed
|
||||
/// <see cref="OnDataChange"/> event.
|
||||
/// </summary>
|
||||
internal void RaiseDataChange(OnDataChangeNotification notif)
|
||||
{
|
||||
var handle = new GalaxySubscriptionHandle(notif.SubscriptionId);
|
||||
// ISubscribable.OnDataChange fires once per changed attribute — fan out the batch.
|
||||
foreach (var v in notif.Values)
|
||||
OnDataChange?.Invoke(this, new DataChangeEventArgs(handle, v.TagReference, ToSnapshot(v)));
|
||||
}
|
||||
|
||||
// ---- IAlarmSource ----
|
||||
|
||||
public async Task<IAlarmSubscriptionHandle> SubscribeAlarmsAsync(
|
||||
IReadOnlyList<string> sourceNodeIds, CancellationToken cancellationToken)
|
||||
{
|
||||
var client = RequireClient();
|
||||
await client.SendOneWayAsync(
|
||||
MessageKind.AlarmSubscribeRequest,
|
||||
new AlarmSubscribeRequest { SessionId = _sessionId },
|
||||
cancellationToken);
|
||||
return new GalaxyAlarmSubscriptionHandle($"alarm-{_sessionId}");
|
||||
}
|
||||
|
||||
public Task UnsubscribeAlarmsAsync(IAlarmSubscriptionHandle handle, CancellationToken cancellationToken)
|
||||
=> Task.CompletedTask;
|
||||
|
||||
public async Task AcknowledgeAsync(
|
||||
IReadOnlyList<AlarmAcknowledgeRequest> acknowledgements, CancellationToken cancellationToken)
|
||||
{
|
||||
var client = RequireClient();
|
||||
foreach (var ack in acknowledgements)
|
||||
{
|
||||
await client.SendOneWayAsync(
|
||||
MessageKind.AlarmAckRequest,
|
||||
new AlarmAckRequest
|
||||
{
|
||||
SessionId = _sessionId,
|
||||
EventId = ack.ConditionId,
|
||||
Comment = ack.Comment ?? string.Empty,
|
||||
},
|
||||
cancellationToken);
|
||||
}
|
||||
}
|
||||
|
||||
internal void RaiseAlarmEvent(GalaxyAlarmEvent ev)
|
||||
{
|
||||
var handle = new GalaxyAlarmSubscriptionHandle($"alarm-{_sessionId}");
|
||||
OnAlarmEvent?.Invoke(this, new AlarmEventArgs(
|
||||
SubscriptionHandle: handle,
|
||||
SourceNodeId: ev.ObjectTagName,
|
||||
ConditionId: ev.EventId,
|
||||
AlarmType: ev.AlarmName,
|
||||
Message: ev.Message,
|
||||
Severity: MapSeverity(ev.Severity),
|
||||
SourceTimestampUtc: DateTimeOffset.FromUnixTimeMilliseconds(ev.UtcUnixMs).UtcDateTime));
|
||||
}
|
||||
|
||||
// ---- IHistoryProvider ----
|
||||
|
||||
public async Task<HistoryReadResult> ReadRawAsync(
|
||||
string fullReference, DateTime startUtc, DateTime endUtc, uint maxValuesPerNode,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
var client = RequireClient();
|
||||
var resp = await client.CallAsync<HistoryReadRequest, HistoryReadResponse>(
|
||||
MessageKind.HistoryReadRequest,
|
||||
new HistoryReadRequest
|
||||
{
|
||||
SessionId = _sessionId,
|
||||
TagReferences = [fullReference],
|
||||
StartUtcUnixMs = new DateTimeOffset(startUtc, TimeSpan.Zero).ToUnixTimeMilliseconds(),
|
||||
EndUtcUnixMs = new DateTimeOffset(endUtc, TimeSpan.Zero).ToUnixTimeMilliseconds(),
|
||||
MaxValuesPerTag = maxValuesPerNode,
|
||||
},
|
||||
MessageKind.HistoryReadResponse,
|
||||
cancellationToken);
|
||||
|
||||
if (!resp.Success)
|
||||
throw new InvalidOperationException($"Galaxy.Host HistoryRead failed: {resp.Error}");
|
||||
|
||||
var first = resp.Tags.FirstOrDefault();
|
||||
IReadOnlyList<DataValueSnapshot> samples = first is null
|
||||
? Array.Empty<DataValueSnapshot>()
|
||||
: [.. first.Values.Select(ToSnapshot)];
|
||||
return new HistoryReadResult(samples, ContinuationPoint: null);
|
||||
}
|
||||
|
||||
public Task<HistoryReadResult> ReadProcessedAsync(
|
||||
string fullReference, DateTime startUtc, DateTime endUtc, TimeSpan interval,
|
||||
HistoryAggregateType aggregate, CancellationToken cancellationToken)
|
||||
=> throw new NotSupportedException("Galaxy historian processed reads are not supported in v2; use ReadRawAsync.");
|
||||
|
||||
// ---- IRediscoverable ----
|
||||
|
||||
/// <summary>
|
||||
/// Triggered by the IPC client when the Host pushes a deploy-watermark notification
|
||||
/// (Galaxy <c>time_of_last_deploy</c> changed per decision #54).
|
||||
/// </summary>
|
||||
internal void RaiseRediscoveryNeeded(string reason, string? scopeHint = null) =>
|
||||
OnRediscoveryNeeded?.Invoke(this, new RediscoveryEventArgs(reason, scopeHint));
|
||||
|
||||
// ---- IHostConnectivityProbe ----
|
||||
|
||||
public IReadOnlyList<Core.Abstractions.HostConnectivityStatus> GetHostStatuses() => _hostStatuses;
|
||||
|
||||
internal void OnHostConnectivityUpdate(IpcHostConnectivityStatus update)
|
||||
{
|
||||
var translated = new Core.Abstractions.HostConnectivityStatus(
|
||||
HostName: update.HostName,
|
||||
State: ParseHostState(update.RuntimeStatus),
|
||||
LastChangedUtc: DateTimeOffset.FromUnixTimeMilliseconds(update.LastObservedUtcUnixMs).UtcDateTime);
|
||||
|
||||
var prior = _hostStatuses.FirstOrDefault(h => h.HostName == translated.HostName);
|
||||
_hostStatuses = [
|
||||
.. _hostStatuses.Where(h => h.HostName != translated.HostName),
|
||||
translated
|
||||
];
|
||||
|
||||
if (prior is null || prior.State != translated.State)
|
||||
{
|
||||
OnHostStatusChanged?.Invoke(this, new HostStatusChangedEventArgs(
|
||||
translated.HostName, prior?.State ?? HostState.Unknown, translated.State));
|
||||
}
|
||||
}
|
||||
|
||||
private static HostState ParseHostState(string s) => s switch
|
||||
{
|
||||
"Running" => HostState.Running,
|
||||
"Stopped" => HostState.Stopped,
|
||||
"Faulted" => HostState.Faulted,
|
||||
_ => HostState.Unknown,
|
||||
};
|
||||
|
||||
// ---- helpers ----
|
||||
|
||||
private GalaxyIpcClient RequireClient() =>
|
||||
_client ?? throw new InvalidOperationException("Driver not initialized");
|
||||
|
||||
private const uint StatusBadInternalError = 0x80020000u;
|
||||
|
||||
private static DataValueSnapshot ToSnapshot(GalaxyDataValue v) => new(
|
||||
Value: v.ValueBytes,
|
||||
StatusCode: v.StatusCode,
|
||||
SourceTimestampUtc: v.SourceTimestampUtcUnixMs > 0
|
||||
? DateTimeOffset.FromUnixTimeMilliseconds(v.SourceTimestampUtcUnixMs).UtcDateTime
|
||||
: null,
|
||||
ServerTimestampUtc: DateTimeOffset.FromUnixTimeMilliseconds(v.ServerTimestampUtcUnixMs).UtcDateTime);
|
||||
|
||||
private static GalaxyDataValue FromWriteRequest(WriteRequest w) => new()
|
||||
{
|
||||
TagReference = w.FullReference,
|
||||
ValueBytes = MessagePack.MessagePackSerializer.Serialize(w.Value),
|
||||
ValueMessagePackType = 0,
|
||||
StatusCode = 0,
|
||||
SourceTimestampUtcUnixMs = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(),
|
||||
ServerTimestampUtcUnixMs = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(),
|
||||
};
|
||||
|
||||
private static DriverDataType MapDataType(int mxDataType) => mxDataType switch
|
||||
{
|
||||
0 => DriverDataType.Boolean,
|
||||
@@ -132,9 +390,27 @@ public sealed class GalaxyProxyDriver(GalaxyProxyOptions options)
|
||||
_ => SecurityClassification.FreeAccess,
|
||||
};
|
||||
|
||||
private static AlarmSeverity MapSeverity(int sev) => sev switch
|
||||
{
|
||||
<= 250 => AlarmSeverity.Low,
|
||||
<= 500 => AlarmSeverity.Medium,
|
||||
<= 800 => AlarmSeverity.High,
|
||||
_ => AlarmSeverity.Critical,
|
||||
};
|
||||
|
||||
public void Dispose() => _client?.DisposeAsync().AsTask().GetAwaiter().GetResult();
|
||||
}
|
||||
|
||||
internal sealed record GalaxySubscriptionHandle(long SubscriptionId) : ISubscriptionHandle
|
||||
{
|
||||
public string DiagnosticId => $"galaxy-sub-{SubscriptionId}";
|
||||
}
|
||||
|
||||
internal sealed record GalaxyAlarmSubscriptionHandle(string Id) : IAlarmSubscriptionHandle
|
||||
{
|
||||
public string DiagnosticId => Id;
|
||||
}
|
||||
|
||||
public sealed class GalaxyProxyOptions
|
||||
{
|
||||
public required string DriverInstanceId { get; init; }
|
||||
|
||||
Reference in New Issue
Block a user