dohertj2/ScadaBridge
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

feat(scripts): realign Test Run with runtime API, add anonymous-object calls and instance binding

Browse Source 
The Test Run sandbox and Monaco analysis modelled a script API that had
drifted from the site runtime's ScriptGlobals, so real scripts failed to
compile in Test Run. Realign both to the runtime surface
(Instance/Scripts/ExternalSystem/Attributes/Children/Parent) and drop the
duplicate ScriptHost stub so the two cannot diverge again.

- Script calls (Scripts.CallShared, Instance.CallScript, Route.To().Call)
  accept an anonymous object instead of a hand-built dictionary, via a
  shared ScriptArgs normalizer; existing dictionary calls still compile.
- Test Run can optionally bind to a deployed instance, so Instance/
  Attributes/CallScript route to it cross-site; adds site-side
  RouteToGetAttributes/RouteToSetAttributes handlers.
- Adds Test Run panels to the API method and template script editors.
- Fixes the TestDatabaseQuery seed script, which queried a table that
  never existed.

Also commits unrelated in-progress work already in the tree: the health
monitoring report loop, site streaming changes, and the Admin/Design
data-connection and SMTP page reorganization.
This commit is contained in:
Joseph Doherty
2026-05-16 03:37:56 -04:00
parent d7b05b40e9
commit 295150751f
50 changed files with 2926 additions and 550 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ScadaLink.SiteRuntime/Streaming/SiteStreamManager.cs
+56 -60
View File
@@ -11,7 +11,8 @@ namespace ScadaLink.SiteRuntime.Streaming;
/// <summary>
/// WP-23: Site-Wide Akka Stream — manages a broadcast stream for attribute value
/// and alarm state changes. Instance Actors publish events via fire-and-forget Tell.
/// Subscribers get per-subscriber bounded buffers with drop-oldest overflow.
/// A BroadcastHub fans events out to per-subscriber graphs, each filtered by
/// instance name and bounded by a drop-oldest buffer.
///
/// Filterable by instance name for debug view (WP-25).
/// Implements ISiteStreamSubscriber so the gRPC server can subscribe actors
@@ -20,11 +21,13 @@ namespace ScadaLink.SiteRuntime.Streaming;
public class SiteStreamManager : ISiteStreamSubscriber
{
private ActorSystem? _system;
private IMaterializer? _materializer;
private readonly int _bufferSize;
private readonly ILogger<SiteStreamManager> _logger;
private readonly object _lock = new();
private IActorRef? _sourceActor;
private Source<ISiteStreamEvent, NotUsed>? _hubSource;
private readonly Dictionary<string, SubscriptionInfo> _subscriptions = new();
public SiteStreamManager(
@@ -36,64 +39,73 @@ public class SiteStreamManager : ISiteStreamSubscriber
}
/// <summary>
/// Initializes the stream source. Must be called after ActorSystem is ready.
/// Initializes the broadcast stream. Must be called after ActorSystem is ready.
/// The ActorSystem is passed here rather than via the constructor so that
/// SiteStreamManager can be created by DI before the actor system exists.
/// </summary>
public void Initialize(ActorSystem system)
{
_system = system;
var materializer = _system.Materializer();
_materializer = _system.Materializer();
var source = Source.ActorRef<ISiteStreamEvent>(
_bufferSize,
OverflowStrategy.DropHead);
var (sourceActor, hubSource) = Source.ActorRef<ISiteStreamEvent>(
_bufferSize,
OverflowStrategy.DropHead)
.ToMaterialized(
BroadcastHub.Sink<ISiteStreamEvent>(bufferSize: 256),
Keep.Both)
.Run(_materializer);
var (actorRef, _) = source
.PreMaterialize(materializer);
_sourceActor = actorRef;
_sourceActor = sourceActor;
_hubSource = hubSource;
_logger.LogInformation(
"SiteStreamManager initialized with buffer size {BufferSize}", _bufferSize);
"SiteStreamManager initialized with publish buffer size {BufferSize}", _bufferSize);
}
/// <summary>
/// Publishes an attribute value change to the stream.
/// Publishes an attribute value change to the broadcast hub.
/// Fire-and-forget — never blocks the calling actor.
/// </summary>
public void PublishAttributeValueChanged(AttributeValueChanged changed)
{
_sourceActor?.Tell(changed);
// Also forward to filtered subscribers
ForwardToSubscribers(changed.InstanceUniqueName, changed);
}
/// <summary>
/// Publishes an alarm state change to the stream.
/// Publishes an alarm state change to the broadcast hub.
/// Fire-and-forget — never blocks the calling actor.
/// </summary>
public void PublishAlarmStateChanged(AlarmStateChanged changed)
{
_sourceActor?.Tell(changed);
// Also forward to filtered subscribers
ForwardToSubscribers(changed.InstanceUniqueName, changed);
}
/// <summary>
/// WP-25: Subscribe to events for a specific instance (debug view).
/// Returns a subscription ID for unsubscribing.
/// Materializes a per-subscriber filtered stream off the BroadcastHub
/// with a drop-oldest buffer; returns a subscription ID for unsubscribing.
/// </summary>
public string Subscribe(string instanceName, IActorRef subscriber)
{
if (_hubSource is null || _materializer is null)
throw new InvalidOperationException("SiteStreamManager.Initialize must be called before Subscribe");
var subscriptionId = Guid.NewGuid().ToString();
var capturedInstance = instanceName;
var capturedSubscriber = subscriber;
var killSwitch = _hubSource
.Where(ev => ev.InstanceUniqueName == capturedInstance)
.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(
instanceName, subscriber, DateTimeOffset.UtcNow);
instanceName, subscriber, killSwitch, DateTimeOffset.UtcNow);
}
_logger.LogDebug(
@@ -104,44 +116,47 @@ public class SiteStreamManager : ISiteStreamSubscriber
}
/// <summary>
/// WP-25: Unsubscribe from instance events.
/// WP-25: Unsubscribe from instance events. Shuts down the per-subscriber
/// stream graph via its KillSwitch.
/// </summary>
public bool Unsubscribe(string subscriptionId)
{
SubscriptionInfo? info;
lock (_lock)
{
var removed = _subscriptions.Remove(subscriptionId);
if (removed)
{
_logger.LogDebug("Subscriber {SubscriptionId} removed", subscriptionId);
}
return removed;
if (!_subscriptions.Remove(subscriptionId, out info))
return false;
}
info.KillSwitch.Shutdown();
_logger.LogDebug("Subscriber {SubscriptionId} removed", subscriptionId);
return true;
}
/// <summary>
/// WP-25: Remove all subscriptions for a specific subscriber actor.
/// Called when connection is interrupted.
/// Called when a connection is interrupted.
/// </summary>
public void RemoveSubscriber(IActorRef subscriber)
{
List<SubscriptionInfo> toShutdown;
lock (_lock)
{
var toRemove = _subscriptions
var matched = _subscriptions
.Where(kvp => kvp.Value.Subscriber.Equals(subscriber))
.Select(kvp => kvp.Key)
.ToList();
foreach (var kvp in matched)
_subscriptions.Remove(kvp.Key);
toShutdown = matched.Select(kvp => kvp.Value).ToList();
}
foreach (var id in toRemove)
{
_subscriptions.Remove(id);
}
foreach (var info in toShutdown)
info.KillSwitch.Shutdown();
if (toRemove.Count > 0)
{
_logger.LogDebug(
"Removed {Count} subscriptions for disconnected subscriber", toRemove.Count);
}
if (toShutdown.Count > 0)
{
_logger.LogDebug(
"Removed {Count} subscriptions for disconnected subscriber", toShutdown.Count);
}
}
@@ -153,28 +168,9 @@ public class SiteStreamManager : ISiteStreamSubscriber
get { lock (_lock) { return _subscriptions.Count; } }
}
private void ForwardToSubscribers(string instanceName, object message)
{
lock (_lock)
{
foreach (var sub in _subscriptions.Values)
{
if (sub.InstanceName == instanceName)
{
// Fire-and-forget to subscriber
sub.Subscriber.Tell(message);
}
}
}
}
private record SubscriptionInfo(
string InstanceName,
IActorRef Subscriber,
IKillSwitch KillSwitch,
DateTimeOffset SubscribedAt);
}
/// <summary>
/// Marker interface for events published to the site stream.
/// </summary>
public interface ISiteStreamEvent { }