lmxopcua/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Hubs/DriverStatusSignalRBridge.cs

using Akka.Actor;
using Akka.Cluster.Tools.PublishSubscribe;
using Akka.Event;
using Microsoft.AspNetCore.SignalR;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Drivers;

namespace ZB.MOM.WW.OtOpcUa.AdminUI.Hubs;

/// <summary>
///     Akka actor that subscribes to the <c>driver-health</c> DistributedPubSub topic and
///     forwards every <see cref="DriverHealthChanged"/> snapshot to (a) the in-memory snapshot
///     store and (b) all SignalR clients connected to <see cref="DriverStatusHub"/> grouped
///     by <see cref="DriverHealthChanged.DriverInstanceId"/>. Spawned on admin-role nodes by
///     <c>AddOtOpcUaSignalRBridges</c>.
/// </summary>
public sealed class DriverStatusSignalRBridge : ReceiveActor
{
    public const string TopicName = DriverHealthChanged.TopicName;

    private readonly IHubContext<DriverStatusHub> _hub;
    private readonly IDriverStatusSnapshotStore _store;
    private readonly ILoggingAdapter _log = Context.GetLogger();

    /// <summary>Creates actor props for a <see cref="DriverStatusSignalRBridge"/>.</summary>
    /// <param name="hub">The SignalR hub context for pushing snapshots to grouped clients.</param>
    /// <param name="store">Snapshot store updated before each SignalR push.</param>
    /// <returns>An Akka.NET <see cref="Props"/> instance for spawning the bridge actor.</returns>
    public static Props Props(IHubContext<DriverStatusHub> hub, IDriverStatusSnapshotStore store) =>
        Akka.Actor.Props.Create(() => new DriverStatusSignalRBridge(hub, store));

    /// <summary>Initializes a new instance of <see cref="DriverStatusSignalRBridge"/>.</summary>
    /// <param name="hub">The SignalR hub context for pushing snapshots to grouped clients.</param>
    /// <param name="store">Snapshot store updated before each SignalR push.</param>
    public DriverStatusSignalRBridge(IHubContext<DriverStatusHub> hub, IDriverStatusSnapshotStore store)
    {
        _hub = hub;
        _store = store;
        ReceiveAsync<DriverHealthChanged>(ForwardAsync);
        Receive<SubscribeAck>(_ => { /* DPS confirmation */ });
    }

    /// <inheritdoc />
    protected override void PreStart() =>
        DistributedPubSub.Get(Context.System).Mediator.Tell(new Subscribe(TopicName, Self));

    private async Task ForwardAsync(DriverHealthChanged msg)
    {
        try
        {
            _store.Upsert(msg);
            await _hub.Clients.Group(DriverStatusHub.GroupName(msg.DriverInstanceId))
                .SendAsync(DriverStatusHub.MethodName, msg);
        }
        catch (Exception ex)
        {
            _log.Warning(ex, "DriverStatusSignalRBridge: SignalR push failed (instance={Instance})", msg.DriverInstanceId);
        }
    }
}