using Akka.Actor;
using Akka.Hosting;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.DependencyInjection.Extensions;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
using ZB.MOM.WW.OtOpcUa.Commons.Interfaces;
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
using ZB.MOM.WW.OtOpcUa.Configuration;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian;
using ZB.MOM.WW.OtOpcUa.OpcUaServer;
using ZB.MOM.WW.OtOpcUa.Runtime.Drivers;
using ZB.MOM.WW.OtOpcUa.Runtime.Health;
using ZB.MOM.WW.OtOpcUa.Runtime.Historian;
using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa;
using ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags;
namespace ZB.MOM.WW.OtOpcUa.Runtime;
public static class ServiceCollectionExtensions
{
public const string DriverRole = "driver";
public const string DriverHostActorName = "driver-host";
public const string DbHealthProbeActorName = "db-health";
public const string HistorianAdapterActorName = "historian-adapter";
public const string DependencyMuxActorName = "dependency-mux";
public const string OpcUaPublishActorName = "opcua-publish";
///
/// Registers shared runtime services. Currently binds
/// to as the default; production deployments
/// override this with SqliteStoreAndForwardSink wrapping WonderwareHistorianClient.
/// Call this BEFORE AddAkka.
///
/// The service collection to register with.
public static IServiceCollection AddOtOpcUaRuntime(this IServiceCollection services)
{
services.TryAddSingleton(NullAlarmHistorianSink.Instance);
services.TryAddSingleton(NullDriverFactory.Instance);
services.TryAddSingleton(NullOpcUaAddressSpaceSink.Instance);
services.TryAddSingleton(NullServiceLevelPublisher.Instance);
services.TryAddSingleton();
return services;
}
///
/// Spawns the per-node driver-role actors on the host's :
/// (one per node),
/// (consumed by the health endpoint + redundancy calc), and
/// wrapping the registered .
///
/// Mirror of WithOtOpcUaControlPlaneSingletons for the driver role. Both must
/// be registered on the same as the cluster
/// bootstrap so the actors share the host's ActorSystem.
///
/// Wire from the fused Host's Program.cs when the node carries the driver role:
///
/// services.AddOtOpcUaRuntime();
/// services.AddAkka("otopcua", (ab, sp) => { ab.WithOtOpcUaClusterBootstrap(sp); if (hasDriver) ab.WithOtOpcUaRuntimeActors(); });
///
///
/// The Akka configuration builder.
public static AkkaConfigurationBuilder WithOtOpcUaRuntimeActors(this AkkaConfigurationBuilder builder)
{
// Production cluster HOCON (akka.conf) carries this dispatcher block, but consumers that
// bootstrap their own HOCON (e.g. ServiceCollectionExtensionsTests) wouldn't pick it up
// — OpcUaPublishActor.Props pins itself to opcua-synchronized-dispatcher and Akka throws
// ConfigurationException if it doesn't exist. Prepend a fallback so the runtime extension
// is self-contained.
builder.AddHocon(@"
opcua-synchronized-dispatcher {
type = ""PinnedDispatcher""
executor = ""thread-pool-executor""
throughput = 1
}
", HoconAddMode.Prepend);
builder.WithActors((system, registry, resolver) =>
{
var dbFactory = resolver.GetService>();
var roleInfo = resolver.GetService();
// Fallback to Null* if AddOtOpcUaRuntime wasn't called (e.g., test harnesses).
var historianSink = resolver.GetService() ?? NullAlarmHistorianSink.Instance;
var driverFactory = resolver.GetService() ?? NullDriverFactory.Instance;
var addressSpaceSink = resolver.GetService() ?? NullOpcUaAddressSpaceSink.Instance;
var serviceLevel = resolver.GetService() ?? NullServiceLevelPublisher.Instance;
var loggerFactory = resolver.GetService() ?? NullLoggerFactory.Instance;
var healthPublisher = resolver.GetService() ?? NullDriverHealthPublisher.Instance;
var dbHealth = system.ActorOf(
DbHealthProbeActor.Props(dbFactory),
DbHealthProbeActorName);
registry.Register(dbHealth);
// Dependency mux must be spawned before DriverHostActor so the host can forward
// AttributeValuePublished into it from the very first driver spawn.
var mux = system.ActorOf(DependencyMuxActor.Props(), DependencyMuxActorName);
registry.Register(mux);
// OPC UA publish actor — pinned dispatcher, owns the address-space side of the
// pipeline. Phase7Applier is constructed here so the actor + applier share the
// same sink reference (when DeferredAddressSpaceSink swaps later, both see it).
var applier = new Phase7Applier(addressSpaceSink, loggerFactory.CreateLogger());
var publishActor = system.ActorOf(
OpcUaPublishActor.Props(
sink: addressSpaceSink,
serviceLevel: serviceLevel,
localNode: roleInfo.LocalNode,
dbFactory: dbFactory,
applier: applier),
OpcUaPublishActorName);
registry.Register(publishActor);
var driverHost = system.ActorOf(
DriverHostActor.Props(dbFactory, roleInfo.LocalNode, coordinator: null,
driverFactory: driverFactory, localRoles: roleInfo.LocalRoles,
dependencyMux: mux,
opcUaPublishActor: publishActor,
healthPublisher: healthPublisher),
DriverHostActorName);
registry.Register(driverHost);
var historian = system.ActorOf(
HistorianAdapterActor.Props(historianSink),
HistorianAdapterActorName);
registry.Register(historian);
});
return builder;
}
}
/// Marker key types used by Akka.Hosting to resolve runtime actors from the registry.
public sealed class DriverHostActorKey { }
public sealed class DbHealthProbeActorKey { }
public sealed class HistorianAdapterActorKey { }
public sealed class DependencyMuxActorKey { }
public sealed class OpcUaPublishActorKey { }