feat: replace debug view polling with real-time SignalR streaming
The debug view polled every 2s by re-subscribing for full snapshots. Now a persistent DebugStreamBridgeActor on central subscribes once and receives incremental Akka stream events from the site, forwarding them to the Blazor component via callbacks and to the CLI via a new SignalR hub at /hubs/debug-stream. Adds `debug stream` CLI command with auto-reconnect.
This commit is contained in:
Joseph Doherty
@@ -126,7 +126,7 @@ This project contains design documentation for a distributed SCADA system built
|
||||
### UI & Monitoring
|
||||
- Central UI: Blazor Server (ASP.NET Core + SignalR) with Bootstrap CSS. No third-party component frameworks (no Blazorise, MudBlazor, Radzen, etc.). Build custom Blazor components for tables, grids, forms, etc.
|
||||
- UI design: Clean, corporate, internal-use aesthetic. Not flashy. Use the `frontend-design` skill when designing UI pages/components.
|
||||
- Debug view: 2s polling timer. Health dashboard: 10s polling timer. Deployment status: real-time push via SignalR.
|
||||
- Debug view: real-time streaming via DebugStreamBridgeActor. Health dashboard: 10s polling timer. Deployment status: real-time push via SignalR.
|
||||
- Health reports: 30s interval, 60s offline threshold, monotonic sequence numbers, raw error counts per interval.
|
||||
- Dead letter monitoring as a health metric.
|
||||
- Site Event Logging: 30-day retention, 1GB storage cap, daily purge, paginated queries with keyword search.
|
||||
|
||||
@@ -168,8 +168,21 @@ scadalink health parked-messages --site-identifier <site-id> [--page <n>] [--pag
|
||||
### Debug Commands
|
||||
```
|
||||
scadalink debug snapshot --id <id> [--format json|table]
|
||||
scadalink debug stream --id <instanceId> [--url ...] [--username ...] [--password ...]
|
||||
```
|
||||
|
||||
The `debug snapshot` command retrieves a point-in-time snapshot via the HTTP Management API.
|
||||
|
||||
The `debug stream` command streams live attribute values and alarm state changes in real-time using a SignalR WebSocket connection. The CLI connects to the `/hubs/debug-stream` SignalR hub on the central server, authenticates with Basic Auth, and subscribes to the specified instance. Events are printed as they arrive — JSON format (default) outputs one NDJSON object per event; table format shows streaming rows. Press Ctrl+C to disconnect.
|
||||
|
||||
Key behaviors:
|
||||
- **Automatic reconnection**: Uses SignalR's `.WithAutomaticReconnect()` to re-establish the connection on loss.
|
||||
- **Re-subscription**: Automatically re-subscribes to the instance after reconnection.
|
||||
- **Traefik compatible**: Works through the Traefik reverse proxy — WebSocket upgrade is proxied natively.
|
||||
- **Required role**: `Deployment`.
|
||||
|
||||
Unlike `debug snapshot` (which uses the HTTP Management API), `debug stream` uses `Microsoft.AspNetCore.SignalR.Client` as a dependency for its WebSocket transport.
|
||||
|
||||
### Shared Script Commands
|
||||
```
|
||||
scadalink shared-script list [--format json|table]
|
||||
@@ -239,8 +252,10 @@ Configuration is resolved in the following priority order (highest wins):
|
||||
|
||||
- **Commons**: Message contracts (`Messages/Management/`) for command type definitions and registry.
|
||||
- **System.CommandLine**: Command-line argument parsing.
|
||||
- **Microsoft.AspNetCore.SignalR.Client**: SignalR client for the `debug stream` command's WebSocket connection.
|
||||
|
||||
## Interactions
|
||||
|
||||
- **Management Service (via HTTP)**: The CLI's sole runtime dependency. All operations are sent as HTTP POST requests to the Management API endpoint on a central node, which dispatches to the ManagementActor.
|
||||
- **Central Host**: Serves the Management API at `POST /management`. Handles LDAP authentication, role resolution, and ManagementActor dispatch.
|
||||
- **Management Service (via HTTP)**: The primary runtime dependency. All operations except `debug stream` are sent as HTTP POST requests to the Management API endpoint on a central node, which dispatches to the ManagementActor.
|
||||
- **Central Host**: Serves the Management API at `POST /management` and the debug stream SignalR hub at `/hubs/debug-stream`. Handles LDAP authentication, role resolution, and ManagementActor dispatch.
|
||||
- **Debug Stream Hub (via SignalR WebSocket)**: The `debug stream` command connects to the `/hubs/debug-stream` hub on the central server for real-time event streaming. This is the only CLI command that uses a persistent connection rather than request/response.
|
||||
|
||||
@@ -19,12 +19,12 @@ Central cluster only. Sites have no user interface.
|
||||
- A **load balancer** sits in front of the central cluster and routes to the active node.
|
||||
- On central failover, the Blazor Server SignalR circuit is interrupted. The browser automatically attempts to reconnect via SignalR's built-in reconnection logic.
|
||||
- Since sessions use **authentication cookies** carrying an embedded JWT (not server-side state), the user's authentication survives failover — the new active node validates the same cookie-embedded JWT. No re-login required if the token is still valid.
|
||||
- Active debug view polling and in-progress deployment status subscriptions are lost on failover and must be re-opened by the user.
|
||||
- Active debug view streams and in-progress deployment status subscriptions are lost on failover and must be re-opened by the user.
|
||||
- Both central nodes share the same **ASP.NET Data Protection keys** (stored in the configuration database or shared configuration) so that tokens and anti-forgery tokens remain valid across failover.
|
||||
|
||||
## Real-Time Updates
|
||||
|
||||
- **Debug view**: Near-real-time display of attribute values and alarm states, updated via a **2-second polling timer**. This avoids the complexity of cross-cluster streaming while providing responsive feedback — 2s latency is imperceptible for debugging purposes.
|
||||
- **Debug view**: Real-time display of attribute values and alarm states via **streaming**. When the user opens a debug view, a `DebugStreamBridgeActor` on the central side subscribes to the site's Akka stream for the selected instance. The bridge actor delivers an initial `DebugViewSnapshot` followed by ongoing `AttributeValueChanged` and `AlarmStateChanged` events to the Blazor component via callbacks, which call `InvokeAsync(StateHasChanged)` to push UI updates through the built-in SignalR circuit.
|
||||
- **Health dashboard**: Site status, connection health, error rates, and buffer depths update via a **10-second auto-refresh timer**. Since health reports arrive from sites every 30 seconds, a 10s poll interval catches updates within one reporting cycle without unnecessary overhead.
|
||||
- **Deployment status**: Pending/in-progress/success/failed transitions **push to the UI immediately** via SignalR (built into Blazor Server). No polling required for deployment tracking.
|
||||
|
||||
@@ -100,8 +100,11 @@ Central cluster only. Sites have no user interface.
|
||||
|
||||
### Debug View (Deployment Role)
|
||||
- Select a deployed instance and open a live debug view.
|
||||
- Near-real-time polling (2s interval) of all attribute values (with quality and timestamp) and alarm states for that instance.
|
||||
- Initial snapshot of current state followed by periodic polling for updates.
|
||||
- Real-time streaming of all attribute values (with quality and timestamp) and alarm states for that instance.
|
||||
- The `DebugStreamService` creates a `DebugStreamBridgeActor` on the central side that subscribes to the site's Akka stream for the selected instance.
|
||||
- The bridge actor receives an initial `DebugViewSnapshot` followed by ongoing `AttributeValueChanged` and `AlarmStateChanged` events from the site.
|
||||
- Events are delivered to the Blazor component via callbacks, which call `InvokeAsync(StateHasChanged)` to push UI updates through the built-in SignalR circuit.
|
||||
- A pulsing "Live" indicator replaces the static "Connected" badge when streaming is active.
|
||||
- Stream includes attribute values formatted as `[InstanceUniqueName].[AttributePath].[AttributeName]` and alarm states formatted as `[InstanceUniqueName].[AlarmName]`.
|
||||
- Subscribe-on-demand — stream starts when opened, stops when closed.
|
||||
|
||||
|
||||
@@ -50,15 +50,22 @@ Both central and site clusters. Each side has communication actors that handle m
|
||||
- Site applies and acknowledges.
|
||||
|
||||
### 6. Debug Streaming (Site → Central)
|
||||
- **Pattern**: Subscribe/stream with initial snapshot.
|
||||
- Central sends a subscribe request for a specific instance (identified by unique name).
|
||||
- Site requests a **snapshot** of all current attribute values and alarm states from the Instance Actor and sends it to central.
|
||||
- Site then subscribes to the **site-wide Akka stream** filtered by the instance's unique name and forwards attribute value changes and alarm state changes to central.
|
||||
- **Pattern**: Subscribe/push with initial snapshot (no polling).
|
||||
- A **DebugStreamBridgeActor** (one per active debug session) is created on the central cluster by the **DebugStreamService**. The bridge actor sends a `SubscribeDebugViewRequest` to the site via `CentralCommunicationActor`, with itself as the `Sender`. The site's `InstanceActor` registers the bridge actor as the debug subscriber.
|
||||
- Site requests a **snapshot** of all current attribute values and alarm states from the Instance Actor and sends it to the bridge actor.
|
||||
- Site then subscribes to the **site-wide Akka stream** filtered by the instance's unique name and forwards `AttributeValueChanged` and `AlarmStateChanged` events to the bridge actor in real time via Akka remoting.
|
||||
- The bridge actor forwards received events to the consumer via callbacks (Blazor component or SignalR hub).
|
||||
- Attribute value stream messages: `[InstanceUniqueName].[AttributePath].[AttributeName]`, value, quality, timestamp.
|
||||
- Alarm state stream messages: `[InstanceUniqueName].[AlarmName]`, state (active/normal), priority, timestamp.
|
||||
- Central sends an unsubscribe request when the debug view closes. The site removes its stream subscription.
|
||||
- Central sends an unsubscribe request when the debug session ends. The site removes its stream subscription and the bridge actor is stopped.
|
||||
- The stream is session-based and temporary.
|
||||
|
||||
#### Central-Side Debug Stream Components
|
||||
|
||||
- **DebugStreamService**: Singleton service that manages debug stream sessions. Resolves instance ID to unique name and site, creates and tears down `DebugStreamBridgeActor` instances, and provides a clean API for both Blazor components and the SignalR hub.
|
||||
- **DebugStreamBridgeActor**: One per active debug session. Acts as the Akka-level subscriber registered with the site's `InstanceActor`. Receives real-time `AttributeValueChanged` and `AlarmStateChanged` events from the site and forwards them to the consumer via callbacks.
|
||||
- **DebugStreamHub**: SignalR hub at `/hubs/debug-stream` for external consumers (e.g., CLI). Authenticates via Basic Auth + LDAP and requires the **Deployment** role. Server-to-client methods: `OnSnapshot`, `OnAttributeChanged`, `OnAlarmChanged`, `OnStreamTerminated`.
|
||||
|
||||
### 6a. Debug Snapshot (Central → Site)
|
||||
- **Pattern**: Request/Response (one-shot, no subscription).
|
||||
- Central sends a `DebugSnapshotRequest` (identified by instance unique name) to the site.
|
||||
@@ -159,7 +166,7 @@ The ManagementActor is registered at the well-known path `/user/management` on c
|
||||
## Connection Failure Behavior
|
||||
|
||||
- **In-flight messages**: When a connection drops while a request is in flight (e.g., deployment sent but no response received), the Akka ask pattern times out and the caller receives a failure. There is **no automatic retry or buffering at central** — the engineer sees the failure in the UI and re-initiates the action. This is consistent with the design principle that central does not buffer messages.
|
||||
- **Debug streams**: Any connection interruption (failover or network blip) kills the debug stream. The engineer must reopen the debug view in the Central UI to re-establish the subscription with a fresh snapshot. There is no auto-resume.
|
||||
- **Debug streams**: Any connection interruption (failover or network blip) kills the debug stream. The `DebugStreamBridgeActor` is stopped and the consumer is notified via `OnStreamTerminated`. The engineer must reopen the debug view to re-establish the subscription with a fresh snapshot. There is no auto-resume.
|
||||
|
||||
## Failover Behavior
|
||||
|
||||
|
||||
@@ -1,5 +1,9 @@
|
||||
using System.CommandLine;
|
||||
using System.CommandLine.Parsing;
|
||||
using System.Net.Http.Headers;
|
||||
using System.Text;
|
||||
using System.Text.Json;
|
||||
using Microsoft.AspNetCore.SignalR.Client;
|
||||
using ScadaLink.Commons.Messages.Management;
|
||||
|
||||
namespace ScadaLink.CLI.Commands;
|
||||
@@ -11,6 +15,7 @@ public static class DebugCommands
|
||||
var command = new Command("debug") { Description = "Runtime debugging" };
|
||||
|
||||
command.Add(BuildSnapshot(urlOption, formatOption, usernameOption, passwordOption));
|
||||
command.Add(BuildStream(urlOption, formatOption, usernameOption, passwordOption));
|
||||
|
||||
return command;
|
||||
}
|
||||
@@ -28,4 +33,230 @@ public static class DebugCommands
|
||||
});
|
||||
return cmd;
|
||||
}
|
||||
|
||||
private static Command BuildStream(Option<string> urlOption, Option<string> formatOption, Option<string> usernameOption, Option<string> passwordOption)
|
||||
{
|
||||
var idOption = new Option<int>("--id") { Description = "Instance ID", Required = true };
|
||||
var cmd = new Command("stream") { Description = "Stream live attribute values and alarm states in real-time (Ctrl+C to stop)" };
|
||||
cmd.Add(idOption);
|
||||
cmd.SetAction(async (ParseResult result) =>
|
||||
{
|
||||
var instanceId = result.GetValue(idOption);
|
||||
var format = result.GetValue(formatOption) ?? "json";
|
||||
var config = CliConfig.Load();
|
||||
|
||||
var url = result.GetValue(urlOption);
|
||||
if (string.IsNullOrWhiteSpace(url))
|
||||
url = config.ManagementUrl;
|
||||
|
||||
if (string.IsNullOrWhiteSpace(url))
|
||||
{
|
||||
OutputFormatter.WriteError(
|
||||
"No management URL specified. Use --url, set SCADALINK_MANAGEMENT_URL, or add 'managementUrl' to ~/.scadalink/config.json.",
|
||||
"NO_URL");
|
||||
return 1;
|
||||
}
|
||||
|
||||
var username = result.GetValue(usernameOption);
|
||||
var password = result.GetValue(passwordOption);
|
||||
|
||||
if (string.IsNullOrWhiteSpace(username) || string.IsNullOrWhiteSpace(password))
|
||||
{
|
||||
OutputFormatter.WriteError(
|
||||
"Credentials required. Use --username and --password options.",
|
||||
"NO_CREDENTIALS");
|
||||
return 1;
|
||||
}
|
||||
|
||||
return await StreamDebugAsync(url, username, password, instanceId, format);
|
||||
});
|
||||
return cmd;
|
||||
}
|
||||
|
||||
private static async Task<int> StreamDebugAsync(string baseUrl, string username, string password, int instanceId, string format)
|
||||
{
|
||||
var hubUrl = baseUrl.TrimEnd('/') + "/hubs/debug-stream";
|
||||
var credentials = Convert.ToBase64String(Encoding.UTF8.GetBytes($"{username}:{password}"));
|
||||
|
||||
var connection = new HubConnectionBuilder()
|
||||
.WithUrl(hubUrl, options =>
|
||||
{
|
||||
options.Headers.Add("Authorization", $"Basic {credentials}");
|
||||
})
|
||||
.WithAutomaticReconnect()
|
||||
.Build();
|
||||
|
||||
var cts = new CancellationTokenSource();
|
||||
var exitTcs = new TaskCompletionSource<int>(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
|
||||
Console.CancelKeyPress += (_, e) =>
|
||||
{
|
||||
e.Cancel = true;
|
||||
cts.Cancel();
|
||||
};
|
||||
|
||||
var isTable = string.Equals(format, "table", StringComparison.OrdinalIgnoreCase);
|
||||
|
||||
// Register event handlers
|
||||
connection.On<JsonElement>("OnSnapshot", snapshot =>
|
||||
{
|
||||
if (isTable)
|
||||
{
|
||||
Console.WriteLine("=== Initial Snapshot ===");
|
||||
PrintSnapshotTable(snapshot);
|
||||
Console.WriteLine("=== Streaming (Ctrl+C to stop) ===");
|
||||
}
|
||||
else
|
||||
{
|
||||
var obj = new { type = "snapshot", data = snapshot };
|
||||
Console.WriteLine(JsonSerializer.Serialize(obj, new JsonSerializerOptions { WriteIndented = false }));
|
||||
}
|
||||
});
|
||||
|
||||
connection.On<JsonElement>("OnAttributeChanged", changed =>
|
||||
{
|
||||
if (isTable)
|
||||
{
|
||||
var name = changed.TryGetProperty("attributeName", out var n) ? n.GetString() : "?";
|
||||
var value = changed.TryGetProperty("value", out var v) ? v.ToString() : "?";
|
||||
var quality = changed.TryGetProperty("quality", out var q) ? q.GetString() : "?";
|
||||
var ts = changed.TryGetProperty("timestamp", out var t) ? t.GetString() : "?";
|
||||
Console.WriteLine($" ATTR {name,-30} {value,-20} {quality,-10} {ts}");
|
||||
}
|
||||
else
|
||||
{
|
||||
var obj = new { type = "attributeChanged", data = changed };
|
||||
Console.WriteLine(JsonSerializer.Serialize(obj, new JsonSerializerOptions { WriteIndented = false }));
|
||||
}
|
||||
});
|
||||
|
||||
connection.On<JsonElement>("OnAlarmChanged", changed =>
|
||||
{
|
||||
if (isTable)
|
||||
{
|
||||
var name = changed.TryGetProperty("alarmName", out var n) ? n.GetString() : "?";
|
||||
var state = changed.TryGetProperty("state", out var s) ? s.ToString() : "?";
|
||||
var priority = changed.TryGetProperty("priority", out var p) ? p.ToString() : "?";
|
||||
var ts = changed.TryGetProperty("timestamp", out var t) ? t.GetString() : "?";
|
||||
Console.WriteLine($" ALARM {name,-30} {state,-20} P{priority,-9} {ts}");
|
||||
}
|
||||
else
|
||||
{
|
||||
var obj = new { type = "alarmChanged", data = changed };
|
||||
Console.WriteLine(JsonSerializer.Serialize(obj, new JsonSerializerOptions { WriteIndented = false }));
|
||||
}
|
||||
});
|
||||
|
||||
connection.On<string>("OnStreamTerminated", reason =>
|
||||
{
|
||||
Console.Error.WriteLine($"Stream terminated: {reason}");
|
||||
exitTcs.TrySetResult(1);
|
||||
});
|
||||
|
||||
connection.Closed += ex =>
|
||||
{
|
||||
if (!cts.IsCancellationRequested)
|
||||
{
|
||||
Console.Error.WriteLine(ex != null
|
||||
? $"Connection lost: {ex.Message}"
|
||||
: "Connection closed.");
|
||||
}
|
||||
exitTcs.TrySetResult(cts.IsCancellationRequested ? 0 : 1);
|
||||
return Task.CompletedTask;
|
||||
};
|
||||
|
||||
connection.Reconnecting += ex =>
|
||||
{
|
||||
Console.Error.WriteLine($"Reconnecting... ({ex?.Message})");
|
||||
return Task.CompletedTask;
|
||||
};
|
||||
|
||||
connection.Reconnected += _ =>
|
||||
{
|
||||
Console.Error.WriteLine("Reconnected. Re-subscribing...");
|
||||
return connection.InvokeAsync("SubscribeInstance", instanceId);
|
||||
};
|
||||
|
||||
// Connect and subscribe
|
||||
try
|
||||
{
|
||||
await connection.StartAsync(cts.Token);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
OutputFormatter.WriteError($"Connection failed: {ex.Message}", "CONNECTION_FAILED");
|
||||
return 1;
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
await connection.InvokeAsync("SubscribeInstance", instanceId, cts.Token);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
OutputFormatter.WriteError($"Subscribe failed: {ex.Message}", "SUBSCRIBE_FAILED");
|
||||
await connection.DisposeAsync();
|
||||
return 1;
|
||||
}
|
||||
|
||||
if (isTable)
|
||||
{
|
||||
Console.WriteLine($"Connected to instance {instanceId}. Waiting for data...");
|
||||
}
|
||||
|
||||
// Wait for cancellation (Ctrl+C) or stream termination
|
||||
try
|
||||
{
|
||||
await exitTcs.Task.WaitAsync(cts.Token);
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
// Ctrl+C — graceful shutdown
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
await connection.InvokeAsync("UnsubscribeInstance");
|
||||
}
|
||||
catch
|
||||
{
|
||||
// Best effort
|
||||
}
|
||||
|
||||
await connection.DisposeAsync();
|
||||
return exitTcs.Task.IsCompletedSuccessfully ? exitTcs.Task.Result : 0;
|
||||
}
|
||||
|
||||
private static void PrintSnapshotTable(JsonElement snapshot)
|
||||
{
|
||||
if (snapshot.TryGetProperty("attributeValues", out var attrs) && attrs.ValueKind == JsonValueKind.Array)
|
||||
{
|
||||
Console.WriteLine(" Attributes:");
|
||||
Console.WriteLine($" {"Name",-30} {"Value",-20} {"Quality",-10} Timestamp");
|
||||
Console.WriteLine($" {new string('-', 30)} {new string('-', 20)} {new string('-', 10)} {new string('-', 25)}");
|
||||
foreach (var av in attrs.EnumerateArray())
|
||||
{
|
||||
var name = av.TryGetProperty("attributeName", out var n) ? n.GetString() : "?";
|
||||
var value = av.TryGetProperty("value", out var v) ? v.ToString() : "?";
|
||||
var quality = av.TryGetProperty("quality", out var q) ? q.GetString() : "?";
|
||||
var ts = av.TryGetProperty("timestamp", out var t) ? t.GetString() : "?";
|
||||
Console.WriteLine($" {name,-30} {value,-20} {quality,-10} {ts}");
|
||||
}
|
||||
}
|
||||
|
||||
if (snapshot.TryGetProperty("alarmStates", out var alarms) && alarms.ValueKind == JsonValueKind.Array)
|
||||
{
|
||||
Console.WriteLine(" Alarms:");
|
||||
Console.WriteLine($" {"Name",-30} {"State",-20} {"Priority",-10} Timestamp");
|
||||
Console.WriteLine($" {new string('-', 30)} {new string('-', 20)} {new string('-', 10)} {new string('-', 25)}");
|
||||
foreach (var al in alarms.EnumerateArray())
|
||||
{
|
||||
var name = al.TryGetProperty("alarmName", out var n) ? n.GetString() : "?";
|
||||
var state = al.TryGetProperty("state", out var s) ? s.ToString() : "?";
|
||||
var priority = al.TryGetProperty("priority", out var p) ? p.ToString() : "?";
|
||||
var ts = al.TryGetProperty("timestamp", out var t) ? t.GetString() : "?";
|
||||
Console.WriteLine($" {name,-30} {state,-20} P{priority,-9} {ts}");
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1038,6 +1038,25 @@ scadalink --url <url> debug snapshot --id <int>
|
||||
|
||||
The command resolves the instance's site internally and routes the request to the correct site cluster. Returns all attribute values (name, value, quality, timestamp) and alarm states (name, state, priority, timestamp) at the moment the request reaches the site.
|
||||
|
||||
#### `debug stream`
|
||||
|
||||
Stream live attribute values and alarm state changes in real-time using a SignalR WebSocket connection to the central server's `/hubs/debug-stream` hub. Events are printed as they arrive. Press Ctrl+C to disconnect.
|
||||
|
||||
```sh
|
||||
scadalink --url <url> debug stream --id <int>
|
||||
```
|
||||
|
||||
| Option | Required | Description |
|
||||
|--------|----------|-------------|
|
||||
| `--id` | yes | Instance ID |
|
||||
|
||||
The default JSON format outputs one NDJSON object per event line with a `type` field (`snapshot`, `attributeChanged`, or `alarmChanged`). Table format (`--format table`) shows a formatted initial snapshot followed by streaming rows prefixed with `ATTR` or `ALARM`.
|
||||
|
||||
Features:
|
||||
- Automatic reconnection on connection loss with re-subscribe.
|
||||
- Works through the Traefik load balancer (WebSocket upgrade proxied natively).
|
||||
- Requires the `Deployment` role.
|
||||
|
||||
---
|
||||
|
||||
### `audit-log` — Audit log queries
|
||||
|
||||
@@ -11,6 +11,7 @@
|
||||
<InternalsVisibleTo Include="ScadaLink.CLI.Tests" />
|
||||
</ItemGroup>
|
||||
<ItemGroup>
|
||||
<PackageReference Include="Microsoft.AspNetCore.SignalR.Client" Version="9.0.3" />
|
||||
<PackageReference Include="System.CommandLine" Version="2.0.5" />
|
||||
</ItemGroup>
|
||||
<ItemGroup>
|
||||
|
||||
@@ -10,7 +10,7 @@
|
||||
@attribute [Authorize(Policy = AuthorizationPolicies.RequireDeployment)]
|
||||
@inject ITemplateEngineRepository TemplateEngineRepository
|
||||
@inject ISiteRepository SiteRepository
|
||||
@inject CommunicationService CommunicationService
|
||||
@inject DebugStreamService DebugStreamService
|
||||
@inject IJSRuntime JS
|
||||
@implements IDisposable
|
||||
|
||||
@@ -38,11 +38,11 @@
|
||||
</div>
|
||||
<div class="col-md-4">
|
||||
<label class="form-label small">Instance</label>
|
||||
<select class="form-select form-select-sm" @bind="_selectedInstanceName">
|
||||
<option value="">Select instance...</option>
|
||||
<select class="form-select form-select-sm" @bind="_selectedInstanceId" @bind:after="OnInstanceSelectionChanged">
|
||||
<option value="0">Select instance...</option>
|
||||
@foreach (var inst in _siteInstances)
|
||||
{
|
||||
<option value="@inst.UniqueName">@inst.UniqueName (@inst.State)</option>
|
||||
<option value="@inst.Id">@inst.UniqueName (@inst.State)</option>
|
||||
}
|
||||
</select>
|
||||
</div>
|
||||
@@ -50,7 +50,7 @@
|
||||
@if (!_connected)
|
||||
{
|
||||
<button class="btn btn-primary btn-sm" @onclick="Connect"
|
||||
disabled="@(string.IsNullOrEmpty(_selectedInstanceName) || _selectedSiteId == 0 || _connecting)">
|
||||
disabled="@(_selectedInstanceId == 0 || _selectedSiteId == 0 || _connecting)">
|
||||
@if (_connecting) { <span class="spinner-border spinner-border-sm me-1"></span> }
|
||||
Connect
|
||||
</button>
|
||||
@@ -58,7 +58,10 @@
|
||||
else
|
||||
{
|
||||
<button class="btn btn-outline-danger btn-sm" @onclick="Disconnect">Disconnect</button>
|
||||
<span class="badge bg-success align-self-center">Connected</span>
|
||||
<span class="badge bg-success align-self-center">
|
||||
<span class="spinner-grow spinner-grow-sm me-1" style="width: 0.5rem; height: 0.5rem;"></span>
|
||||
Live
|
||||
</span>
|
||||
}
|
||||
</div>
|
||||
</div>
|
||||
@@ -153,7 +156,7 @@
|
||||
private List<Site> _sites = new();
|
||||
private List<Instance> _siteInstances = new();
|
||||
private int _selectedSiteId;
|
||||
private string _selectedInstanceName = string.Empty;
|
||||
private int _selectedInstanceId;
|
||||
private bool _loading = true;
|
||||
private bool _connected;
|
||||
private bool _connecting;
|
||||
@@ -162,7 +165,7 @@
|
||||
private Dictionary<string, AttributeValueChanged> _attributeValues = new();
|
||||
private Dictionary<string, AlarmStateChanged> _alarmStates = new();
|
||||
|
||||
private Timer? _refreshTimer;
|
||||
private DebugStreamSession? _session;
|
||||
private ToastNotification _toast = default!;
|
||||
|
||||
protected override async Task OnInitializedAsync()
|
||||
@@ -183,14 +186,14 @@
|
||||
if (!firstRender) return;
|
||||
|
||||
var storedSiteId = await JS.InvokeAsync<string>("localStorage.getItem", "debugView.siteId");
|
||||
var storedInstanceName = await JS.InvokeAsync<string>("localStorage.getItem", "debugView.instanceName");
|
||||
var storedInstanceId = await JS.InvokeAsync<string>("localStorage.getItem", "debugView.instanceId");
|
||||
|
||||
if (!string.IsNullOrEmpty(storedSiteId) && int.TryParse(storedSiteId, out var siteId)
|
||||
&& !string.IsNullOrEmpty(storedInstanceName))
|
||||
&& !string.IsNullOrEmpty(storedInstanceId) && int.TryParse(storedInstanceId, out var instanceId))
|
||||
{
|
||||
_selectedSiteId = siteId;
|
||||
await LoadInstancesForSite();
|
||||
_selectedInstanceName = storedInstanceName;
|
||||
_selectedInstanceId = instanceId;
|
||||
StateHasChanged();
|
||||
await Connect();
|
||||
}
|
||||
@@ -199,7 +202,7 @@
|
||||
private async Task LoadInstancesForSite()
|
||||
{
|
||||
_siteInstances.Clear();
|
||||
_selectedInstanceName = string.Empty;
|
||||
_selectedInstanceId = 0;
|
||||
if (_selectedSiteId == 0) return;
|
||||
try
|
||||
{
|
||||
@@ -213,58 +216,64 @@
|
||||
}
|
||||
}
|
||||
|
||||
private void OnInstanceSelectionChanged()
|
||||
{
|
||||
// No-op; selection is tracked via _selectedInstanceId binding
|
||||
}
|
||||
|
||||
private async Task Connect()
|
||||
{
|
||||
if (string.IsNullOrEmpty(_selectedInstanceName) || _selectedSiteId == 0) return;
|
||||
if (_selectedInstanceId == 0 || _selectedSiteId == 0) return;
|
||||
_connecting = true;
|
||||
try
|
||||
{
|
||||
var site = _sites.FirstOrDefault(s => s.Id == _selectedSiteId);
|
||||
if (site == null) return;
|
||||
var session = await DebugStreamService.StartStreamAsync(
|
||||
_selectedInstanceId,
|
||||
onEvent: evt =>
|
||||
{
|
||||
switch (evt)
|
||||
{
|
||||
case AttributeValueChanged av:
|
||||
_attributeValues[av.AttributeName] = av;
|
||||
_ = InvokeAsync(StateHasChanged);
|
||||
break;
|
||||
case AlarmStateChanged al:
|
||||
_alarmStates[al.AlarmName] = al;
|
||||
_ = InvokeAsync(StateHasChanged);
|
||||
break;
|
||||
}
|
||||
},
|
||||
onTerminated: () =>
|
||||
{
|
||||
_connected = false;
|
||||
_session = null;
|
||||
_ = InvokeAsync(() =>
|
||||
{
|
||||
_toast.ShowError("Debug stream terminated (site disconnected).");
|
||||
StateHasChanged();
|
||||
});
|
||||
});
|
||||
|
||||
var request = new SubscribeDebugViewRequest(_selectedInstanceName, Guid.NewGuid().ToString("N"));
|
||||
_snapshot = await CommunicationService.SubscribeDebugViewAsync(site.SiteIdentifier, request);
|
||||
_session = session;
|
||||
|
||||
// Populate initial state from snapshot
|
||||
_attributeValues.Clear();
|
||||
foreach (var av in _snapshot.AttributeValues)
|
||||
{
|
||||
foreach (var av in session.InitialSnapshot.AttributeValues)
|
||||
_attributeValues[av.AttributeName] = av;
|
||||
}
|
||||
_alarmStates.Clear();
|
||||
foreach (var al in _snapshot.AlarmStates)
|
||||
{
|
||||
_alarmStates[al.AlarmName] = al;
|
||||
}
|
||||
|
||||
_alarmStates.Clear();
|
||||
foreach (var al in session.InitialSnapshot.AlarmStates)
|
||||
_alarmStates[al.AlarmName] = al;
|
||||
|
||||
_snapshot = session.InitialSnapshot;
|
||||
_connected = true;
|
||||
|
||||
// Persist selection to localStorage for auto-reconnect on refresh
|
||||
await JS.InvokeVoidAsync("localStorage.setItem", "debugView.siteId", _selectedSiteId.ToString());
|
||||
await JS.InvokeVoidAsync("localStorage.setItem", "debugView.instanceName", _selectedInstanceName);
|
||||
await JS.InvokeVoidAsync("localStorage.setItem", "debugView.siteIdentifier", site.SiteIdentifier);
|
||||
await JS.InvokeVoidAsync("localStorage.setItem", "debugView.instanceId", _selectedInstanceId.ToString());
|
||||
|
||||
_toast.ShowSuccess($"Connected to {_selectedInstanceName}");
|
||||
|
||||
// Periodic refresh (simulating SignalR push by re-subscribing)
|
||||
_refreshTimer = new Timer(async _ =>
|
||||
{
|
||||
try
|
||||
{
|
||||
var refreshRequest = new SubscribeDebugViewRequest(_selectedInstanceName, Guid.NewGuid().ToString("N"));
|
||||
var newSnapshot = await CommunicationService.SubscribeDebugViewAsync(site.SiteIdentifier, refreshRequest);
|
||||
foreach (var av in newSnapshot.AttributeValues)
|
||||
_attributeValues[av.AttributeName] = av;
|
||||
foreach (var al in newSnapshot.AlarmStates)
|
||||
_alarmStates[al.AlarmName] = al;
|
||||
_snapshot = newSnapshot;
|
||||
await InvokeAsync(StateHasChanged);
|
||||
}
|
||||
catch
|
||||
{
|
||||
// Connection may have dropped
|
||||
}
|
||||
}, null, TimeSpan.FromSeconds(2), TimeSpan.FromSeconds(2));
|
||||
var instance = _siteInstances.FirstOrDefault(i => i.Id == _selectedInstanceId);
|
||||
_toast.ShowSuccess($"Streaming {instance?.UniqueName ?? "instance"}");
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
@@ -275,23 +284,15 @@
|
||||
|
||||
private async Task Disconnect()
|
||||
{
|
||||
_refreshTimer?.Dispose();
|
||||
_refreshTimer = null;
|
||||
|
||||
if (_connected && _selectedSiteId > 0 && !string.IsNullOrEmpty(_selectedInstanceName))
|
||||
if (_session != null)
|
||||
{
|
||||
var site = _sites.FirstOrDefault(s => s.Id == _selectedSiteId);
|
||||
if (site != null)
|
||||
{
|
||||
var request = new UnsubscribeDebugViewRequest(_selectedInstanceName, Guid.NewGuid().ToString("N"));
|
||||
CommunicationService.UnsubscribeDebugView(site.SiteIdentifier, request);
|
||||
}
|
||||
DebugStreamService.StopStream(_session.SessionId);
|
||||
_session = null;
|
||||
}
|
||||
|
||||
// Clear persisted selection — user explicitly disconnected
|
||||
await JS.InvokeVoidAsync("localStorage.removeItem", "debugView.siteId");
|
||||
await JS.InvokeVoidAsync("localStorage.removeItem", "debugView.instanceName");
|
||||
await JS.InvokeVoidAsync("localStorage.removeItem", "debugView.siteIdentifier");
|
||||
await JS.InvokeVoidAsync("localStorage.removeItem", "debugView.instanceId");
|
||||
|
||||
_connected = false;
|
||||
_snapshot = null;
|
||||
@@ -314,6 +315,9 @@
|
||||
|
||||
public void Dispose()
|
||||
{
|
||||
_refreshTimer?.Dispose();
|
||||
if (_session != null)
|
||||
{
|
||||
DebugStreamService.StopStream(_session.SessionId);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
100
src/ScadaLink.Communication/Actors/DebugStreamBridgeActor.cs
Normal file
100
src/ScadaLink.Communication/Actors/DebugStreamBridgeActor.cs
Normal file
@@ -0,0 +1,100 @@
|
||||
using Akka.Actor;
|
||||
using Akka.Event;
|
||||
using ScadaLink.Commons.Messages.DebugView;
|
||||
using ScadaLink.Commons.Messages.Streaming;
|
||||
|
||||
namespace ScadaLink.Communication.Actors;
|
||||
|
||||
/// <summary>
|
||||
/// Persistent actor (one per active debug session) on the central side.
|
||||
/// Sends SubscribeDebugViewRequest to the site via CentralCommunicationActor (with THIS actor
|
||||
/// as the Sender), so the site's InstanceActor registers this actor as the debug subscriber.
|
||||
/// Stream events flow back via Akka remoting and are forwarded to the consumer via callbacks.
|
||||
/// </summary>
|
||||
public class DebugStreamBridgeActor : ReceiveActor
|
||||
{
|
||||
private readonly ILoggingAdapter _log = Context.GetLogger();
|
||||
private readonly string _siteIdentifier;
|
||||
private readonly string _instanceUniqueName;
|
||||
private readonly string _correlationId;
|
||||
private readonly IActorRef _centralCommunicationActor;
|
||||
private readonly Action<object> _onEvent;
|
||||
private readonly Action _onTerminated;
|
||||
|
||||
public DebugStreamBridgeActor(
|
||||
string siteIdentifier,
|
||||
string instanceUniqueName,
|
||||
string correlationId,
|
||||
IActorRef centralCommunicationActor,
|
||||
Action<object> onEvent,
|
||||
Action onTerminated)
|
||||
{
|
||||
_siteIdentifier = siteIdentifier;
|
||||
_instanceUniqueName = instanceUniqueName;
|
||||
_correlationId = correlationId;
|
||||
_centralCommunicationActor = centralCommunicationActor;
|
||||
_onEvent = onEvent;
|
||||
_onTerminated = onTerminated;
|
||||
|
||||
// Initial snapshot response from the site
|
||||
Receive<DebugViewSnapshot>(snapshot =>
|
||||
{
|
||||
_log.Info("Received initial snapshot for {0} ({1} attrs, {2} alarms)",
|
||||
_instanceUniqueName, snapshot.AttributeValues.Count, snapshot.AlarmStates.Count);
|
||||
_onEvent(snapshot);
|
||||
});
|
||||
|
||||
// Ongoing stream events from the site's InstanceActor
|
||||
Receive<AttributeValueChanged>(changed => _onEvent(changed));
|
||||
Receive<AlarmStateChanged>(changed => _onEvent(changed));
|
||||
|
||||
// Consumer requests stop
|
||||
Receive<StopDebugStream>(_ =>
|
||||
{
|
||||
_log.Info("Stopping debug stream for {0}", _instanceUniqueName);
|
||||
SendUnsubscribe();
|
||||
Context.Stop(Self);
|
||||
});
|
||||
|
||||
// Site disconnected — CentralCommunicationActor notifies us
|
||||
Receive<DebugStreamTerminated>(msg =>
|
||||
{
|
||||
_log.Warning("Debug stream terminated for {0} (site {1} disconnected)", _instanceUniqueName, msg.SiteId);
|
||||
_onTerminated();
|
||||
Context.Stop(Self);
|
||||
});
|
||||
|
||||
// Orphan safety net — if nobody stops us within 5 minutes, self-terminate
|
||||
Context.SetReceiveTimeout(TimeSpan.FromMinutes(5));
|
||||
Receive<ReceiveTimeout>(_ =>
|
||||
{
|
||||
_log.Warning("Debug stream for {0} timed out (orphaned session), stopping", _instanceUniqueName);
|
||||
SendUnsubscribe();
|
||||
_onTerminated();
|
||||
Context.Stop(Self);
|
||||
});
|
||||
}
|
||||
|
||||
protected override void PreStart()
|
||||
{
|
||||
_log.Info("Starting debug stream bridge for {0} on site {1}", _instanceUniqueName, _siteIdentifier);
|
||||
|
||||
// Send subscribe request via CentralCommunicationActor.
|
||||
// THIS actor is the Sender, so the site's InstanceActor registers us as the subscriber.
|
||||
var request = new SubscribeDebugViewRequest(_instanceUniqueName, _correlationId);
|
||||
var envelope = new SiteEnvelope(_siteIdentifier, request);
|
||||
_centralCommunicationActor.Tell(envelope, Self);
|
||||
}
|
||||
|
||||
private void SendUnsubscribe()
|
||||
{
|
||||
var request = new UnsubscribeDebugViewRequest(_instanceUniqueName, _correlationId);
|
||||
var envelope = new SiteEnvelope(_siteIdentifier, request);
|
||||
_centralCommunicationActor.Tell(envelope, Self);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Message sent to a DebugStreamBridgeActor to stop the debug stream session.
|
||||
/// </summary>
|
||||
public record StopDebugStream;
|
||||
@@ -48,12 +48,17 @@ public class CommunicationService
|
||||
GetActor().Tell(new RefreshSiteAddresses());
|
||||
}
|
||||
|
||||
private IActorRef GetActor()
|
||||
/// <summary>
|
||||
/// Gets the central communication actor reference. Throws if not yet initialized.
|
||||
/// </summary>
|
||||
public IActorRef GetCommunicationActor()
|
||||
{
|
||||
return _centralCommunicationActor
|
||||
?? throw new InvalidOperationException("CommunicationService not initialized. CentralCommunicationActor not set.");
|
||||
}
|
||||
|
||||
private IActorRef GetActor() => GetCommunicationActor();
|
||||
|
||||
// ── Pattern 1: Instance Deployment ──
|
||||
|
||||
public async Task<DeploymentStatusResponse> DeployInstanceAsync(
|
||||
|
||||
146
src/ScadaLink.Communication/DebugStreamService.cs
Normal file
146
src/ScadaLink.Communication/DebugStreamService.cs
Normal file
@@ -0,0 +1,146 @@
|
||||
using System.Collections.Concurrent;
|
||||
using Akka.Actor;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using ScadaLink.Commons.Interfaces.Repositories;
|
||||
using ScadaLink.Commons.Messages.DebugView;
|
||||
using ScadaLink.Communication.Actors;
|
||||
|
||||
namespace ScadaLink.Communication;
|
||||
|
||||
/// <summary>
|
||||
/// Manages debug stream sessions by creating DebugStreamBridgeActors that persist
|
||||
/// as subscribers on the site side. Both the Blazor debug view and the SignalR hub
|
||||
/// use this service to start/stop streams.
|
||||
/// </summary>
|
||||
public class DebugStreamService
|
||||
{
|
||||
private readonly CommunicationService _communicationService;
|
||||
private readonly IServiceProvider _serviceProvider;
|
||||
private readonly ILogger<DebugStreamService> _logger;
|
||||
private readonly ConcurrentDictionary<string, IActorRef> _sessions = new();
|
||||
private ActorSystem? _actorSystem;
|
||||
|
||||
public DebugStreamService(
|
||||
CommunicationService communicationService,
|
||||
IServiceProvider serviceProvider,
|
||||
ILogger<DebugStreamService> logger)
|
||||
{
|
||||
_communicationService = communicationService;
|
||||
_serviceProvider = serviceProvider;
|
||||
_logger = logger;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Sets the ActorSystem reference. Called during actor system startup (from AkkaHostedService).
|
||||
/// </summary>
|
||||
public void SetActorSystem(ActorSystem actorSystem)
|
||||
{
|
||||
_actorSystem = actorSystem;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Starts a debug stream session. Returns the initial snapshot.
|
||||
/// Ongoing events are delivered via the onEvent callback.
|
||||
/// The onTerminated callback fires if the stream is killed (site disconnect, timeout).
|
||||
/// </summary>
|
||||
public async Task<DebugStreamSession> StartStreamAsync(
|
||||
int instanceId,
|
||||
Action<object> onEvent,
|
||||
Action onTerminated,
|
||||
CancellationToken ct = default)
|
||||
{
|
||||
var system = _actorSystem
|
||||
?? throw new InvalidOperationException("DebugStreamService not initialized. ActorSystem not set.");
|
||||
|
||||
// Resolve instance → unique name + site
|
||||
string instanceUniqueName;
|
||||
string siteIdentifier;
|
||||
|
||||
using (var scope = _serviceProvider.CreateScope())
|
||||
{
|
||||
var instanceRepo = scope.ServiceProvider.GetRequiredService<ITemplateEngineRepository>();
|
||||
var instance = await instanceRepo.GetInstanceByIdAsync(instanceId)
|
||||
?? throw new InvalidOperationException($"Instance {instanceId} not found.");
|
||||
|
||||
var siteRepo = scope.ServiceProvider.GetRequiredService<ISiteRepository>();
|
||||
var site = await siteRepo.GetSiteByIdAsync(instance.SiteId)
|
||||
?? throw new InvalidOperationException($"Site {instance.SiteId} not found.");
|
||||
|
||||
instanceUniqueName = instance.UniqueName;
|
||||
siteIdentifier = site.SiteIdentifier;
|
||||
}
|
||||
|
||||
var sessionId = Guid.NewGuid().ToString("N");
|
||||
|
||||
// Capture the initial snapshot via a TaskCompletionSource
|
||||
var snapshotTcs = new TaskCompletionSource<DebugViewSnapshot>(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
|
||||
Action<object> onEventWrapper = evt =>
|
||||
{
|
||||
if (evt is DebugViewSnapshot snapshot && !snapshotTcs.Task.IsCompleted)
|
||||
{
|
||||
snapshotTcs.TrySetResult(snapshot);
|
||||
}
|
||||
else
|
||||
{
|
||||
onEvent(evt);
|
||||
}
|
||||
};
|
||||
|
||||
Action onTerminatedWrapper = () =>
|
||||
{
|
||||
_sessions.TryRemove(sessionId, out _);
|
||||
snapshotTcs.TrySetException(new InvalidOperationException("Debug stream terminated before snapshot received."));
|
||||
onTerminated();
|
||||
};
|
||||
|
||||
// Create the bridge actor — use type-based Props to avoid expression tree limitations with closures
|
||||
var commActor = _communicationService.GetCommunicationActor();
|
||||
|
||||
var props = Props.Create(typeof(DebugStreamBridgeActor),
|
||||
siteIdentifier,
|
||||
instanceUniqueName,
|
||||
sessionId,
|
||||
commActor,
|
||||
onEventWrapper,
|
||||
onTerminatedWrapper);
|
||||
|
||||
var bridgeActor = system.ActorOf(props, $"debug-stream-{sessionId}");
|
||||
|
||||
_sessions[sessionId] = bridgeActor;
|
||||
|
||||
// Wait for the initial snapshot (with timeout)
|
||||
using var timeoutCts = CancellationTokenSource.CreateLinkedTokenSource(ct);
|
||||
timeoutCts.CancelAfter(TimeSpan.FromSeconds(30));
|
||||
|
||||
try
|
||||
{
|
||||
var snapshot = await snapshotTcs.Task.WaitAsync(timeoutCts.Token);
|
||||
|
||||
_logger.LogInformation("Debug stream {SessionId} started for {Instance} on site {Site}",
|
||||
sessionId, instanceUniqueName, siteIdentifier);
|
||||
|
||||
return new DebugStreamSession(sessionId, snapshot);
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
StopStream(sessionId);
|
||||
throw new TimeoutException($"Timed out waiting for debug snapshot from {instanceUniqueName} on site {siteIdentifier}.");
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Stops an active debug stream session.
|
||||
/// </summary>
|
||||
public void StopStream(string sessionId)
|
||||
{
|
||||
if (_sessions.TryRemove(sessionId, out var bridgeActor))
|
||||
{
|
||||
bridgeActor.Tell(new StopDebugStream());
|
||||
_logger.LogInformation("Debug stream {SessionId} stopped", sessionId);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public record DebugStreamSession(string SessionId, DebugViewSnapshot InitialSnapshot);
|
||||
@@ -10,6 +10,7 @@ public static class ServiceCollectionExtensions
|
||||
.BindConfiguration("Communication");
|
||||
|
||||
services.AddSingleton<CommunicationService>();
|
||||
services.AddSingleton<DebugStreamService>();
|
||||
|
||||
return services;
|
||||
}
|
||||
|
||||
@@ -185,6 +185,11 @@ akka {{
|
||||
var commService = _serviceProvider.GetService<CommunicationService>();
|
||||
commService?.SetCommunicationActor(centralCommActor);
|
||||
|
||||
// Wire up the DebugStreamService with the ActorSystem
|
||||
var debugStreamService = _serviceProvider.GetService<DebugStreamService>();
|
||||
debugStreamService?.SetActorSystem(_actorSystem!);
|
||||
|
||||
|
||||
// Management Service — accessible via ClusterClient
|
||||
var mgmtLogger = _serviceProvider.GetRequiredService<ILoggerFactory>()
|
||||
.CreateLogger<ScadaLink.ManagementService.ManagementActor>();
|
||||
|
||||
@@ -138,6 +138,7 @@ try
|
||||
app.MapCentralUI<ScadaLink.Host.Components.App>();
|
||||
app.MapInboundAPI();
|
||||
app.MapManagementAPI();
|
||||
app.MapHub<ScadaLink.ManagementService.DebugStreamHub>("/hubs/debug-stream");
|
||||
|
||||
// Compile and register all Inbound API method scripts at startup
|
||||
using (var scope = app.Services.CreateScope())
|
||||
|
||||
164
src/ScadaLink.ManagementService/DebugStreamHub.cs
Normal file
164
src/ScadaLink.ManagementService/DebugStreamHub.cs
Normal file
@@ -0,0 +1,164 @@
|
||||
using System.Text;
|
||||
using Microsoft.AspNetCore.SignalR;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using ScadaLink.Commons.Messages.DebugView;
|
||||
using ScadaLink.Commons.Messages.Streaming;
|
||||
using ScadaLink.Communication;
|
||||
using ScadaLink.Security;
|
||||
|
||||
namespace ScadaLink.ManagementService;
|
||||
|
||||
/// <summary>
|
||||
/// SignalR hub for real-time debug stream subscriptions.
|
||||
/// External consumers (CLI) connect via WebSocket, authenticate with Basic Auth,
|
||||
/// and receive streaming attribute value and alarm state changes.
|
||||
/// </summary>
|
||||
public class DebugStreamHub : Hub
|
||||
{
|
||||
private const string SessionIdKey = "DebugStreamSessionId";
|
||||
|
||||
private readonly DebugStreamService _debugStreamService;
|
||||
private readonly ILogger<DebugStreamHub> _logger;
|
||||
|
||||
public DebugStreamHub(DebugStreamService debugStreamService, ILogger<DebugStreamHub> logger)
|
||||
{
|
||||
_debugStreamService = debugStreamService;
|
||||
_logger = logger;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Authenticates the connection using Basic Auth from the HTTP negotiate request.
|
||||
/// Validates credentials via LDAP and checks for the Deployment role.
|
||||
/// </summary>
|
||||
public override async Task OnConnectedAsync()
|
||||
{
|
||||
var httpContext = Context.GetHttpContext();
|
||||
if (httpContext == null)
|
||||
{
|
||||
_logger.LogWarning("DebugStreamHub connection rejected: no HTTP context");
|
||||
Context.Abort();
|
||||
return;
|
||||
}
|
||||
|
||||
// Extract Basic Auth credentials
|
||||
var authHeader = httpContext.Request.Headers.Authorization.ToString();
|
||||
if (string.IsNullOrEmpty(authHeader) || !authHeader.StartsWith("Basic ", StringComparison.OrdinalIgnoreCase))
|
||||
{
|
||||
_logger.LogWarning("DebugStreamHub connection rejected: missing Basic Auth header");
|
||||
Context.Abort();
|
||||
return;
|
||||
}
|
||||
|
||||
string username, password;
|
||||
try
|
||||
{
|
||||
var decoded = Encoding.UTF8.GetString(Convert.FromBase64String(authHeader["Basic ".Length..]));
|
||||
var colon = decoded.IndexOf(':');
|
||||
if (colon < 0) throw new FormatException();
|
||||
username = decoded[..colon];
|
||||
password = decoded[(colon + 1)..];
|
||||
}
|
||||
catch
|
||||
{
|
||||
_logger.LogWarning("DebugStreamHub connection rejected: malformed Basic Auth");
|
||||
Context.Abort();
|
||||
return;
|
||||
}
|
||||
|
||||
// LDAP authentication
|
||||
var ldapAuth = httpContext.RequestServices.GetRequiredService<LdapAuthService>();
|
||||
var authResult = await ldapAuth.AuthenticateAsync(username, password);
|
||||
if (!authResult.Success)
|
||||
{
|
||||
_logger.LogWarning("DebugStreamHub connection rejected: LDAP auth failed for {Username}", username);
|
||||
Context.Abort();
|
||||
return;
|
||||
}
|
||||
|
||||
// Role check — Deployment role required
|
||||
var roleMapper = httpContext.RequestServices.GetRequiredService<RoleMapper>();
|
||||
var mappingResult = await roleMapper.MapGroupsToRolesAsync(
|
||||
authResult.Groups ?? (IReadOnlyList<string>)Array.Empty<string>());
|
||||
|
||||
if (!mappingResult.Roles.Contains("Deployment"))
|
||||
{
|
||||
_logger.LogWarning("DebugStreamHub connection rejected: {Username} lacks Deployment role", username);
|
||||
Context.Abort();
|
||||
return;
|
||||
}
|
||||
|
||||
_logger.LogInformation("DebugStreamHub connection established for {Username}", username);
|
||||
await base.OnConnectedAsync();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Subscribes to a debug stream for the specified instance.
|
||||
/// Sends the initial snapshot immediately, then streams incremental changes.
|
||||
/// </summary>
|
||||
public async Task SubscribeInstance(int instanceId)
|
||||
{
|
||||
// Stop any existing subscription for this connection
|
||||
await UnsubscribeInstance();
|
||||
|
||||
var connectionId = Context.ConnectionId;
|
||||
|
||||
try
|
||||
{
|
||||
var session = await _debugStreamService.StartStreamAsync(
|
||||
instanceId,
|
||||
onEvent: evt =>
|
||||
{
|
||||
// Fire-and-forget — if the client disconnects, SendAsync will fail silently
|
||||
_ = evt switch
|
||||
{
|
||||
AttributeValueChanged changed =>
|
||||
Clients.Client(connectionId).SendAsync("OnAttributeChanged", changed),
|
||||
AlarmStateChanged changed =>
|
||||
Clients.Client(connectionId).SendAsync("OnAlarmChanged", changed),
|
||||
DebugViewSnapshot snapshot =>
|
||||
Clients.Client(connectionId).SendAsync("OnSnapshot", snapshot),
|
||||
_ => Task.CompletedTask
|
||||
};
|
||||
},
|
||||
onTerminated: () =>
|
||||
{
|
||||
_ = Clients.Client(connectionId).SendAsync("OnStreamTerminated", "Site disconnected");
|
||||
});
|
||||
|
||||
Context.Items[SessionIdKey] = session.SessionId;
|
||||
|
||||
// Send the initial snapshot
|
||||
await Clients.Caller.SendAsync("OnSnapshot", session.InitialSnapshot);
|
||||
|
||||
_logger.LogInformation("DebugStreamHub: {ConnectionId} subscribed to instance {InstanceId}",
|
||||
connectionId, instanceId);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "DebugStreamHub: Failed to subscribe {ConnectionId} to instance {InstanceId}",
|
||||
connectionId, instanceId);
|
||||
await Clients.Caller.SendAsync("OnStreamTerminated", ex.Message);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Unsubscribes from the current debug stream.
|
||||
/// </summary>
|
||||
public Task UnsubscribeInstance()
|
||||
{
|
||||
if (Context.Items.TryGetValue(SessionIdKey, out var sessionIdObj) && sessionIdObj is string sessionId)
|
||||
{
|
||||
_debugStreamService.StopStream(sessionId);
|
||||
Context.Items.Remove(SessionIdKey);
|
||||
_logger.LogInformation("DebugStreamHub: {ConnectionId} unsubscribed", Context.ConnectionId);
|
||||
}
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public override async Task OnDisconnectedAsync(Exception? exception)
|
||||
{
|
||||
await UnsubscribeInstance();
|
||||
await base.OnDisconnectedAsync(exception);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user