Joseph Doherty
a04196d725
The driver-status and resilience stores had no removal path, so a deleted driver stayed on /hosts as a frozen row until restart. Both stores gain RetainOnly, which /hosts calls after each successful config load (never on a failed one). The stores and InProcessBroadcaster now raise their events one subscriber at a time, so a throwing handler cannot starve later subscribers or escape onto the bridge actor. The alert/script-log bridges publish inside their try, and the resilience bridge catches a store failure instead of restarting. Refs #745 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
47 lines
2.4 KiB
C#
47 lines
2.4 KiB
C#
using Akka.Actor;
|
|
using Akka.Cluster.Tools.PublishSubscribe;
|
|
using Akka.Event;
|
|
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Drivers;
|
|
|
|
namespace ZB.MOM.WW.OtOpcUa.AdminUI.Hubs;
|
|
|
|
/// <summary>
|
|
/// Akka actor that subscribes to the <c>driver-resilience-status</c> DistributedPubSub topic and
|
|
/// upserts every <see cref="DriverResilienceStatusChanged"/> snapshot into the in-process
|
|
/// <see cref="IDriverResilienceStatusStore"/>. The driver status panel's resilience section reads
|
|
/// that store directly (per the house ban on server-side self-targeted SignalR connections), so —
|
|
/// unlike <see cref="DriverStatusSignalRBridge"/> — there is no SignalR hub here; resilience has no
|
|
/// browser-JS consumer. Spawned per admin-role node by <c>WithOtOpcUaSignalRBridges</c>.
|
|
/// </summary>
|
|
public sealed class DriverResilienceStatusBridge : ReceiveActor
|
|
{
|
|
public const string TopicName = DriverResilienceStatusChanged.TopicName;
|
|
|
|
private readonly IDriverResilienceStatusStore _store;
|
|
private readonly ILoggingAdapter _log = Context.GetLogger();
|
|
|
|
/// <summary>Creates actor props for a <see cref="DriverResilienceStatusBridge"/>.</summary>
|
|
/// <param name="store">Resilience status store updated on every forwarded snapshot.</param>
|
|
/// <returns>An Akka.NET <see cref="Props"/> instance for spawning the bridge actor.</returns>
|
|
public static Props Props(IDriverResilienceStatusStore store) =>
|
|
Akka.Actor.Props.Create(() => new DriverResilienceStatusBridge(store));
|
|
|
|
/// <summary>Initializes a new instance of <see cref="DriverResilienceStatusBridge"/>.</summary>
|
|
/// <param name="store">Resilience status store updated on every forwarded snapshot.</param>
|
|
public DriverResilienceStatusBridge(IDriverResilienceStatusStore store)
|
|
{
|
|
_store = store;
|
|
Receive<DriverResilienceStatusChanged>(msg =>
|
|
{
|
|
// A store failure must not restart the bridge (a restart re-subscribes DPS and drops messages).
|
|
try { _store.Upsert(msg); }
|
|
catch (Exception ex) { _log.Warning(ex, "DriverResilienceStatusBridge: upsert failed for {0}", msg.DriverInstanceId); }
|
|
});
|
|
Receive<SubscribeAck>(_ => { /* DPS confirmation */ });
|
|
}
|
|
|
|
/// <inheritdoc />
|
|
protected override void PreStart() =>
|
|
DistributedPubSub.Get(Context.System).Mediator.Tell(new Subscribe(TopicName, Self));
|
|
}
|