feat: add SiteStreamGrpcServer with Channel<T> bridge and stream limits
- Define ISiteStreamSubscriber interface for decoupling from SiteRuntime - Implement SiteStreamGrpcServer (inherits SiteStreamServiceBase) with: - Readiness gate (SetReady) - Max concurrent stream enforcement - Duplicate correlationId replacement (cancels previous stream) - StreamRelayActor creation per subscription - Bounded Channel<SiteStreamEvent> bridge (1000 capacity, drop-oldest) - Clean teardown: unsubscribe, stop actor, remove tracking entry - Identity-safe cleanup using ConcurrentDictionary.TryRemove(KeyValuePair) to prevent replacement streams from being removed by predecessor cleanup - 7 unit tests covering reject-not-ready, max-streams, duplicate cancel, cleanup-on-cancel, subscribe/remove lifecycle, event forwarding
This commit is contained in:
Joseph Doherty
@@ -0,0 +1,237 @@
|
||||
using System.Threading.Channels;
|
||||
using Akka.Actor;
|
||||
using Akka.TestKit.Xunit2;
|
||||
using Grpc.Core;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using NSubstitute;
|
||||
using ScadaLink.Communication.Grpc;
|
||||
|
||||
namespace ScadaLink.Communication.Tests.Grpc;
|
||||
|
||||
public class SiteStreamGrpcServerTests : TestKit
|
||||
{
|
||||
private readonly ISiteStreamSubscriber _subscriber;
|
||||
private readonly ILogger<SiteStreamGrpcServer> _logger;
|
||||
|
||||
public SiteStreamGrpcServerTests()
|
||||
{
|
||||
_subscriber = Substitute.For<ISiteStreamSubscriber>();
|
||||
_subscriber.Subscribe(Arg.Any<string>(), Arg.Any<IActorRef>())
|
||||
.Returns("sub-1");
|
||||
_logger = NullLogger<SiteStreamGrpcServer>.Instance;
|
||||
}
|
||||
|
||||
private SiteStreamGrpcServer CreateServer(int maxStreams = 100)
|
||||
{
|
||||
return new SiteStreamGrpcServer(_subscriber, Sys, _logger, maxStreams);
|
||||
}
|
||||
|
||||
private static InstanceStreamRequest MakeRequest(string correlationId = "corr-1", string instance = "Site1.Pump01")
|
||||
{
|
||||
return new InstanceStreamRequest
|
||||
{
|
||||
CorrelationId = correlationId,
|
||||
InstanceUniqueName = instance
|
||||
};
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task RejectsWhenNotReady()
|
||||
{
|
||||
var server = CreateServer();
|
||||
// Do NOT call SetReady()
|
||||
|
||||
var writer = Substitute.For<IServerStreamWriter<SiteStreamEvent>>();
|
||||
var context = CreateMockContext();
|
||||
|
||||
var ex = await Assert.ThrowsAsync<RpcException>(
|
||||
() => server.SubscribeInstance(MakeRequest(), writer, context));
|
||||
|
||||
Assert.Equal(StatusCode.Unavailable, ex.StatusCode);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task RejectsWhenMaxStreamsReached()
|
||||
{
|
||||
var server = CreateServer(maxStreams: 1);
|
||||
server.SetReady();
|
||||
|
||||
// Start one stream that blocks
|
||||
var cts1 = new CancellationTokenSource();
|
||||
var context1 = CreateMockContext(cts1.Token);
|
||||
var writer1 = Substitute.For<IServerStreamWriter<SiteStreamEvent>>();
|
||||
|
||||
var stream1Task = Task.Run(() => server.SubscribeInstance(
|
||||
MakeRequest("corr-1"), writer1, context1));
|
||||
|
||||
// Wait for the first stream to register
|
||||
await WaitForConditionAsync(() => server.ActiveStreamCount == 1);
|
||||
|
||||
// Second stream should be rejected
|
||||
var writer2 = Substitute.For<IServerStreamWriter<SiteStreamEvent>>();
|
||||
var context2 = CreateMockContext();
|
||||
|
||||
var ex = await Assert.ThrowsAsync<RpcException>(
|
||||
() => server.SubscribeInstance(MakeRequest("corr-2"), writer2, context2));
|
||||
|
||||
Assert.Equal(StatusCode.ResourceExhausted, ex.StatusCode);
|
||||
|
||||
// Clean up first stream
|
||||
cts1.Cancel();
|
||||
await stream1Task;
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task CancelsDuplicateCorrelationId()
|
||||
{
|
||||
var server = CreateServer();
|
||||
server.SetReady();
|
||||
|
||||
var cts1 = new CancellationTokenSource();
|
||||
var context1 = CreateMockContext(cts1.Token);
|
||||
var writer1 = Substitute.For<IServerStreamWriter<SiteStreamEvent>>();
|
||||
|
||||
// Start first stream
|
||||
var stream1Task = Task.Run(() => server.SubscribeInstance(
|
||||
MakeRequest("corr-dup"), writer1, context1));
|
||||
|
||||
await WaitForConditionAsync(() => server.ActiveStreamCount == 1);
|
||||
|
||||
// Start second stream with same correlationId -- should cancel first
|
||||
var cts2 = new CancellationTokenSource();
|
||||
var context2 = CreateMockContext(cts2.Token);
|
||||
var writer2 = Substitute.For<IServerStreamWriter<SiteStreamEvent>>();
|
||||
|
||||
var stream2Task = Task.Run(() => server.SubscribeInstance(
|
||||
MakeRequest("corr-dup"), writer2, context2));
|
||||
|
||||
// First stream should complete (cancelled by duplicate replacement)
|
||||
await stream1Task;
|
||||
|
||||
// Second stream should be active
|
||||
await WaitForConditionAsync(() => server.ActiveStreamCount == 1);
|
||||
|
||||
// Clean up
|
||||
cts2.Cancel();
|
||||
await stream2Task;
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task CleansUpOnCancellation()
|
||||
{
|
||||
var server = CreateServer();
|
||||
server.SetReady();
|
||||
|
||||
var cts = new CancellationTokenSource();
|
||||
var context = CreateMockContext(cts.Token);
|
||||
var writer = Substitute.For<IServerStreamWriter<SiteStreamEvent>>();
|
||||
|
||||
var streamTask = Task.Run(() => server.SubscribeInstance(
|
||||
MakeRequest("corr-cleanup"), writer, context));
|
||||
|
||||
await WaitForConditionAsync(() => server.ActiveStreamCount == 1);
|
||||
|
||||
cts.Cancel();
|
||||
await streamTask;
|
||||
|
||||
Assert.Equal(0, server.ActiveStreamCount);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task SubscribesAndRemovesFromStreamManager()
|
||||
{
|
||||
var server = CreateServer();
|
||||
server.SetReady();
|
||||
|
||||
var cts = new CancellationTokenSource();
|
||||
var context = CreateMockContext(cts.Token);
|
||||
var writer = Substitute.For<IServerStreamWriter<SiteStreamEvent>>();
|
||||
|
||||
var streamTask = Task.Run(() => server.SubscribeInstance(
|
||||
MakeRequest("corr-sub", "Site1.Motor01"), writer, context));
|
||||
|
||||
await WaitForConditionAsync(() => server.ActiveStreamCount == 1);
|
||||
|
||||
// Verify Subscribe was called
|
||||
_subscriber.Received(1).Subscribe("Site1.Motor01", Arg.Any<IActorRef>());
|
||||
|
||||
cts.Cancel();
|
||||
await streamTask;
|
||||
|
||||
// Verify RemoveSubscriber was called
|
||||
_subscriber.Received(1).RemoveSubscriber(Arg.Any<IActorRef>());
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task WritesEventsToResponseStream()
|
||||
{
|
||||
var server = CreateServer();
|
||||
server.SetReady();
|
||||
|
||||
// Capture the relay actor so we can send it events
|
||||
IActorRef? capturedActor = null;
|
||||
_subscriber.Subscribe(Arg.Any<string>(), Arg.Any<IActorRef>())
|
||||
.Returns(ci =>
|
||||
{
|
||||
capturedActor = ci.Arg<IActorRef>();
|
||||
return "sub-write";
|
||||
});
|
||||
|
||||
var cts = new CancellationTokenSource();
|
||||
var context = CreateMockContext(cts.Token);
|
||||
var writer = Substitute.For<IServerStreamWriter<SiteStreamEvent>>();
|
||||
var writtenEvents = new List<SiteStreamEvent>();
|
||||
writer.WriteAsync(Arg.Any<SiteStreamEvent>(), Arg.Any<CancellationToken>())
|
||||
.Returns(Task.CompletedTask)
|
||||
.AndDoes(ci => writtenEvents.Add(ci.Arg<SiteStreamEvent>()));
|
||||
|
||||
var streamTask = Task.Run(() => server.SubscribeInstance(
|
||||
MakeRequest("corr-write", "Site1.Pump01"), writer, context));
|
||||
|
||||
await WaitForConditionAsync(() => capturedActor != null);
|
||||
|
||||
// Send a domain event to the relay actor
|
||||
var ts = DateTimeOffset.UtcNow;
|
||||
capturedActor!.Tell(new Commons.Messages.Streaming.AttributeValueChanged(
|
||||
"Site1.Pump01", "Path", "Attr", 99.5, "Good", ts));
|
||||
|
||||
// Wait for event to be written
|
||||
await WaitForConditionAsync(() => writtenEvents.Count >= 1);
|
||||
|
||||
Assert.Single(writtenEvents);
|
||||
Assert.Equal("corr-write", writtenEvents[0].CorrelationId);
|
||||
Assert.Equal(SiteStreamEvent.EventOneofCase.AttributeChanged, writtenEvents[0].EventCase);
|
||||
|
||||
cts.Cancel();
|
||||
await streamTask;
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void SetReady_AllowsStreamCreation()
|
||||
{
|
||||
var server = CreateServer();
|
||||
// Initially not ready -- just verify the property works
|
||||
server.SetReady();
|
||||
// No assertion needed -- the other tests verify that SetReady enables streaming
|
||||
Assert.Equal(0, server.ActiveStreamCount);
|
||||
}
|
||||
|
||||
private static ServerCallContext CreateMockContext(CancellationToken cancellationToken = default)
|
||||
{
|
||||
var context = Substitute.For<ServerCallContext>();
|
||||
context.CancellationToken.Returns(cancellationToken);
|
||||
return context;
|
||||
}
|
||||
|
||||
private static async Task WaitForConditionAsync(Func<bool> condition, int timeoutMs = 5000)
|
||||
{
|
||||
var deadline = DateTime.UtcNow.AddMilliseconds(timeoutMs);
|
||||
while (!condition() && DateTime.UtcNow < deadline)
|
||||
{
|
||||
await Task.Delay(25);
|
||||
}
|
||||
|
||||
Assert.True(condition(), $"Condition not met within {timeoutMs}ms");
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user