feat(mesh): route deploy, driver-control and alarm commands through the mesh router

Phase 2 Task 7. The five central->node publish sites -- ConfigPublishCoordinator's deploy dispatch, and AdminOperationsActor's restart, reconnect, acknowledge and shelve -- now hand a MeshCommand to the transport router instead of telling the DistributedPubSub mediator directly. Neither actor knows which transport is in force any more. _meshRouter is NULLABLE with a mediator fallback, and that is deliberate: about a dozen existing coordinator and admin-operations tests construct these actors without a router, and they must keep exercising the real DPS path rather than a silently-disabled one. The fallback is not a permanent seam -- Phase 6 deletes it with the DPS branch and the single mesh. Wiring note: the comm actor's WithActors block moved ABOVE the singleton registrations so both singletons can resolve it with registry.Get, which is the ordering-by-registration idiom AdminOperationsActor already uses for the coordinator. An earlier version used ActorSelection.ResolveOne().GetAwaiter() .GetResult() inside the props factory -- that blocks inside actor construction and races the very spawn it waits for. Sabotage-verified by inverting the branch so the mediator always wins: all six new tests go red, including the fallback test (which proves it is asserting the mediator path rather than passing by accident). Suites after: ControlPlane 118/118, Runtime 450/450, Host.IntegrationTests 196/202 -- sole failure AbCip_Green_AgainstSim, the fixture baseline that fails on master too. Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW
2026-07-22 16:50:27 -04:00
parent 16d598560a
commit 2a2e54f083
4 changed files with 243 additions and 45 deletions
@@ -4,6 +4,7 @@ using Akka.Event;
 using Microsoft.EntityFrameworkCore;
 using ZB.MOM.WW.OtOpcUa.Commons.Messages.Admin;
 using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy;
+using ZB.MOM.WW.OtOpcUa.Commons.Messages.Mesh;
 using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
 using ZB.MOM.WW.OtOpcUa.Commons.Types;
 using ZB.MOM.WW.OtOpcUa.Configuration;
@@ -26,6 +27,13 @@ public sealed class AdminOperationsActor : ReceiveActor
    private readonly IActorRef _coordinator;
    private readonly IReadOnlyDictionary<string, IDriverProbe> _probesByType;
    private readonly TagConfigValidationMode _tagConfigValidationMode;
+
+    /// <summary>
+    ///     Where central→node commands are handed off (per-cluster mesh Phase 2), or
+    ///     <see langword="null"/> to publish on the mediator directly.
+    /// </summary>
+    private readonly IActorRef? _meshRouter;
+
    private readonly ILoggingAdapter _log = Context.GetLogger();

    /// <summary>Creates actor props for the admin operations actor.</summary>
@@ -33,29 +41,38 @@ public sealed class AdminOperationsActor : ReceiveActor
    /// <param name="coordinator">Reference to the deployment coordinator actor.</param>
    /// <param name="probes">Driver probes registered in DI; keyed by DriverType (case-insensitive).</param>
    /// <param name="tagConfigValidationMode">Deploy-gate TagConfig strictness mode (Warn default | Error opt-in).</param>
+    /// <param name="meshRouter">
+    ///     Central comm actor commands are handed to, or <see langword="null"/> to publish on the
+    ///     DistributedPubSub mediator directly (the pre-Phase-2 behaviour).
+    /// </param>
    /// <returns>Props configured to create an AdminOperationsActor.</returns>
    public static Props Props(
        IDbContextFactory<OtOpcUaConfigDbContext> dbFactory,
        IActorRef coordinator,
        IEnumerable<IDriverProbe> probes,
-        TagConfigValidationMode tagConfigValidationMode = TagConfigValidationMode.Warn) =>
-        Akka.Actor.Props.Create(() => new AdminOperationsActor(dbFactory, coordinator, probes, tagConfigValidationMode));
+        TagConfigValidationMode tagConfigValidationMode = TagConfigValidationMode.Warn,
+        IActorRef? meshRouter = null) =>
+        Akka.Actor.Props.Create(() =>
+            new AdminOperationsActor(dbFactory, coordinator, probes, tagConfigValidationMode, meshRouter));

    /// <summary>Initializes a new instance of the AdminOperationsActor.</summary>
    /// <param name="dbFactory">Factory for creating config database contexts.</param>
    /// <param name="coordinator">Reference to the deployment coordinator actor.</param>
    /// <param name="probes">Driver probes registered in DI; keyed by DriverType (case-insensitive).</param>
    /// <param name="tagConfigValidationMode">Deploy-gate TagConfig strictness mode (Warn default | Error opt-in).</param>
+    /// <param name="meshRouter">Central comm actor, or <see langword="null"/> for the mediator.</param>
    public AdminOperationsActor(
        IDbContextFactory<OtOpcUaConfigDbContext> dbFactory,
        IActorRef coordinator,
        IEnumerable<IDriverProbe> probes,
-        TagConfigValidationMode tagConfigValidationMode = TagConfigValidationMode.Warn)
+        TagConfigValidationMode tagConfigValidationMode = TagConfigValidationMode.Warn,
+        IActorRef? meshRouter = null)
    {
        _dbFactory = dbFactory;
        _coordinator = coordinator;
        _probesByType = probes.ToDictionary(p => p.DriverType, StringComparer.OrdinalIgnoreCase);
        _tagConfigValidationMode = tagConfigValidationMode;
+        _meshRouter = meshRouter;

        ReceiveAsync<StartDeployment>(HandleStartDeploymentAsync);
        ReceiveAsync<TestDriverConnect>(HandleTestDriverConnectAsync);
@@ -65,6 +82,23 @@ public sealed class AdminOperationsActor : ReceiveActor
        Receive<ShelveAlarmCommand>(HandleShelveAlarm);
    }

+    /// <summary>
+    ///     Hands a central→node command to the mesh transport router, or publishes it directly when
+    ///     no router is wired.
+    /// </summary>
+    /// <remarks>
+    ///     The fallback keeps ~12 existing tests (and any node wired before per-cluster mesh Phase 2)
+    ///     on exactly the pre-Phase-2 behaviour. It is not a permanent seam: Phase 6 deletes it along
+    ///     with the DistributedPubSub branch and the single mesh.
+    /// </remarks>
+    /// <param name="topic">The legacy DistributedPubSub topic.</param>
+    /// <param name="message">The command.</param>
+    private void Dispatch(string topic, object message)
+    {
+        if (_meshRouter is not null) _meshRouter.Tell(new MeshCommand(topic, message));
+        else DistributedPubSub.Get(Context.System).Mediator.Tell(new Publish(topic, message));
+    }
+
    /// <summary>
    ///     AdminUI Acknowledge path. Maps the control-plane command to a
    ///     <see cref="AlarmCommand"/> (<c>Operation = "Acknowledge"</c>) and publishes it onto the
@@ -85,7 +119,7 @@ public sealed class AdminOperationsActor : ReceiveActor
                Comment: msg.Comment,
                UnshelveAtUtc: null);

-            DistributedPubSub.Get(Context.System).Mediator.Tell(new Publish(AlarmCommandsTopic.Name, alarmCmd));
+            Dispatch(AlarmCommandsTopic.Name, alarmCmd);

            _log.Info("AdminOps: Acknowledge published for alarm {AlarmId} by {User}", msg.AlarmId, msg.User);
            replyTo.Tell(new AcknowledgeAlarmResult(true, null, msg.CorrelationId));
@@ -132,7 +166,7 @@ public sealed class AdminOperationsActor : ReceiveActor
                Comment: msg.Comment,
                UnshelveAtUtc: unshelveAt);

-            DistributedPubSub.Get(Context.System).Mediator.Tell(new Publish(AlarmCommandsTopic.Name, alarmCmd));
+            Dispatch(AlarmCommandsTopic.Name, alarmCmd);

            _log.Info("AdminOps: {Operation} published for alarm {AlarmId} by {User}", operation, msg.AlarmId, msg.User);
            replyTo.Tell(new ShelveAlarmResult(true, null, msg.CorrelationId));
@@ -394,7 +428,7 @@ public sealed class AdminOperationsActor : ReceiveActor
        {
            // Broadcast to every DriverHostActor on every node via the driver-control DPS topic.
            // Only the host that owns the instance will act; others ignore it (id not found in _children).
-            DistributedPubSub.Get(Context.System).Mediator.Tell(new Publish(DriverControlTopic.Name, msg));
+            Dispatch(DriverControlTopic.Name, msg);

            await using var db = await _dbFactory.CreateDbContextAsync();
            db.ConfigEdits.Add(new ConfigEdit
@@ -424,7 +458,7 @@ public sealed class AdminOperationsActor : ReceiveActor
        try
        {
            // Broadcast to every DriverHostActor; only the one owning the instance reacts.
-            DistributedPubSub.Get(Context.System).Mediator.Tell(new Publish(DriverControlTopic.Name, msg));
+            Dispatch(DriverControlTopic.Name, msg);

            await using var db = await _dbFactory.CreateDbContextAsync();
            db.ConfigEdits.Add(new ConfigEdit
@@ -3,6 +3,7 @@ using Akka.Cluster.Tools.PublishSubscribe;
 using Akka.Event;
 using Microsoft.EntityFrameworkCore;
 using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy;
+using ZB.MOM.WW.OtOpcUa.Commons.Messages.Mesh;
 using ZB.MOM.WW.OtOpcUa.Commons.Types;
 using ZB.MOM.WW.OtOpcUa.Configuration;
 using ZB.MOM.WW.OtOpcUa.Configuration.Entities;
@@ -60,6 +61,17 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers

    private readonly IDbContextFactory<OtOpcUaConfigDbContext> _dbFactory;
    private readonly TimeSpan _applyDeadline;
+
+    /// <summary>
+    ///     Where central→node commands are handed off (per-cluster mesh Phase 2), or
+    ///     <see langword="null"/> to publish on the mediator directly.
+    /// </summary>
+    /// <remarks>
+    ///     Null in tests and on any node wired before Phase 2, where the fallback reproduces the
+    ///     pre-Phase-2 behaviour exactly. The fallback is not a permanent seam — Phase 6 deletes it
+    ///     along with the DistributedPubSub branch and the single mesh.
+    /// </remarks>
+    private readonly IActorRef? _meshRouter;
    private readonly ILoggingAdapter _log = Context.GetLogger();
    // NodeId equality here is case-insensitive (by Value) to match the case-insensitive ClusterId/
    // NodeId scoping in DeploymentArtifact.ResolveClusterScope — so an ack from a node whose address
@@ -78,22 +90,31 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers
    /// <param name="dbFactory">The database context factory for accessing configuration data.</param>
    /// <param name="applyDeadline">The timeout for waiting for apply acknowledgments from cluster nodes.</param>
    /// <returns>The Akka.NET <see cref="Akka.Actor.Props"/> used to spawn this actor.</returns>
+    /// <param name="meshRouter">
+    ///     Central comm actor the dispatch is handed to, or <see langword="null"/> to publish on the
+    ///     DistributedPubSub mediator directly (the pre-Phase-2 behaviour).
+    /// </param>
    public static Props Props(
        IDbContextFactory<OtOpcUaConfigDbContext> dbFactory,
-        TimeSpan? applyDeadline = null) =>
-        Akka.Actor.Props.Create(() => new ConfigPublishCoordinator(dbFactory, applyDeadline ?? DefaultApplyDeadline));
+        TimeSpan? applyDeadline = null,
+        IActorRef? meshRouter = null) =>
+        Akka.Actor.Props.Create(() =>
+            new ConfigPublishCoordinator(dbFactory, applyDeadline ?? DefaultApplyDeadline, meshRouter));

    /// <summary>
    ///     Initializes a new instance of the <see cref="ConfigPublishCoordinator"/> class.
    /// </summary>
    /// <param name="dbFactory">The database context factory for persisting deployment state.</param>
    /// <param name="applyDeadline">The timeout for waiting for per-node apply acknowledgments.</param>
+    /// <param name="meshRouter">Central comm actor, or <see langword="null"/> for the mediator.</param>
    public ConfigPublishCoordinator(
        IDbContextFactory<OtOpcUaConfigDbContext> dbFactory,
-        TimeSpan applyDeadline)
+        TimeSpan applyDeadline,
+        IActorRef? meshRouter = null)
    {
        _dbFactory = dbFactory;
        _applyDeadline = applyDeadline;
+        _meshRouter = meshRouter;
        Receive<DispatchDeployment>(HandleDispatch);
        Receive<ApplyAck>(HandleAck);
        Receive<DeadlineElapsed>(HandleDeadline);
@@ -172,7 +193,10 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers
            db.SaveChanges();
        }

-        DistributedPubSub.Get(Context.System).Mediator.Tell(new Publish(DeploymentsTopic, msg));
+        // Per-cluster mesh Phase 2: hand off to the transport router rather than publishing
+        // directly, so this actor no longer knows which transport is in force.
+        if (_meshRouter is not null) _meshRouter.Tell(new MeshCommand(DeploymentsTopic, msg));
+        else DistributedPubSub.Get(Context.System).Mediator.Tell(new Publish(DeploymentsTopic, msg));
        Timers.StartSingleTimer(DeadlineTimerKey, new DeadlineElapsed(msg.DeploymentId), _applyDeadline);

        if (_expectedAcks.Count == 0)
@@ -53,12 +53,49 @@ public static class ServiceCollectionExtensions
        var singletonOptions = new ClusterSingletonOptions { Role = AdminRole };
        var proxyOptions = new ClusterSingletonOptions { Role = AdminRole };

+        // Per-cluster mesh Phase 2: the central end of the central↔node command boundary.
+        //
+        // A plain WithActors, NOT a singleton — and that is the load-bearing part. A driver node's
+        // ClusterClient rotates across contact points and must find a live comm actor at whichever
+        // central node answers. As a singleton it would exist on one central node only, and the
+        // rotation would fail silently against the other: acks would land sometimes, depending on
+        // which contact the client happened to settle on.
+        builder.WithActors((system, registry, resolver) =>
+        {
+            var dbFactory = resolver.GetService<IDbContextFactory<OtOpcUaConfigDbContext>>();
+            var options = resolver.GetService<IOptions<MeshTransportOptions>>().Value;
+
+            var actor = system.ActorOf(
+                CentralCommunicationActor.Props(
+                    dbFactory,
+                    options,
+                    new DefaultMeshClusterClientFactory(),
+                    // Resolved lazily, per message, so this actor never depends on the order in
+                    // which Akka.Hosting materialises the singleton proxy relative to this block.
+                    () => registry.TryGet<ConfigPublishCoordinatorKey>(out var coordinator)
+                        ? coordinator
+                        : null),
+                MeshPaths.CentralCommunicationName);
+
+            // Registered unconditionally, in both transport modes. Under Dps nothing dials it, and
+            // an idle registration costs nothing — whereas registering only under ClusterClient
+            // would mean flipping the flag on a running fleet requires a central restart before any
+            // node can ack, turning a config change back into a deployment.
+            ClusterClientReceptionist.Get(system).RegisterService(actor);
+            registry.Register<CentralCommunicationActorKey>(actor);
+        });
+
        builder.WithSingleton<ConfigPublishCoordinatorKey>(
            ConfigPublishSingletonName,
            (system, registry, resolver) =>
            {
                var dbFactory = resolver.GetService<IDbContextFactory<OtOpcUaConfigDbContext>>();
-                return ConfigPublishCoordinator.Props(dbFactory);
+                // Registered above, so it is already in the registry — the same
+                // ordering-by-registration the AdminOperations singleton relies on to resolve the
+                // coordinator. Deliberately NOT an ActorSelection.ResolveOne here: that blocks
+                // inside a props factory and races the spawn it is waiting for.
+                var meshRouter = registry.Get<CentralCommunicationActorKey>();
+                return ConfigPublishCoordinator.Props(dbFactory, applyDeadline: null, meshRouter: meshRouter);
            },
            singletonOptions);

@@ -74,7 +111,8 @@ public static class ServiceCollectionExtensions
                var mode = Enum.TryParse<TagConfigValidationMode>(
                    resolver.GetService<IConfiguration>()?["Deployment:TagConfigValidationMode"],
                    ignoreCase: true, out var m) ? m : TagConfigValidationMode.Warn;
-                return AdminOperationsActor.Props(dbFactory, coordinator, probes, mode);
+                var meshRouter = registry.Get<CentralCommunicationActorKey>();
+                return AdminOperationsActor.Props(dbFactory, coordinator, probes, mode, meshRouter);
            },
            singletonOptions);

@@ -117,38 +155,6 @@ public static class ServiceCollectionExtensions
            singletonOptions,
            createProxyToo: false);

-        // Per-cluster mesh Phase 2: the central end of the central↔node command boundary.
-        //
-        // A plain WithActors, NOT a singleton — and that is the load-bearing part. A driver node's
-        // ClusterClient rotates across contact points and must find a live comm actor at whichever
-        // central node answers. As a singleton it would exist on one central node only, and the
-        // rotation would fail silently against the other: acks would land sometimes, depending on
-        // which contact the client happened to settle on.
-        builder.WithActors((system, registry, resolver) =>
-        {
-            var dbFactory = resolver.GetService<IDbContextFactory<OtOpcUaConfigDbContext>>();
-            var options = resolver.GetService<IOptions<MeshTransportOptions>>().Value;
-
-            var actor = system.ActorOf(
-                CentralCommunicationActor.Props(
-                    dbFactory,
-                    options,
-                    new DefaultMeshClusterClientFactory(),
-                    // Resolved lazily, per message, so this actor never depends on the order in
-                    // which Akka.Hosting materialises the singleton proxy relative to this block.
-                    () => registry.TryGet<ConfigPublishCoordinatorKey>(out var coordinator)
-                        ? coordinator
-                        : null),
-                MeshPaths.CentralCommunicationName);
-
-            // Registered unconditionally, in both transport modes. Under Dps nothing dials it, and
-            // an idle registration costs nothing — whereas registering only under ClusterClient
-            // would mean flipping the flag on a running fleet requires a central restart before any
-            // node can ack, turning a config change back into a deployment.
-            ClusterClientReceptionist.Get(system).RegisterService(actor);
-            registry.Register<CentralCommunicationActorKey>(actor);
-        });
-
        return builder;
    }
 }
@@ -0,0 +1,134 @@
+using Akka.Actor;
+using Akka.Cluster.Tools.PublishSubscribe;
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Commons.Messages.Admin;
+using ZB.MOM.WW.OtOpcUa.Commons.Messages.Mesh;
+using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
+using ZB.MOM.WW.OtOpcUa.Commons.Types;
+using ZB.MOM.WW.OtOpcUa.ControlPlane.AdminOperations;
+using ZB.MOM.WW.OtOpcUa.ControlPlane.Coordinators;
+using ZB.MOM.WW.OtOpcUa.ControlPlane.Tests.Harness;
+using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+
+namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Tests.Communication;
+
+/// <summary>
+///     Proves the five central→node publish sites go through the transport router when one is
+///     wired, and fall back to the mediator when it is not.
+/// </summary>
+/// <remarks>
+///     The fallback is what keeps the ~12 pre-existing coordinator and admin-operations tests
+///     meaningful: they construct these actors without a router and must still exercise the real
+///     DistributedPubSub path, not a silently-disabled one.
+/// </remarks>
+public class MeshRouterDispatchTests : ControlPlaneActorTestBase
+{
+    private const string DeploymentsTopic = "deployments";
+
+    [Fact]
+    public void Deploy_dispatch_goes_to_the_router_when_one_is_wired()
+    {
+        var router = CreateTestProbe();
+        var dbFactory = NewInMemoryDbFactory();
+        var coordinator = Sys.ActorOf(
+            ConfigPublishCoordinator.Props(dbFactory, applyDeadline: null, meshRouter: router.Ref));
+
+        var deploymentId = new DeploymentId(Guid.NewGuid());
+        coordinator.Tell(new Commons.Messages.Deploy.DispatchDeployment(
+            deploymentId, new RevisionHash("rev-1"), CorrelationId.NewId()));
+
+        var cmd = router.ExpectMsg<MeshCommand>(TimeSpan.FromSeconds(5));
+        cmd.Topic.ShouldBe(DeploymentsTopic);
+        cmd.Message.ShouldBeOfType<Commons.Messages.Deploy.DispatchDeployment>();
+    }
+
+    [Fact]
+    public void Deploy_dispatch_falls_back_to_the_mediator_when_no_router_is_wired()
+    {
+        var subscriber = CreateTestProbe();
+        DistributedPubSub.Get(Sys).Mediator.Tell(
+            new Subscribe(DeploymentsTopic, subscriber.Ref), subscriber.Ref);
+        subscriber.ExpectMsg<SubscribeAck>(TimeSpan.FromSeconds(5));
+
+        var coordinator = Sys.ActorOf(ConfigPublishCoordinator.Props(NewInMemoryDbFactory()));
+
+        var deploymentId = new DeploymentId(Guid.NewGuid());
+        coordinator.Tell(new Commons.Messages.Deploy.DispatchDeployment(
+            deploymentId, new RevisionHash("rev-1"), CorrelationId.NewId()));
+
+        subscriber.ExpectMsg<Commons.Messages.Deploy.DispatchDeployment>(TimeSpan.FromSeconds(5));
+    }
+
+    [Fact]
+    public void Restart_driver_goes_to_the_router_on_the_driver_control_topic()
+    {
+        var router = CreateTestProbe();
+        var admin = Sys.ActorOf(AdminOperationsActor.Props(
+            NewInMemoryDbFactory(),
+            CreateTestProbe().Ref,
+            Enumerable.Empty<IDriverProbe>(),
+            TagConfigValidationMode.Warn,
+            router.Ref));
+
+        admin.Tell(new RestartDriver("C1", "driver-1", "alice", Guid.NewGuid()));
+
+        var cmd = router.ExpectMsg<MeshCommand>(TimeSpan.FromSeconds(5));
+        cmd.Topic.ShouldBe(DriverControlTopic.Name);
+        cmd.Message.ShouldBeOfType<RestartDriver>();
+    }
+
+    [Fact]
+    public void Reconnect_driver_goes_to_the_router_on_the_driver_control_topic()
+    {
+        var router = CreateTestProbe();
+        var admin = Sys.ActorOf(AdminOperationsActor.Props(
+            NewInMemoryDbFactory(),
+            CreateTestProbe().Ref,
+            Enumerable.Empty<IDriverProbe>(),
+            TagConfigValidationMode.Warn,
+            router.Ref));
+
+        admin.Tell(new ReconnectDriver("C1", "driver-1", "alice", Guid.NewGuid()));
+
+        var cmd = router.ExpectMsg<MeshCommand>(TimeSpan.FromSeconds(5));
+        cmd.Topic.ShouldBe(DriverControlTopic.Name);
+        cmd.Message.ShouldBeOfType<ReconnectDriver>();
+    }
+
+    [Fact]
+    public void Alarm_acknowledge_goes_to_the_router_on_the_alarm_commands_topic()
+    {
+        var router = CreateTestProbe();
+        var admin = Sys.ActorOf(AdminOperationsActor.Props(
+            NewInMemoryDbFactory(),
+            CreateTestProbe().Ref,
+            Enumerable.Empty<IDriverProbe>(),
+            TagConfigValidationMode.Warn,
+            router.Ref));
+
+        admin.Tell(new AcknowledgeAlarmCommand("alarm-1", "alice", "c", CorrelationId.NewId()));
+
+        var cmd = router.ExpectMsg<MeshCommand>(TimeSpan.FromSeconds(5));
+        cmd.Topic.ShouldBe(AlarmCommandsTopic.Name);
+        cmd.Message.ShouldBeOfType<AlarmCommand>();
+    }
+
+    [Fact]
+    public void Alarm_shelve_goes_to_the_router_on_the_alarm_commands_topic()
+    {
+        var router = CreateTestProbe();
+        var admin = Sys.ActorOf(AdminOperationsActor.Props(
+            NewInMemoryDbFactory(),
+            CreateTestProbe().Ref,
+            Enumerable.Empty<IDriverProbe>(),
+            TagConfigValidationMode.Warn,
+            router.Ref));
+
+        admin.Tell(new ShelveAlarmCommand("alarm-1", "alice", ShelveKind.OneShot, null, "c", CorrelationId.NewId()));
+
+        var cmd = router.ExpectMsg<MeshCommand>(TimeSpan.FromSeconds(5));
+        cmd.Topic.ShouldBe(AlarmCommandsTopic.Name);
+        cmd.Message.ShouldBeOfType<AlarmCommand>();
+    }
+}