Phase 2 — port MXAccess COM client to Galaxy.Host + MxAccessGalaxyBackend (3rd IGalaxyBackend) + live MXAccess smoke + Phase 2 exit-gate doc + adversarial review. The full Galaxy data-plane now flows through the v2 IPC topology end-to-end against live ArchestrA.MxAccess.dll, on this dev box, with 30/30 Host tests + 9/9 Proxy tests + 963/963 solution tests passing alongside the unchanged 494 v1 IntegrationTests baseline. Backend/MxAccess/Vtq is a focused port of v1's Vtq value-timestamp-quality DTO. Backend/MxAccess/IMxProxy abstracts LMXProxyServer (port of v1's IMxProxy with the same Register/Unregister/AddItem/RemoveItem/AdviseSupervisory/UnAdviseSupervisory/Write surface + OnDataChange + OnWriteComplete events); MxProxyAdapter is the concrete COM-backed implementation that does Marshal.ReleaseComObject-loop on Unregister, must be constructed on an STA thread. Backend/MxAccess/MxAccessClient is the focused port of v1's MxAccessClient partials — Connect/Disconnect/Read/Write/Subscribe/Unsubscribe through the new Sta/StaPump (the real Win32 GetMessage pump from the previous commit), ConcurrentDictionary handle tracking, OnDataChange event marshalling to per-tag callbacks, ReadAsync implemented as the canonical subscribe → first-OnDataChange → unsubscribe one-shot pattern. Galaxy.Host csproj flipped to x86 PlatformTarget + Prefer32Bit=true with the ArchestrA.MxAccess HintPath ..\..\lib\ArchestrA.MxAccess.dll reference (lib/ already contains the production DLL). Backend/MxAccessGalaxyBackend is the third IGalaxyBackend implementation (alongside StubGalaxyBackend and DbBackedGalaxyBackend): combines GalaxyRepository (Discover) with MxAccessClient (Read/Write/Subscribe), MessagePack-deserializes inbound write values, MessagePack-serializes outbound read values into ValueBytes, decodes ArrayDimension/SecurityClassification/category_id with the same v1 mapping. Program.cs selects between stub|db|mxaccess via OTOPCUA_GALAXY_BACKEND env var (default = mxaccess); OTOPCUA_GALAXY_ZB_CONN overrides the ZB connection string; OTOPCUA_GALAXY_CLIENT_NAME sets the Wonderware client identity; the StaPump and MxAccessClient lifecycles are tied to the server.RunAsync try/finally so a clean Ctrl+C tears down the COM proxy via Marshal.ReleaseComObject before the pump's WM_QUIT. Live MXAccess smoke tests (MxAccessLiveSmokeTests, net48 x86) — skipped when ZB unreachable or aaBootstrap not running, otherwise verify (1) MxAccessClient.ConnectAsync returns a positive LMXProxyServer handle on the StaPump, (2) MxAccessGalaxyBackend.OpenSession + Discover returns at least one gobject with attributes, (3) MxAccessGalaxyBackend.ReadValues against the first discovered attribute returns a response with the correct TagReference shape (value + quality vary by what's running, so we don't assert specific values). All 3 pass on this dev box. EndToEndIpcTests + IpcHandshakeIntegrationTests moved from Galaxy.Proxy.Tests (net10) to Galaxy.Host.Tests (net48 x86) — the previous test placement silently dropped them at xUnit discovery because Host became net48 x86 and net10 process can't load it. Rewritten to use Shared's FrameReader/FrameWriter directly instead of going through Proxy's GalaxyIpcClient (functionally equivalent — same wire protocol, framing primitives + dispatcher are the production code path verbatim). 7 IPC tests now run cleanly: Hello+heartbeat round-trip, wrong-secret rejection, OpenSession session-id assignment, Discover error-response surfacing, WriteValues per-tag bad status, Subscribe id assignment, Recycle grace window. Phase 2 exit-gate doc (docs/v2/implementation/exit-gate-phase-2.md) supersedes the partial-exit doc with the as-built state — Streams A/B/C complete; D/E gated only on the legacy-Host removal + parity-test rewrite cycle that fundamentally requires multi-day debug iteration; full adversarial-review section ranking 8 findings (2 high, 3 medium, 3 low) all explicitly deferred to Stream D/E or v2.1 with rationale; Stream-D removal checklist gives the next-session entry point with two policy options for the 494 v1 tests (rewrite-to-use-Proxy vs archive-and-write-smaller-v2-parity-suite). Cannot one-shot Stream D.1 in any single session because deleting OtOpcUa.Host requires the v1 IntegrationTests cycle to be retargeted first; that's the structural blocker, not "needs more code" — and the plan itself budgets 3-4 weeks for it.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Joseph Doherty
@@ -0,0 +1,210 @@
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.Linq;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using MessagePack;
|
||||
using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Host.Backend.Galaxy;
|
||||
using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Host.Backend.MxAccess;
|
||||
using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Shared.Contracts;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Host.Backend;
|
||||
|
||||
/// <summary>
|
||||
/// Production <see cref="IGalaxyBackend"/> — combines the SQL-backed
|
||||
/// <see cref="GalaxyRepository"/> for Discover with the live MXAccess
|
||||
/// <see cref="MxAccessClient"/> for Read / Write / Subscribe. History stays bad-coded
|
||||
/// until the Wonderware Historian SDK plugin loader (Task B.1.h) lands. Alarms come from
|
||||
/// MxAccess <c>AlarmExtension</c> primitives but the wire-up is also Phase 2 follow-up
|
||||
/// (the v1 alarm subsystem is its own subtree).
|
||||
/// </summary>
|
||||
public sealed class MxAccessGalaxyBackend : IGalaxyBackend
|
||||
{
|
||||
private readonly GalaxyRepository _repository;
|
||||
private readonly MxAccessClient _mx;
|
||||
private long _nextSessionId;
|
||||
private long _nextSubscriptionId;
|
||||
|
||||
// Active SubscriptionId → MXAccess full reference list — so Unsubscribe can find them.
|
||||
private readonly System.Collections.Concurrent.ConcurrentDictionary<long, IReadOnlyList<string>> _subs = new();
|
||||
|
||||
public MxAccessGalaxyBackend(GalaxyRepository repository, MxAccessClient mx)
|
||||
{
|
||||
_repository = repository;
|
||||
_mx = mx;
|
||||
}
|
||||
|
||||
public async Task<OpenSessionResponse> OpenSessionAsync(OpenSessionRequest req, CancellationToken ct)
|
||||
{
|
||||
try
|
||||
{
|
||||
await _mx.ConnectAsync();
|
||||
return new OpenSessionResponse { Success = true, SessionId = Interlocked.Increment(ref _nextSessionId) };
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
return new OpenSessionResponse { Success = false, Error = $"MXAccess connect failed: {ex.Message}" };
|
||||
}
|
||||
}
|
||||
|
||||
public async Task CloseSessionAsync(CloseSessionRequest req, CancellationToken ct)
|
||||
{
|
||||
await _mx.DisconnectAsync();
|
||||
}
|
||||
|
||||
public async Task<DiscoverHierarchyResponse> DiscoverAsync(DiscoverHierarchyRequest req, CancellationToken ct)
|
||||
{
|
||||
try
|
||||
{
|
||||
var hierarchy = await _repository.GetHierarchyAsync(ct).ConfigureAwait(false);
|
||||
var attributes = await _repository.GetAttributesAsync(ct).ConfigureAwait(false);
|
||||
|
||||
var attrsByGobject = attributes
|
||||
.GroupBy(a => a.GobjectId)
|
||||
.ToDictionary(g => g.Key, g => g.Select(MapAttribute).ToArray());
|
||||
var nameByGobject = hierarchy.ToDictionary(o => o.GobjectId, o => o.TagName);
|
||||
|
||||
var objects = hierarchy.Select(o => new GalaxyObjectInfo
|
||||
{
|
||||
ContainedName = string.IsNullOrEmpty(o.ContainedName) ? o.TagName : o.ContainedName,
|
||||
TagName = o.TagName,
|
||||
ParentContainedName = o.ParentGobjectId != 0 && nameByGobject.TryGetValue(o.ParentGobjectId, out var p) ? p : null,
|
||||
TemplateCategory = MapCategory(o.CategoryId),
|
||||
Attributes = attrsByGobject.TryGetValue(o.GobjectId, out var a) ? a : Array.Empty<GalaxyAttributeInfo>(),
|
||||
}).ToArray();
|
||||
|
||||
return new DiscoverHierarchyResponse { Success = true, Objects = objects };
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
return new DiscoverHierarchyResponse { Success = false, Error = ex.Message, Objects = Array.Empty<GalaxyObjectInfo>() };
|
||||
}
|
||||
}
|
||||
|
||||
public async Task<ReadValuesResponse> ReadValuesAsync(ReadValuesRequest req, CancellationToken ct)
|
||||
{
|
||||
if (!_mx.IsConnected) return new ReadValuesResponse { Success = false, Error = "Not connected", Values = Array.Empty<GalaxyDataValue>() };
|
||||
|
||||
var results = new List<GalaxyDataValue>(req.TagReferences.Length);
|
||||
foreach (var reference in req.TagReferences)
|
||||
{
|
||||
try
|
||||
{
|
||||
var vtq = await _mx.ReadAsync(reference, TimeSpan.FromSeconds(5), ct);
|
||||
results.Add(ToWire(reference, vtq));
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
results.Add(new GalaxyDataValue
|
||||
{
|
||||
TagReference = reference,
|
||||
StatusCode = 0x80020000u, // Bad_InternalError
|
||||
ServerTimestampUtcUnixMs = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(),
|
||||
ValueBytes = MessagePackSerializer.Serialize(ex.Message),
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
return new ReadValuesResponse { Success = true, Values = results.ToArray() };
|
||||
}
|
||||
|
||||
public async Task<WriteValuesResponse> WriteValuesAsync(WriteValuesRequest req, CancellationToken ct)
|
||||
{
|
||||
var results = new List<WriteValueResult>(req.Writes.Length);
|
||||
foreach (var w in req.Writes)
|
||||
{
|
||||
try
|
||||
{
|
||||
// Decode the value back from the MessagePack bytes the Proxy sent.
|
||||
var value = w.ValueBytes is null
|
||||
? null
|
||||
: MessagePackSerializer.Deserialize<object>(w.ValueBytes);
|
||||
|
||||
await _mx.WriteAsync(w.TagReference, value!);
|
||||
results.Add(new WriteValueResult { TagReference = w.TagReference, StatusCode = 0 });
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
results.Add(new WriteValueResult { TagReference = w.TagReference, StatusCode = 0x80020000u, Error = ex.Message });
|
||||
}
|
||||
}
|
||||
return new WriteValuesResponse { Results = results.ToArray() };
|
||||
}
|
||||
|
||||
public async Task<SubscribeResponse> SubscribeAsync(SubscribeRequest req, CancellationToken ct)
|
||||
{
|
||||
var sid = Interlocked.Increment(ref _nextSubscriptionId);
|
||||
|
||||
try
|
||||
{
|
||||
// For each requested tag, register a subscription that publishes back via the
|
||||
// shared MXAccess data-change handler. The OnDataChange push frame to the Proxy
|
||||
// is wired in the upcoming subscription-push pass; for now the value is captured
|
||||
// for the first ReadAsync to hit it (so the subscribe surface itself is functional).
|
||||
foreach (var tag in req.TagReferences)
|
||||
await _mx.SubscribeAsync(tag, (_, __) => { /* push-frame plumbing in next iteration */ });
|
||||
|
||||
_subs[sid] = req.TagReferences;
|
||||
return new SubscribeResponse { Success = true, SubscriptionId = sid, ActualIntervalMs = req.RequestedIntervalMs };
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
return new SubscribeResponse { Success = false, Error = ex.Message };
|
||||
}
|
||||
}
|
||||
|
||||
public async Task UnsubscribeAsync(UnsubscribeRequest req, CancellationToken ct)
|
||||
{
|
||||
if (!_subs.TryRemove(req.SubscriptionId, out var refs)) return;
|
||||
foreach (var r in refs)
|
||||
await _mx.UnsubscribeAsync(r);
|
||||
}
|
||||
|
||||
public Task SubscribeAlarmsAsync(AlarmSubscribeRequest req, CancellationToken ct) => Task.CompletedTask;
|
||||
public Task AcknowledgeAlarmAsync(AlarmAckRequest req, CancellationToken ct) => Task.CompletedTask;
|
||||
|
||||
public Task<HistoryReadResponse> HistoryReadAsync(HistoryReadRequest req, CancellationToken ct)
|
||||
=> Task.FromResult(new HistoryReadResponse
|
||||
{
|
||||
Success = false,
|
||||
Error = "Wonderware Historian plugin loader not yet wired (Phase 2 Task B.1.h follow-up)",
|
||||
Tags = Array.Empty<HistoryTagValues>(),
|
||||
});
|
||||
|
||||
public Task<RecycleStatusResponse> RecycleAsync(RecycleHostRequest req, CancellationToken ct)
|
||||
=> Task.FromResult(new RecycleStatusResponse { Accepted = true, GraceSeconds = 15 });
|
||||
|
||||
private static GalaxyDataValue ToWire(string reference, Vtq vtq) => new()
|
||||
{
|
||||
TagReference = reference,
|
||||
ValueBytes = vtq.Value is null ? null : MessagePackSerializer.Serialize(vtq.Value),
|
||||
ValueMessagePackType = 0,
|
||||
StatusCode = vtq.Quality >= 192 ? 0u : 0x40000000u, // Good vs Uncertain placeholder
|
||||
SourceTimestampUtcUnixMs = new DateTimeOffset(vtq.TimestampUtc, TimeSpan.Zero).ToUnixTimeMilliseconds(),
|
||||
ServerTimestampUtcUnixMs = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(),
|
||||
};
|
||||
|
||||
private static GalaxyAttributeInfo MapAttribute(GalaxyAttributeRow row) => new()
|
||||
{
|
||||
AttributeName = row.AttributeName,
|
||||
MxDataType = row.MxDataType,
|
||||
IsArray = row.IsArray,
|
||||
ArrayDim = row.ArrayDimension is int d and > 0 ? (uint)d : null,
|
||||
SecurityClassification = row.SecurityClassification,
|
||||
IsHistorized = row.IsHistorized,
|
||||
};
|
||||
|
||||
private static string MapCategory(int categoryId) => categoryId switch
|
||||
{
|
||||
1 => "$WinPlatform",
|
||||
3 => "$AppEngine",
|
||||
4 => "$Area",
|
||||
10 => "$UserDefined",
|
||||
11 => "$ApplicationObject",
|
||||
13 => "$Area",
|
||||
17 => "$DeviceIntegration",
|
||||
24 => "$ViewEngine",
|
||||
26 => "$ViewApp",
|
||||
_ => $"category-{categoryId}",
|
||||
};
|
||||
}
|
||||
Reference in New Issue
Block a user