feat(sitestream): add SubscribeSiteAlarms site-wide alarm-only broadcast (plan #10 T1)
Adds SiteStreamManager.SubscribeSiteAlarms(IActorRef) — same broadcast-hub wiring as Subscribe(...) but with no per-instance filter and forwarding only AlarmStateChanged events (AttributeValueChanged dropped). Returns a subscription id torn down via the existing Unsubscribe, symmetric with the per-instance subscribe. Backs the Task 2 SubscribeSite gRPC server handler. Claude-Session: https://claude.ai/code/session_01MtdgwpEeCUn6cUA5f1LMPj
This commit is contained in:
@@ -20,6 +20,9 @@ namespace ZB.MOM.WW.ScadaBridge.SiteRuntime.Streaming;
|
|||||||
/// </summary>
|
/// </summary>
|
||||||
public class SiteStreamManager : ISiteStreamSubscriber
|
public class SiteStreamManager : ISiteStreamSubscriber
|
||||||
{
|
{
|
||||||
|
/// <summary>Sentinel instance name recorded for site-wide (non-instance-scoped) subscriptions.</summary>
|
||||||
|
private const string SiteWideInstanceName = "*";
|
||||||
|
|
||||||
private ActorSystem? _system;
|
private ActorSystem? _system;
|
||||||
private IMaterializer? _materializer;
|
private IMaterializer? _materializer;
|
||||||
private readonly int _bufferSize;
|
private readonly int _bufferSize;
|
||||||
@@ -117,6 +120,44 @@ public class SiteStreamManager : ISiteStreamSubscriber
|
|||||||
return subscriptionId;
|
return subscriptionId;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Subscribe to ALARM events for ALL instances on the site (no per-instance
|
||||||
|
/// filter). Only <see cref="AlarmStateChanged"/> events are forwarded;
|
||||||
|
/// <see cref="AttributeValueChanged"/> events are dropped (attributes are far
|
||||||
|
/// higher-volume and the aggregated Alarm Summary never shows them). Same
|
||||||
|
/// broadcast-hub wiring as <see cref="Subscribe"/>, and the returned
|
||||||
|
/// subscription id is torn down via <see cref="Unsubscribe"/> exactly like the
|
||||||
|
/// per-instance variant.
|
||||||
|
/// </summary>
|
||||||
|
/// <param name="subscriber">The actor that receives forwarded <see cref="AlarmStateChanged"/> events.</param>
|
||||||
|
/// <returns>A subscription id to pass to <see cref="Unsubscribe"/>.</returns>
|
||||||
|
public string SubscribeSiteAlarms(IActorRef subscriber)
|
||||||
|
{
|
||||||
|
if (_hubSource is null || _materializer is null)
|
||||||
|
throw new InvalidOperationException("SiteStreamManager.Initialize must be called before SubscribeSiteAlarms");
|
||||||
|
|
||||||
|
var subscriptionId = Guid.NewGuid().ToString();
|
||||||
|
var capturedSubscriber = subscriber;
|
||||||
|
|
||||||
|
var killSwitch = _hubSource
|
||||||
|
.Where(ev => ev is AlarmStateChanged)
|
||||||
|
.Buffer(_bufferSize, OverflowStrategy.DropHead)
|
||||||
|
.ViaMaterialized(KillSwitches.Single<ISiteStreamEvent>(), Keep.Right)
|
||||||
|
.To(Sink.ForEach<ISiteStreamEvent>(ev => capturedSubscriber.Tell(ev)))
|
||||||
|
.Run(_materializer);
|
||||||
|
|
||||||
|
lock (_lock)
|
||||||
|
{
|
||||||
|
_subscriptions[subscriptionId] = new SubscriptionInfo(
|
||||||
|
SiteWideInstanceName, subscriber, killSwitch, DateTimeOffset.UtcNow);
|
||||||
|
}
|
||||||
|
|
||||||
|
_logger.LogDebug(
|
||||||
|
"Subscriber {SubscriptionId} registered for site-wide alarm events", subscriptionId);
|
||||||
|
|
||||||
|
return subscriptionId;
|
||||||
|
}
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Unsubscribe from instance events. Shuts down the per-subscriber
|
/// Unsubscribe from instance events. Shuts down the per-subscriber
|
||||||
/// stream graph via its KillSwitch.
|
/// stream graph via its KillSwitch.
|
||||||
|
|||||||
@@ -103,6 +103,50 @@ public class SiteStreamManagerTests : TestKit, IDisposable
|
|||||||
probe2.ExpectNoMsg(TimeSpan.FromMilliseconds(500));
|
probe2.ExpectNoMsg(TimeSpan.FromMilliseconds(500));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public void SubscribeSiteAlarms_ForwardsAlarmsForAllInstances_ButNotAttributes()
|
||||||
|
{
|
||||||
|
var probe = CreateTestProbe();
|
||||||
|
_streamManager.SubscribeSiteAlarms(probe.Ref);
|
||||||
|
|
||||||
|
// Alarm events from two different instances — both should arrive.
|
||||||
|
_streamManager.PublishAlarmStateChanged(new AlarmStateChanged(
|
||||||
|
"Pump1", "HighTemp", AlarmState.Active, 1, DateTimeOffset.UtcNow));
|
||||||
|
_streamManager.PublishAlarmStateChanged(new AlarmStateChanged(
|
||||||
|
"Pump2", "LowFlow", AlarmState.Active, 2, DateTimeOffset.UtcNow));
|
||||||
|
|
||||||
|
// Attribute events must be filtered out entirely.
|
||||||
|
_streamManager.PublishAttributeValueChanged(new AttributeValueChanged(
|
||||||
|
"Pump1", "Temperature", "Temperature", "100", "Good", DateTimeOffset.UtcNow));
|
||||||
|
_streamManager.PublishAttributeValueChanged(new AttributeValueChanged(
|
||||||
|
"Pump3", "Flow", "Flow", "42", "Good", DateTimeOffset.UtcNow));
|
||||||
|
|
||||||
|
// Collect the two alarms (order across instances is not guaranteed).
|
||||||
|
var received = new[]
|
||||||
|
{
|
||||||
|
probe.ExpectMsg<AlarmStateChanged>(TimeSpan.FromSeconds(3)),
|
||||||
|
probe.ExpectMsg<AlarmStateChanged>(TimeSpan.FromSeconds(3)),
|
||||||
|
};
|
||||||
|
|
||||||
|
var instances = received.Select(a => a.InstanceUniqueName).OrderBy(n => n).ToArray();
|
||||||
|
Assert.Equal(new[] { "Pump1", "Pump2" }, instances);
|
||||||
|
|
||||||
|
// No attribute events (nor any further alarm) should be delivered.
|
||||||
|
probe.ExpectNoMsg(TimeSpan.FromMilliseconds(500));
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public void SubscribeSiteAlarms_SubscriptionRemovableViaUnsubscribe()
|
||||||
|
{
|
||||||
|
var probe = CreateTestProbe();
|
||||||
|
var id = _streamManager.SubscribeSiteAlarms(probe.Ref);
|
||||||
|
|
||||||
|
Assert.NotNull(id);
|
||||||
|
Assert.Equal(1, _streamManager.SubscriptionCount);
|
||||||
|
Assert.True(_streamManager.Unsubscribe(id));
|
||||||
|
Assert.Equal(0, _streamManager.SubscriptionCount);
|
||||||
|
}
|
||||||
|
|
||||||
[Fact]
|
[Fact]
|
||||||
public void RemoveSubscriber_RemovesAllSubscriptionsForActor()
|
public void RemoveSubscriber_RemovesAllSubscriptionsForActor()
|
||||||
{
|
{
|
||||||
|
|||||||
Reference in New Issue
Block a user