test(galaxy): live gate for the stream-attached ack against a real gateway (#877)
v2-ci / unit-tests (push) Failing after 10m42s
v2-ci / unit-tests (push) Failing after 10m42s
Adds GatewayGalaxyStreamAttachedAckLiveTests (skip-gated on MXGW_ENDPOINT + GALAXY_MXGW_API_KEY, like the other gateway live tests). Proves what the unit suite cannot: the request flag reaches a real gateway and the sentinel comes back. - live session advertises stream-events-attached-ack; GatewayGalaxySubscriber reports it - raw stream: sentinel is the first message when requested (family Unspecified, worker_sequence 0, live_after 0) and never sent when not requested (control, with a data event as the positive control) - real GalaxyDriver + EventPump + GatewayGalaxySubscriber behind a pass-through recorder: exactly one sentinel, ack observed before the first SubscribeBulk, pump StreamAttached completed, sentinel never dispatched nor counted as received (galaxy.events.stream_attached=1, received == non-sentinel events), no missing-ack warning within the production 1 s budget, and the static attribute TestMachine_002.Tagname delivers its initial value - control: a subscriber that opts out gets no sentinel and the pump never reports one, while the #865 fallback still delivers Run 2026-11-12 against windev 10.100.0.48:5120 (mxaccessgw 447070a5): 4/4. Mutation checks, each caught: subscriber not requesting the ack (sentinel count 0); driver skipping the ack wait (wire order became stream | subscribe | ack, 2/2 runs, so the race is real on a live gateway); pump not intercepting the sentinel (received=2 vs 1 non-sentinel). Claude-Session: https://claude.ai/code/session_01GT67ACmpMR7b9LGRPLijMi
This commit is contained in:
Joseph Doherty
+382
@@ -0,0 +1,382 @@
|
|||||||
|
using System.Diagnostics.Metrics;
|
||||||
|
using System.Runtime.CompilerServices;
|
||||||
|
using Microsoft.Extensions.Logging;
|
||||||
|
using Shouldly;
|
||||||
|
using Xunit;
|
||||||
|
using ZB.MOM.WW.MxGateway.Client;
|
||||||
|
using ZB.MOM.WW.MxGateway.Contracts;
|
||||||
|
using ZB.MOM.WW.MxGateway.Contracts.Proto;
|
||||||
|
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||||
|
using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Config;
|
||||||
|
using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Runtime;
|
||||||
|
|
||||||
|
namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests.Runtime;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// #877 live gate — the stream-attached acknowledgement (mxaccessgw#227) against a REAL gateway. The unit
|
||||||
|
/// suite (<see cref="StreamAttachedAckTests"/>) proves the driver logic over a fake; what it cannot prove is
|
||||||
|
/// that the request flag reaches a real gateway and the sentinel comes back. These do:
|
||||||
|
/// <list type="number">
|
||||||
|
/// <item><description>the live session advertises <c>stream-events-attached-ack</c> and the production
|
||||||
|
/// <see cref="GatewayGalaxySubscriber"/> reports it;</description></item>
|
||||||
|
/// <item><description>the gateway sends the sentinel first when asked, and NEVER when not asked — the
|
||||||
|
/// control that makes the driver-level assertions falsifiable;</description></item>
|
||||||
|
/// <item><description>through the real <see cref="GalaxyDriver"/> + <see cref="EventPump"/>: the
|
||||||
|
/// subscriber's stream carries exactly one sentinel (so it requested it), the pump's attached signal
|
||||||
|
/// completes BEFORE the first SubscribeBulk is issued, the sentinel is never dispatched or counted as
|
||||||
|
/// a received event, and a static attribute's initial value then arrives;</description></item>
|
||||||
|
/// <item><description>control: a subscriber that does not request the ack gets no sentinel, the pump
|
||||||
|
/// never reports one, and the driver still delivers through the old best-effort wait.</description></item>
|
||||||
|
/// </list>
|
||||||
|
/// <para>
|
||||||
|
/// Skip-gated like the other gateway live tests: runs only when <c>MXGW_ENDPOINT</c> +
|
||||||
|
/// <c>GALAXY_MXGW_API_KEY</c> are set. Needs a gateway at mxaccessgw >= <c>afd504d5</c>. Captured
|
||||||
|
/// 2026-11-12 against windev <c>10.100.0.48:5120</c> (mxaccessgw <c>447070a5</c>). Read-only: subscribes
|
||||||
|
/// to <c>TestMachine_002.Tagname</c>, a static string attribute, so the value can only come from the
|
||||||
|
/// advise's initial snapshot (no concurrent writer can supply it, unlike <c>TestFloat</c>).
|
||||||
|
/// </para>
|
||||||
|
/// </summary>
|
||||||
|
[Trait("Category", "Integration")]
|
||||||
|
public sealed class GatewayGalaxyStreamAttachedAckLiveTests
|
||||||
|
{
|
||||||
|
/// <summary>A static, read-only attribute: its only event after an advise is the initial value.</summary>
|
||||||
|
private const string StaticRef = "TestMachine_002.Tagname";
|
||||||
|
|
||||||
|
private const string ExpectedStaticValue = "TestMachine_002";
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task Live_session_advertises_the_attach_ack_capability()
|
||||||
|
{
|
||||||
|
var (endpoint, apiKey) = RequireLiveGatewayOrSkip();
|
||||||
|
var ct = TestContext.Current.CancellationToken;
|
||||||
|
|
||||||
|
await using var session = new GalaxyMxSession(new GalaxyMxAccessOptions(ClientName: "OtOpcUa877Caps"));
|
||||||
|
await session.ConnectAsync(BuildClientOptions(endpoint, apiKey), ct);
|
||||||
|
|
||||||
|
var live = session.Session.ShouldNotBeNull();
|
||||||
|
live.OpenSessionReply.Capabilities.ShouldContain(GatewayCapabilities.StreamEventsAttachedAck,
|
||||||
|
$"the gateway must advertise the capability; got [{string.Join(", ", live.OpenSessionReply.Capabilities)}]");
|
||||||
|
live.SupportsStreamAttachedAck.ShouldBeTrue();
|
||||||
|
new GatewayGalaxySubscriber(session).SupportsStreamAttachedAck.ShouldBeTrue(
|
||||||
|
"the production subscriber reads the capability from the live session");
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Raw-stream contract, both ways on one gateway. Requested: the first message is the sentinel
|
||||||
|
/// (no family, no body, worker_sequence 0, live_after 0 on a fresh stream). Not requested: no sentinel
|
||||||
|
/// at all — and a data event (the advise's initial value) is the positive control that the stream was
|
||||||
|
/// live long enough for a sentinel to have shown up had the gateway sent one.
|
||||||
|
/// </summary>
|
||||||
|
[Fact]
|
||||||
|
public async Task Live_gateway_sends_the_sentinel_first_when_requested_and_never_otherwise()
|
||||||
|
{
|
||||||
|
var (endpoint, apiKey) = RequireLiveGatewayOrSkip();
|
||||||
|
var ct = TestContext.Current.CancellationToken;
|
||||||
|
var clientOptions = BuildClientOptions(endpoint, apiKey);
|
||||||
|
|
||||||
|
// Requested.
|
||||||
|
await using (var session = new GalaxyMxSession(new GalaxyMxAccessOptions(ClientName: "OtOpcUa877RawOn")))
|
||||||
|
{
|
||||||
|
await session.ConnectAsync(clientOptions, ct);
|
||||||
|
using var cts = CancellationTokenSource.CreateLinkedTokenSource(ct);
|
||||||
|
cts.CancelAfter(TimeSpan.FromSeconds(20));
|
||||||
|
await using var e = session.Session!.StreamEventsAsync(0, emitStreamAttached: true, cts.Token)
|
||||||
|
.GetAsyncEnumerator(cts.Token);
|
||||||
|
(await e.MoveNextAsync()).ShouldBeTrue("a requested stream must yield the sentinel first");
|
||||||
|
var first = e.Current;
|
||||||
|
first.StreamAttached.ShouldNotBeNull($"the first message must be the attach sentinel; got {first}");
|
||||||
|
first.Family.ShouldBe(MxEventFamily.Unspecified);
|
||||||
|
first.WorkerSequence.ShouldBe(0ul, "the sentinel must never look like a resume cursor");
|
||||||
|
first.StreamAttached.LiveAfterSequence.ShouldBe(0ul, "a fresh stream filters live events at 0");
|
||||||
|
await cts.CancelAsync();
|
||||||
|
}
|
||||||
|
|
||||||
|
// Not requested (control).
|
||||||
|
await using (var session = new GalaxyMxSession(new GalaxyMxAccessOptions(ClientName: "OtOpcUa877RawOff")))
|
||||||
|
{
|
||||||
|
await session.ConnectAsync(clientOptions, ct);
|
||||||
|
var seen = new List<MxEvent>();
|
||||||
|
var firstData = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||||
|
using var cts = CancellationTokenSource.CreateLinkedTokenSource(ct);
|
||||||
|
var pump = Task.Run(async () =>
|
||||||
|
{
|
||||||
|
try
|
||||||
|
{
|
||||||
|
await foreach (var ev in session.Session!.StreamEventsAsync(0, emitStreamAttached: false, cts.Token))
|
||||||
|
{
|
||||||
|
lock (seen) seen.Add(ev);
|
||||||
|
if (ev.Family == MxEventFamily.OnDataChange) firstData.TrySetResult();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
catch (Exception) when (cts.IsCancellationRequested) { }
|
||||||
|
}, ct);
|
||||||
|
|
||||||
|
// No ack to wait on here (that is the point): give the attach ample time before advising.
|
||||||
|
await Task.Delay(TimeSpan.FromSeconds(2), ct);
|
||||||
|
await new GatewayGalaxySubscriber(session).SubscribeBulkAsync([StaticRef], 0, ct);
|
||||||
|
await TestWait.ForAsync(firstData.Task, "positive control: the unrequested stream still carries data",
|
||||||
|
() => Describe(seen));
|
||||||
|
await cts.CancelAsync();
|
||||||
|
await pump;
|
||||||
|
|
||||||
|
lock (seen)
|
||||||
|
{
|
||||||
|
seen.ShouldNotContain(ev => ev.StreamAttached != null,
|
||||||
|
$"the gateway must not send the sentinel unless asked; {Describe(seen)}");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// The production path end to end: real subscriber, real driver, real pump. Ordering is recorded by a
|
||||||
|
/// pass-through decorator, so "ack before advise" is observed, not inferred. The production 1 s
|
||||||
|
/// <see cref="GalaxyDriver.StreamRequestWait"/> budget is kept, and the driver must not log the
|
||||||
|
/// missing-ack warning — so this also fails if the ack arrived only after the budget ran out.
|
||||||
|
/// </summary>
|
||||||
|
[Fact]
|
||||||
|
public async Task Live_driver_waits_for_the_ack_before_advising_and_receives_the_initial_value()
|
||||||
|
{
|
||||||
|
var (endpoint, apiKey) = RequireLiveGatewayOrSkip();
|
||||||
|
var ct = TestContext.Current.CancellationToken;
|
||||||
|
const string clientName = "OtOpcUa877Driver";
|
||||||
|
|
||||||
|
using var meters = new PumpMeters(clientName);
|
||||||
|
await using var session = new GalaxyMxSession(new GalaxyMxAccessOptions(ClientName: clientName));
|
||||||
|
await session.ConnectAsync(BuildClientOptions(endpoint, apiKey), ct);
|
||||||
|
var recorder = new RecordingSubscriber(new GatewayGalaxySubscriber(session));
|
||||||
|
var log = new CapturingLogger();
|
||||||
|
|
||||||
|
await using var driver = new GalaxyDriver("live-877", Opts(endpoint, clientName), hierarchySource: null,
|
||||||
|
subscriber: recorder, logger: new TypedLogger(log));
|
||||||
|
driver.StreamRequestWait.ShouldBe(TimeSpan.FromSeconds(1), "exercise the production budget, not a test one");
|
||||||
|
await driver.InitializeAsync("{}", ct);
|
||||||
|
|
||||||
|
var dispatched = new List<DataChangeEventArgs>();
|
||||||
|
var initial = new TaskCompletionSource<DataChangeEventArgs>(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||||
|
driver.OnDataChange += (_, e) =>
|
||||||
|
{
|
||||||
|
lock (dispatched) dispatched.Add(e);
|
||||||
|
if (string.Equals(e.FullReference, StaticRef, StringComparison.OrdinalIgnoreCase)) initial.TrySetResult(e);
|
||||||
|
};
|
||||||
|
|
||||||
|
await driver.SubscribeAsync([StaticRef], TimeSpan.Zero, ct);
|
||||||
|
var value = await TestWait.ForAsync(initial.Task,
|
||||||
|
"the advise's initial value must arrive (the static attribute has no other event)", recorder.Describe);
|
||||||
|
|
||||||
|
// 1. The production subscriber requested the ack: the gateway only sends it when asked (raw control).
|
||||||
|
recorder.SentinelCount.ShouldBe(1, $"exactly one sentinel on the stream; {recorder.Describe()}");
|
||||||
|
|
||||||
|
// 2. Ack before the first advise — observed on the wire order the decorator saw.
|
||||||
|
var calls = recorder.Snapshot();
|
||||||
|
var ackAt = Array.FindIndex(calls, c => c.StartsWith("ack", StringComparison.Ordinal));
|
||||||
|
var subscribeAt = Array.FindIndex(calls, c => c.StartsWith("subscribe:", StringComparison.Ordinal));
|
||||||
|
calls[0].ShouldBe("stream", recorder.Describe());
|
||||||
|
ackAt.ShouldBeGreaterThan(0, recorder.Describe());
|
||||||
|
subscribeAt.ShouldBeGreaterThan(ackAt, $"the advise must follow the attach ack; {recorder.Describe()}");
|
||||||
|
|
||||||
|
// 3. The pump consumed the ack as a control signal.
|
||||||
|
var pump = (EventPump)driver.EventPumpInstanceForTests!;
|
||||||
|
pump.AttachAckExpected.ShouldBeTrue();
|
||||||
|
pump.StreamAttachedReceived.ShouldBeTrue();
|
||||||
|
pump.StreamAttached.IsCompletedSuccessfully.ShouldBeTrue();
|
||||||
|
|
||||||
|
// 4. Never dispatched, never counted as received.
|
||||||
|
lock (dispatched)
|
||||||
|
{
|
||||||
|
dispatched.ShouldAllBe(e => string.Equals(e.FullReference, StaticRef, StringComparison.OrdinalIgnoreCase),
|
||||||
|
"only the subscribed attribute may be dispatched");
|
||||||
|
dispatched.Count.ShouldBeLessThanOrEqualTo(recorder.DataEventCount,
|
||||||
|
"every dispatch must come from a real data event, never the sentinel");
|
||||||
|
}
|
||||||
|
await TestWait.UntilAsync(() => meters.StreamAttached == 1 && meters.Received == recorder.NonSentinelCount,
|
||||||
|
"the pump counts the ack once on its own counter and never as a received event",
|
||||||
|
() => $"stream_attached={meters.StreamAttached} received={meters.Received} nonSentinel={recorder.NonSentinelCount}");
|
||||||
|
|
||||||
|
// 5. The value is the galaxy's, and the budget was not exhausted.
|
||||||
|
value.Snapshot.Value.ShouldBe(ExpectedStaticValue);
|
||||||
|
lock (log.Entries)
|
||||||
|
{
|
||||||
|
log.Entries.ShouldNotContain(e => e.Text.Contains("attach acknowledgement", StringComparison.Ordinal),
|
||||||
|
$"the ack must arrive within the production budget; log=\n{log.AllText}");
|
||||||
|
}
|
||||||
|
|
||||||
|
TestContext.Current.SendDiagnosticMessage($"#877 live: {recorder.Describe()} value={value.Snapshot.Value}");
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Control for the driver test: the same driver over a subscriber that does NOT request the ack (as
|
||||||
|
/// against an older gateway). The ack-specific assertions above must fail here — no sentinel, no
|
||||||
|
/// attached signal — while the driver still works through the #865 best-effort wait.
|
||||||
|
/// </summary>
|
||||||
|
[Fact]
|
||||||
|
public async Task Live_control_without_the_request_no_ack_arrives_and_the_fallback_still_delivers()
|
||||||
|
{
|
||||||
|
var (endpoint, apiKey) = RequireLiveGatewayOrSkip();
|
||||||
|
var ct = TestContext.Current.CancellationToken;
|
||||||
|
const string clientName = "OtOpcUa877NoAck";
|
||||||
|
|
||||||
|
using var meters = new PumpMeters(clientName);
|
||||||
|
await using var session = new GalaxyMxSession(new GalaxyMxAccessOptions(ClientName: clientName));
|
||||||
|
await session.ConnectAsync(BuildClientOptions(endpoint, apiKey), ct);
|
||||||
|
session.Session!.SupportsStreamAttachedAck.ShouldBeTrue("the gateway does support it; this client opts out");
|
||||||
|
var recorder = new RecordingSubscriber(new GatewayGalaxySubscriber(session), optOutSession: session);
|
||||||
|
|
||||||
|
await using var driver = new GalaxyDriver("live-877-noack", Opts(endpoint, clientName), hierarchySource: null,
|
||||||
|
subscriber: recorder);
|
||||||
|
await driver.InitializeAsync("{}", ct);
|
||||||
|
var initial = new TaskCompletionSource<DataChangeEventArgs>(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||||
|
driver.OnDataChange += (_, e) => initial.TrySetResult(e);
|
||||||
|
|
||||||
|
await driver.SubscribeAsync([StaticRef], TimeSpan.Zero, ct);
|
||||||
|
// Positive control first: data flowed, so the stream was live long enough to have carried a sentinel.
|
||||||
|
(await TestWait.ForAsync(initial.Task, "the fallback path still delivers the initial value", recorder.Describe))
|
||||||
|
.Snapshot.Value.ShouldBe(ExpectedStaticValue);
|
||||||
|
|
||||||
|
recorder.SentinelCount.ShouldBe(0, $"no ack without the request; {recorder.Describe()}");
|
||||||
|
var pump = (EventPump)driver.EventPumpInstanceForTests!;
|
||||||
|
pump.AttachAckExpected.ShouldBeFalse();
|
||||||
|
pump.StreamAttachedReceived.ShouldBeFalse();
|
||||||
|
pump.StreamAttached.IsCompleted.ShouldBeFalse("no ack was requested and the stream is alive");
|
||||||
|
meters.StreamAttached.ShouldBe(0);
|
||||||
|
}
|
||||||
|
|
||||||
|
private static GalaxyDriverOptions Opts(string endpoint, string clientName) => new(
|
||||||
|
new GalaxyGatewayOptions(endpoint, "unused-seam-path", UseTls: false),
|
||||||
|
new GalaxyMxAccessOptions(clientName),
|
||||||
|
new GalaxyRepositoryOptions(WatchDeployEvents: false),
|
||||||
|
new GalaxyReconnectOptions());
|
||||||
|
|
||||||
|
private static (string Endpoint, string ApiKey) RequireLiveGatewayOrSkip()
|
||||||
|
{
|
||||||
|
var endpoint = Environment.GetEnvironmentVariable("MXGW_ENDPOINT");
|
||||||
|
var apiKey = Environment.GetEnvironmentVariable("GALAXY_MXGW_API_KEY");
|
||||||
|
if (string.IsNullOrWhiteSpace(endpoint) || string.IsNullOrWhiteSpace(apiKey))
|
||||||
|
Assert.Skip("Set MXGW_ENDPOINT + GALAXY_MXGW_API_KEY to run the live stream-attached-ack gate.");
|
||||||
|
return (endpoint!, apiKey!);
|
||||||
|
}
|
||||||
|
|
||||||
|
private static MxGatewayClientOptions BuildClientOptions(string endpoint, string apiKey) => new()
|
||||||
|
{
|
||||||
|
Endpoint = new Uri(endpoint, UriKind.Absolute),
|
||||||
|
ApiKey = apiKey,
|
||||||
|
UseTls = false,
|
||||||
|
ConnectTimeout = TimeSpan.FromSeconds(10),
|
||||||
|
DefaultCallTimeout = TimeSpan.FromSeconds(30),
|
||||||
|
StreamTimeout = TimeSpan.FromSeconds(30),
|
||||||
|
};
|
||||||
|
|
||||||
|
private static string Describe(List<MxEvent> seen)
|
||||||
|
{
|
||||||
|
lock (seen) return $"events=[{string.Join(" | ", seen.Select(e => e.StreamAttached != null ? "ack" : e.Family.ToString()))}]";
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Pass-through <see cref="IGalaxySubscriber"/> that records the order of stream request, sentinel
|
||||||
|
/// arrival and SubscribeBulk. With <c>optOutSession</c> set it opens the stream WITHOUT requesting the
|
||||||
|
/// ack and reports no capability — an older-gateway client against the same live gateway.
|
||||||
|
/// </summary>
|
||||||
|
private sealed class RecordingSubscriber(GatewayGalaxySubscriber inner, GalaxyMxSession? optOutSession = null)
|
||||||
|
: IGalaxySubscriber
|
||||||
|
{
|
||||||
|
private readonly Lock _gate = new();
|
||||||
|
private readonly List<string> _calls = [];
|
||||||
|
private int _sentinels;
|
||||||
|
private int _nonSentinels;
|
||||||
|
private int _dataEvents;
|
||||||
|
|
||||||
|
public int SentinelCount => Volatile.Read(ref _sentinels);
|
||||||
|
|
||||||
|
public int NonSentinelCount => Volatile.Read(ref _nonSentinels);
|
||||||
|
|
||||||
|
public int DataEventCount => Volatile.Read(ref _dataEvents);
|
||||||
|
|
||||||
|
public bool SupportsStreamAttachedAck => optOutSession is null && inner.SupportsStreamAttachedAck;
|
||||||
|
|
||||||
|
public string[] Snapshot() { lock (_gate) return [.. _calls]; }
|
||||||
|
|
||||||
|
public string Describe() => $"calls=[{string.Join(" | ", Snapshot())}] sentinels={SentinelCount} nonSentinel={NonSentinelCount}";
|
||||||
|
|
||||||
|
private void Record(string call) { lock (_gate) _calls.Add(call); }
|
||||||
|
|
||||||
|
public async Task<IReadOnlyList<SubscribeResult>> SubscribeBulkAsync(
|
||||||
|
IReadOnlyList<string> fullReferences, int bufferedUpdateIntervalMs, CancellationToken cancellationToken)
|
||||||
|
{
|
||||||
|
Record("subscribe:" + string.Join(",", fullReferences));
|
||||||
|
return await inner.SubscribeBulkAsync(fullReferences, bufferedUpdateIntervalMs, cancellationToken)
|
||||||
|
.ConfigureAwait(false);
|
||||||
|
}
|
||||||
|
|
||||||
|
public Task UnsubscribeBulkAsync(IReadOnlyList<int> itemHandles, CancellationToken cancellationToken)
|
||||||
|
=> inner.UnsubscribeBulkAsync(itemHandles, cancellationToken);
|
||||||
|
|
||||||
|
public async IAsyncEnumerable<MxEvent> StreamEventsAsync([EnumeratorCancellation] CancellationToken cancellationToken)
|
||||||
|
{
|
||||||
|
Record("stream");
|
||||||
|
var source = optOutSession is null
|
||||||
|
? inner.StreamEventsAsync(cancellationToken)
|
||||||
|
: optOutSession.Session!.StreamEventsAsync(0, emitStreamAttached: false, cancellationToken);
|
||||||
|
await foreach (var ev in source.WithCancellation(cancellationToken).ConfigureAwait(false))
|
||||||
|
{
|
||||||
|
if (ev.StreamAttached is not null)
|
||||||
|
{
|
||||||
|
Interlocked.Increment(ref _sentinels);
|
||||||
|
Record($"ack(seq={ev.WorkerSequence},liveAfter={ev.StreamAttached.LiveAfterSequence})");
|
||||||
|
}
|
||||||
|
else
|
||||||
|
{
|
||||||
|
Interlocked.Increment(ref _nonSentinels);
|
||||||
|
if (ev.Family == MxEventFamily.OnDataChange) Interlocked.Increment(ref _dataEvents);
|
||||||
|
Record("event:" + ev.Family);
|
||||||
|
}
|
||||||
|
yield return ev;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>Reads the pump's received / stream_attached counters for one client name only.</summary>
|
||||||
|
private sealed class PumpMeters : IDisposable
|
||||||
|
{
|
||||||
|
private readonly MeterListener _listener = new();
|
||||||
|
private long _received;
|
||||||
|
private long _attached;
|
||||||
|
|
||||||
|
public PumpMeters(string clientName)
|
||||||
|
{
|
||||||
|
_listener.InstrumentPublished = (instrument, listener) =>
|
||||||
|
{
|
||||||
|
if (instrument.Meter.Name == EventPump.MeterName
|
||||||
|
&& instrument.Name is "galaxy.events.received" or "galaxy.events.stream_attached")
|
||||||
|
listener.EnableMeasurementEvents(instrument);
|
||||||
|
};
|
||||||
|
_listener.SetMeasurementEventCallback<long>((instrument, value, tags, _) =>
|
||||||
|
{
|
||||||
|
foreach (var tag in tags)
|
||||||
|
{
|
||||||
|
if (tag.Key != "galaxy.client" || !Equals(tag.Value, clientName)) continue;
|
||||||
|
if (instrument.Name == "galaxy.events.received") Interlocked.Add(ref _received, value);
|
||||||
|
else Interlocked.Add(ref _attached, value);
|
||||||
|
}
|
||||||
|
});
|
||||||
|
_listener.Start();
|
||||||
|
}
|
||||||
|
|
||||||
|
public long Received => Interlocked.Read(ref _received);
|
||||||
|
|
||||||
|
public long StreamAttached => Interlocked.Read(ref _attached);
|
||||||
|
|
||||||
|
public void Dispose() => _listener.Dispose();
|
||||||
|
}
|
||||||
|
|
||||||
|
private sealed class TypedLogger(CapturingLogger inner) : ILogger<GalaxyDriver>
|
||||||
|
{
|
||||||
|
public IDisposable? BeginScope<TState>(TState state) where TState : notnull => null;
|
||||||
|
|
||||||
|
public bool IsEnabled(LogLevel logLevel) => true;
|
||||||
|
|
||||||
|
public void Log<TState>(LogLevel logLevel, EventId eventId, TState state, Exception? exception,
|
||||||
|
Func<TState, Exception?, string> formatter)
|
||||||
|
{
|
||||||
|
lock (inner.Entries) inner.Log(logLevel, eventId, state, exception, formatter);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user