feat(historian-gateway): read cutover — AddServerHistorian builds GatewayHistorianDataSource

Claude-Session: https://claude.ai/code/session_012SDSQ3AcaXqPcBtDESBRii
2026-06-26 17:07:59 -04:00
parent 1d5fa8230e
commit 36f7c3c5bf
7 changed files with 213 additions and 9 deletions
@@ -0,0 +1,126 @@
+using Microsoft.Extensions.Logging;
+using ZB.MOM.WW.HistorianGateway.Client;
+using ZB.MOM.WW.HistorianGateway.Contracts.Grpc;
+using ZB.MOM.WW.OtOpcUa.Runtime.Historian;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway;
+
+/// <summary>
+///     Concrete <see cref="IHistorianGatewayClient"/> backed by the published
+///     <see cref="HistorianGatewayClient"/> package client. Each seam method forwards directly to the
+///     matching client wrapper — both sides speak the same generated <c>historian_gateway.v1</c> proto
+///     types, so no shape translation happens here. The package client's typed exception hierarchy
+///     (<c>HistorianGatewayUnavailableException</c> et al.) is allowed to surface unchanged; the
+///     <see cref="GatewayHistorianDataSource"/> records it as a health failure and the node manager
+///     turns it into a Bad HistoryRead result.
+/// </summary>
+/// <remarks>
+///     <para>
+///         <b>Lazy channel.</b> <see cref="Create"/> calls <see cref="HistorianGatewayClient.Create"/>,
+///         which constructs a <c>GrpcChannel</c> over a <c>SocketsHttpHandler</c> without opening a
+///         connection — the first RPC dials. Constructing the adapter therefore performs no network I/O,
+///         which the offline seam tests rely on (they build from bogus endpoints and must not connect).
+///     </para>
+/// </remarks>
+public sealed class HistorianGatewayClientAdapter : IHistorianGatewayClient, IDisposable
+{
+    private readonly HistorianGatewayClient _inner;
+
+    private HistorianGatewayClientAdapter(HistorianGatewayClient inner) => _inner = inner;
+
+    /// <summary>
+    ///     Builds an adapter over a freshly created package client mapped from the bound
+    ///     <see cref="ServerHistorianOptions"/>. No connection is opened (lazy channel).
+    /// </summary>
+    /// <param name="options">The bound <c>ServerHistorian</c> configuration (endpoint, key, TLS posture).</param>
+    /// <param name="loggerFactory">Logger factory threaded into the package client's channel diagnostics.</param>
+    /// <returns>A ready-to-use adapter whose underlying channel has not yet dialed the gateway.</returns>
+    public static HistorianGatewayClientAdapter Create(ServerHistorianOptions options, ILoggerFactory loggerFactory)
+    {
+        ArgumentNullException.ThrowIfNull(options);
+        ArgumentNullException.ThrowIfNull(loggerFactory);
+
+        var clientOptions = new HistorianGatewayClientOptions
+        {
+            Endpoint = new Uri(options.Endpoint),
+            ApiKey = options.ApiKey,
+            UseTls = options.UseTls,
+            CaCertificatePath = options.CaCertificatePath,
+            // INVERTED mapping: ServerHistorianOptions.AllowUntrustedServerCertificate (opt-in to accept
+            // a self-signed cert) is the negation of the client's RequireCertificateValidation. Allowing
+            // an untrusted cert == not requiring validation; a pinned CaCertificatePath always verifies.
+            RequireCertificateValidation = !options.AllowUntrustedServerCertificate,
+            DefaultCallTimeout = options.CallTimeout,
+            LoggerFactory = loggerFactory,
+        };
+
+        return new HistorianGatewayClientAdapter(HistorianGatewayClient.Create(clientOptions));
+    }
+
+    /// <inheritdoc />
+    public IAsyncEnumerable<HistorianSample> ReadRawAsync(
+        string tag, DateTime startUtc, DateTime endUtc, int maxValues, CancellationToken ct) =>
+        _inner.ReadRawAsync(tag, startUtc, endUtc, maxValues, ct);
+
+    /// <inheritdoc />
+    public IAsyncEnumerable<HistorianAggregateSample> ReadAggregateAsync(
+        string tag, DateTime startUtc, DateTime endUtc, RetrievalMode mode, TimeSpan interval, CancellationToken ct) =>
+        _inner.ReadAggregateAsync(tag, startUtc, endUtc, mode, interval, ct);
+
+    /// <inheritdoc />
+    public Task<IReadOnlyList<HistorianSample>> ReadAtTimeAsync(
+        string tag, IReadOnlyList<DateTime> timestampsUtc, CancellationToken ct) =>
+        _inner.ReadAtTimeAsync(tag, timestampsUtc, ct);
+
+    /// <inheritdoc />
+    /// <remarks>
+    ///     <paramref name="sourceName"/> is rendered into the gateway's one server-filterable predicate —
+    ///     a <c>Source_Object</c> <see cref="HistorianEventComparison.Equal"/> filter the SQL ReadEvents
+    ///     path binds as <c>WHERE Source_Object = @source</c>. A <c>null</c> source passes a null filter
+    ///     (full window). <paramref name="maxEvents"/> is intentionally ignored here: the gateway wire
+    ///     contract carries no per-call cap, so the cap is enforced upstream by
+    ///     <see cref="GatewayHistorianDataSource"/> via early stream termination.
+    /// </remarks>
+    public IAsyncEnumerable<HistorianEvent> ReadEventsAsync(
+        string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents, CancellationToken ct)
+    {
+        HistorianEventFilter? filter = sourceName is null
+            ? null
+            : new HistorianEventFilter
+            {
+                PropertyName = "Source_Object",
+                Comparison = HistorianEventComparison.Equal,
+                Value = sourceName,
+            };
+
+        return _inner.ReadEventsAsync(startUtc, endUtc, filter, ct);
+    }
+
+    /// <inheritdoc />
+    public Task<WriteAck> WriteLiveValuesAsync(
+        string tag, IReadOnlyList<HistorianLiveValue> values, CancellationToken ct) =>
+        _inner.WriteLiveValuesAsync(tag, values, ct);
+
+    /// <inheritdoc />
+    public Task<WriteAck> SendEventAsync(HistorianEvent evt, CancellationToken ct) =>
+        _inner.SendEventAsync(evt, ct);
+
+    /// <inheritdoc />
+    public Task<TagOperationResults> EnsureTagsAsync(
+        IReadOnlyList<HistorianTagDefinition> definitions, CancellationToken ct) =>
+        _inner.EnsureTagsAsync(definitions, ct);
+
+    /// <inheritdoc />
+    public Task<bool> ProbeAsync(CancellationToken ct) => _inner.ProbeAsync(ct);
+
+    /// <inheritdoc />
+    public Task<ConnectionStatus> GetConnectionStatusAsync(CancellationToken ct) =>
+        _inner.GetConnectionStatusAsync(ct);
+
+    /// <summary>Disposes the underlying package client (and its channel). Prefer <see cref="DisposeAsync"/>.</summary>
+    public void Dispose() => _inner.Dispose();
+
+    /// <summary>Asynchronously disposes the underlying package client (and its channel).</summary>
+    /// <returns>A task that completes when the client has been disposed.</returns>
+    public ValueTask DisposeAsync() => _inner.DisposeAsync();
+}