Phase 7 follow-up #247 — Galaxy.Host historian writer + SQLite sink activation
Closes the historian leg of Phase 7. Scripted alarm transitions now batch-flow through the existing Galaxy.Host pipe + queue durably in a local SQLite store- and-forward when Galaxy is the registered driver, instead of being dropped into NullAlarmHistorianSink. ## GalaxyHistorianWriter (Driver.Galaxy.Proxy.Ipc) IAlarmHistorianWriter implementation. Translates AlarmHistorianEvent → HistorianAlarmEventDto (Stream D contract), batches via the existing GalaxyIpcClient.CallAsync round-trip on MessageKind.HistorianAlarmEventRequest / Response, maps per-event HistorianAlarmEventOutcomeDto bytes back to HistorianWriteOutcome (Ack/RetryPlease/PermanentFail) so the SQLite drain worker knows what to ack vs dead-letter vs retry. Empty-batch fast path. Pipe-level transport faults (broken pipe, host crash) bubble up as GalaxyIpcException which the SQLite sink's drain worker translates to whole-batch RetryPlease per its catch contract. ## GalaxyProxyDriver implements IAlarmHistorianWriter Marker interface lets Phase7Composer discover it via type check at compose time. WriteBatchAsync delegates to a thin GalaxyHistorianWriter wrapping the driver's existing _client. Throws InvalidOperationException if InitializeAsync hasn't connected yet — the SQLite drain worker treats that as a transient batch failure and retries. ## Phase7Composer.ResolveHistorianSink Replaces the injected sink dep when any registered driver implements IAlarmHistorianWriter. Constructs SqliteStoreAndForwardSink at %ProgramData%/OtOpcUa/alarm-historian-queue.db (falls back to %TEMP% when ProgramData unavailable, e.g. dev), starts the 2s drain timer, owns the sink disposable for clean teardown. When no driver provides the writer, keeps the NullAlarmHistorianSink wired by Program.cs (#246). DisposeAsync now also disposes the owned SQLite sink in the right order: bridge → engines → owned sink → injected fallback. ## Tests — 7 new GalaxyHistorianWriterMappingTests ToDto round-trips every field; preserves null Comment; per-byte outcome enum mapping (Ack / RetryPlease / PermanentFail) via [Theory]; unknown byte throws; ctor null-guard. The IPC round-trip itself is covered by the live Host suite (task #240) which constructs a real pipe. Server.Phase7 tests: 34/34 still pass; Galaxy.Proxy tests: 25/25 (+7 = 32 total). ## Phase 7 production wiring chain — COMPLETE - ✅ #243 composition kernel - ✅ #245 scripted-alarm IReadable adapter - ✅ #244 driver bridge - ✅ #246 Program.cs wire-in - ✅ #247 this — Galaxy.Host historian writer + SQLite sink activation What unblocks now: task #240 live OPC UA E2E smoke. With a Galaxy driver registered, scripted alarm transitions flow end-to-end through the engine → SQLite queue → drain worker → Galaxy.Host IPC → Aveva Historian alarm schema. Without Galaxy, NullSink keeps the engines functional and the queue dormant.
This commit is contained in:
@@ -0,0 +1,90 @@
|
||||
using ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian;
|
||||
using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Shared;
|
||||
using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Shared.Contracts;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Proxy.Ipc;
|
||||
|
||||
/// <summary>
|
||||
/// Phase 7 follow-up (task #247) — bridges <see cref="SqliteStoreAndForwardSink"/>'s
|
||||
/// drain worker to <c>Driver.Galaxy.Host</c> over the existing <see cref="GalaxyIpcClient"/>
|
||||
/// pipe. Translates <see cref="AlarmHistorianEvent"/> batches into the
|
||||
/// <see cref="HistorianAlarmEventDto"/> wire format the Host expects + maps per-event
|
||||
/// <see cref="HistorianAlarmEventOutcomeDto"/> responses back to
|
||||
/// <see cref="HistorianWriteOutcome"/> so the SQLite queue knows what to ack /
|
||||
/// dead-letter / retry.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// Reuses the IPC channel <see cref="GalaxyProxyDriver"/> already opens for the
|
||||
/// Galaxy data plane — no second pipe to <c>Driver.Galaxy.Host</c>, no separate
|
||||
/// auth handshake. The IPC client's call gate serializes historian batches with
|
||||
/// driver Reads/Writes/Subscribes; historian batches are infrequent (every few
|
||||
/// seconds at most under the SQLite sink's drain cadence) so the contention is
|
||||
/// negligible compared to per-tag-read pressure.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// Pipe-level transport faults (broken pipe, host crash) bubble up as
|
||||
/// <see cref="GalaxyIpcException"/> which the SQLite sink's drain worker catches +
|
||||
/// translates to a whole-batch RetryPlease per the
|
||||
/// <see cref="SqliteStoreAndForwardSink"/> docstring — failed events stay queued
|
||||
/// for the next drain tick after backoff.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
public sealed class GalaxyHistorianWriter : IAlarmHistorianWriter
|
||||
{
|
||||
private readonly GalaxyIpcClient _client;
|
||||
|
||||
public GalaxyHistorianWriter(GalaxyIpcClient client)
|
||||
{
|
||||
_client = client ?? throw new ArgumentNullException(nameof(client));
|
||||
}
|
||||
|
||||
public async Task<IReadOnlyList<HistorianWriteOutcome>> WriteBatchAsync(
|
||||
IReadOnlyList<AlarmHistorianEvent> batch, CancellationToken cancellationToken)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(batch);
|
||||
if (batch.Count == 0) return [];
|
||||
|
||||
var request = new HistorianAlarmEventRequest
|
||||
{
|
||||
Events = batch.Select(ToDto).ToArray(),
|
||||
};
|
||||
|
||||
var response = await _client.CallAsync<HistorianAlarmEventRequest, HistorianAlarmEventResponse>(
|
||||
requestKind: MessageKind.HistorianAlarmEventRequest,
|
||||
request: request,
|
||||
expectedResponseKind: MessageKind.HistorianAlarmEventResponse,
|
||||
ct: cancellationToken).ConfigureAwait(false);
|
||||
|
||||
if (response.Outcomes.Length != batch.Count)
|
||||
throw new InvalidOperationException(
|
||||
$"Galaxy.Host returned {response.Outcomes.Length} outcomes for a batch of {batch.Count} — protocol mismatch");
|
||||
|
||||
var outcomes = new HistorianWriteOutcome[response.Outcomes.Length];
|
||||
for (var i = 0; i < response.Outcomes.Length; i++)
|
||||
outcomes[i] = MapOutcome(response.Outcomes[i]);
|
||||
return outcomes;
|
||||
}
|
||||
|
||||
internal static HistorianAlarmEventDto ToDto(AlarmHistorianEvent e) => new()
|
||||
{
|
||||
AlarmId = e.AlarmId,
|
||||
EquipmentPath = e.EquipmentPath,
|
||||
AlarmName = e.AlarmName,
|
||||
AlarmTypeName = e.AlarmTypeName,
|
||||
Severity = (int)e.Severity,
|
||||
EventKind = e.EventKind,
|
||||
Message = e.Message,
|
||||
User = e.User,
|
||||
Comment = e.Comment,
|
||||
TimestampUtcUnixMs = new DateTimeOffset(e.TimestampUtc, TimeSpan.Zero).ToUnixTimeMilliseconds(),
|
||||
};
|
||||
|
||||
internal static HistorianWriteOutcome MapOutcome(HistorianAlarmEventOutcomeDto wire) => wire switch
|
||||
{
|
||||
HistorianAlarmEventOutcomeDto.Ack => HistorianWriteOutcome.Ack,
|
||||
HistorianAlarmEventOutcomeDto.RetryPlease => HistorianWriteOutcome.RetryPlease,
|
||||
HistorianAlarmEventOutcomeDto.PermanentFail => HistorianWriteOutcome.PermanentFail,
|
||||
_ => throw new InvalidOperationException($"Unknown HistorianAlarmEventOutcomeDto byte {(byte)wire}"),
|
||||
};
|
||||
}
|
||||
Reference in New Issue
Block a user