Merge ServiceLevel replay fixes for freshness integration
This commit is contained in:
Joseph Doherty
+136
-5
@@ -10,10 +10,10 @@ namespace ZB.MOM.WW.OtOpcUa.Commons.Tests.OpcUa;
|
||||
public class DeferredServiceLevelPublisherTests
|
||||
{
|
||||
[Fact]
|
||||
public void Before_SetInner_Publish_is_a_noop()
|
||||
public void Before_SetInner_Publish_accepts_a_cached_request()
|
||||
{
|
||||
var publisher = new DeferredServiceLevelPublisher();
|
||||
// Must not throw; the NullServiceLevelPublisher absorbs it.
|
||||
// Accept the request before an SDK publisher exists.
|
||||
publisher.Publish(200);
|
||||
}
|
||||
|
||||
@@ -30,7 +30,7 @@ public class DeferredServiceLevelPublisherTests
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void SetInner_null_reverts_to_null_publisher()
|
||||
public void SetInner_null_detaches_the_previous_publisher()
|
||||
{
|
||||
var spy = new SpyPublisher();
|
||||
var publisher = new DeferredServiceLevelPublisher();
|
||||
@@ -39,8 +39,139 @@ public class DeferredServiceLevelPublisherTests
|
||||
|
||||
publisher.Publish(100);
|
||||
|
||||
// The spy should NOT have received the publish after revert.
|
||||
spy.LastPublished.ShouldBe((byte)0, "spy should not have been updated after revert to null");
|
||||
// The previous publisher must not receive an unbound request.
|
||||
spy.LastPublished.ShouldBe((byte)0, "the previous publisher should not be updated after unbinding");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Binding_without_a_request_does_not_invent_a_level()
|
||||
{
|
||||
var received = new List<byte>();
|
||||
new DeferredServiceLevelPublisher().SetInner(new CallbackPublisher(received.Add));
|
||||
received.ShouldBeEmpty();
|
||||
}
|
||||
|
||||
[Theory]
|
||||
[InlineData((byte)100)]
|
||||
[InlineData((byte)0)]
|
||||
public void Binding_replays_the_latest_request_including_zero(byte latest)
|
||||
{
|
||||
var publisher = new DeferredServiceLevelPublisher();
|
||||
publisher.Publish(240);
|
||||
publisher.Publish(latest);
|
||||
var received = new List<byte>();
|
||||
publisher.SetInner(new CallbackPublisher(received.Add));
|
||||
received.ShouldBe(new[] { latest });
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Unbinding_retains_latest_for_rebinding()
|
||||
{
|
||||
var publisher = new DeferredServiceLevelPublisher();
|
||||
var old = new List<byte>();
|
||||
publisher.SetInner(new CallbackPublisher(old.Add));
|
||||
publisher.Publish(240);
|
||||
publisher.SetInner(null);
|
||||
publisher.Publish(0);
|
||||
var replacement = new List<byte>();
|
||||
publisher.SetInner(new CallbackPublisher(replacement.Add));
|
||||
old.ShouldBe(new byte[] { 240 });
|
||||
replacement.ShouldBe(new byte[] { 0 });
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Failed_publish_retains_the_requested_value_for_replay()
|
||||
{
|
||||
var publisher = new DeferredServiceLevelPublisher();
|
||||
publisher.SetInner(new CallbackPublisher(_ => throw new InvalidOperationException()));
|
||||
Should.Throw<InvalidOperationException>(() => publisher.Publish(0));
|
||||
var received = new List<byte>();
|
||||
publisher.SetInner(new CallbackPublisher(received.Add));
|
||||
received.ShouldBe(new byte[] { 0 });
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Failed_replay_keeps_candidate_bound_for_later_publish()
|
||||
{
|
||||
var publisher = new DeferredServiceLevelPublisher();
|
||||
publisher.Publish(100);
|
||||
var fail = true;
|
||||
var received = new List<byte>();
|
||||
var candidate = new CallbackPublisher(value =>
|
||||
{
|
||||
if (fail) throw new InvalidOperationException();
|
||||
received.Add(value);
|
||||
});
|
||||
Should.Throw<InvalidOperationException>(() => publisher.SetInner(candidate));
|
||||
fail = false;
|
||||
publisher.Publish(0);
|
||||
publisher.SetInner(null);
|
||||
publisher.SetInner(candidate);
|
||||
received.ShouldBe(new byte[] { 0, 0 });
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Failed_replay_retains_latest_for_a_replacement_publisher()
|
||||
{
|
||||
var publisher = new DeferredServiceLevelPublisher();
|
||||
publisher.Publish(100);
|
||||
Should.Throw<InvalidOperationException>(() => publisher.SetInner(
|
||||
new CallbackPublisher(_ => throw new InvalidOperationException())));
|
||||
var received = new List<byte>();
|
||||
publisher.SetInner(new CallbackPublisher(received.Add));
|
||||
received.ShouldBe(new byte[] { 100 });
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Binding_waits_for_inflight_publish_then_replays_latest()
|
||||
{
|
||||
var publisher = new DeferredServiceLevelPublisher();
|
||||
using var entered = new ManualResetEventSlim();
|
||||
using var release = new ManualResetEventSlim();
|
||||
publisher.SetInner(new CallbackPublisher(_ =>
|
||||
{
|
||||
entered.Set();
|
||||
release.Wait(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken).ShouldBeTrue();
|
||||
}));
|
||||
var publish = Task.Run(() => publisher.Publish(100), TestContext.Current.CancellationToken);
|
||||
var received = new List<byte>();
|
||||
using var bindingStarted = new ManualResetEventSlim();
|
||||
Exception? bindingError = null;
|
||||
var binding = new Thread(() =>
|
||||
{
|
||||
bindingStarted.Set();
|
||||
try { publisher.SetInner(new CallbackPublisher(received.Add)); }
|
||||
catch (Exception ex) { bindingError = ex; }
|
||||
});
|
||||
var bindingWasStarted = false;
|
||||
try
|
||||
{
|
||||
entered.Wait(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken).ShouldBeTrue();
|
||||
binding.Start();
|
||||
bindingWasStarted = true;
|
||||
bindingStarted.Wait(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken).ShouldBeTrue();
|
||||
// Past the marker, SetInner is the thread's only potentially blocking operation.
|
||||
// Observe actual contention rather than assuming Task.Run has entered the adapter.
|
||||
SpinWait.SpinUntil(() => !binding.IsAlive ||
|
||||
(binding.ThreadState & System.Threading.ThreadState.WaitSleepJoin) != 0,
|
||||
TimeSpan.FromSeconds(3)).ShouldBeTrue();
|
||||
(binding.ThreadState & System.Threading.ThreadState.WaitSleepJoin).ShouldNotBe(
|
||||
(System.Threading.ThreadState)0, "binding must wait for the in-flight delivery");
|
||||
}
|
||||
finally
|
||||
{
|
||||
release.Set();
|
||||
if (bindingWasStarted) binding.Join(TimeSpan.FromSeconds(5)).ShouldBeTrue();
|
||||
await publish.WaitAsync(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken);
|
||||
}
|
||||
bindingError.ShouldBeNull();
|
||||
publisher.Publish(0);
|
||||
received.ShouldBe(new byte[] { 100, 0 });
|
||||
}
|
||||
|
||||
private sealed class CallbackPublisher(Action<byte> publish) : IServiceLevelPublisher
|
||||
{
|
||||
public void Publish(byte serviceLevel) => publish(serviceLevel);
|
||||
}
|
||||
|
||||
// ---- test double ----
|
||||
|
||||
+45
@@ -0,0 +1,45 @@
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Host.OpcUa;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Host.Tests.OpcUa;
|
||||
|
||||
public sealed class FailedServiceLevelBindingCleanupTests
|
||||
{
|
||||
[Fact]
|
||||
public async Task Successful_cleanup_observes_stop_before_disposal_and_releases_ownership()
|
||||
{
|
||||
var events = new List<string>();
|
||||
var released = await FailedServiceLevelBindingCleanup.StopAndDisposeAsync(
|
||||
() => { events.Add("stop"); return Task.CompletedTask; },
|
||||
() => { events.Add("dispose"); return ValueTask.CompletedTask; },
|
||||
NullLogger.Instance);
|
||||
events.ShouldBe(new[] { "stop", "dispose" });
|
||||
released.ShouldBeTrue();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Failed_stop_still_disposes_and_retains_ownership_for_retry()
|
||||
{
|
||||
var events = new List<string>();
|
||||
var released = await FailedServiceLevelBindingCleanup.StopAndDisposeAsync(
|
||||
() => { events.Add("stop"); throw new InvalidOperationException("controlled stop failure"); },
|
||||
() => { events.Add("dispose"); return ValueTask.CompletedTask; },
|
||||
NullLogger.Instance);
|
||||
events.ShouldBe(new[] { "stop", "dispose" });
|
||||
released.ShouldBeFalse();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Failed_disposal_retains_ownership_without_escaping_the_binding_failure_handler()
|
||||
{
|
||||
var events = new List<string>();
|
||||
var released = await FailedServiceLevelBindingCleanup.StopAndDisposeAsync(
|
||||
() => { events.Add("stop"); return Task.CompletedTask; },
|
||||
() => { events.Add("dispose"); throw new InvalidOperationException("controlled disposal failure"); },
|
||||
NullLogger.Instance);
|
||||
events.ShouldBe(new[] { "stop", "dispose" });
|
||||
released.ShouldBeFalse();
|
||||
}
|
||||
}
|
||||
+103
@@ -0,0 +1,103 @@
|
||||
using System.Net;
|
||||
using System.Net.Sockets;
|
||||
using Akka.Hosting;
|
||||
using Microsoft.Extensions.Configuration;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Hosting;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using Microsoft.Extensions.Options;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Authorization;
|
||||
using ZB.MOM.WW.OtOpcUa.Host.OpcUa;
|
||||
using ZB.MOM.WW.OtOpcUa.OpcUaServer;
|
||||
using ZB.MOM.WW.OtOpcUa.OpcUaServer.Security;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Host.Tests.OpcUa;
|
||||
|
||||
public sealed class OtOpcUaServerHostedServiceBindingFailureTests
|
||||
{
|
||||
[Fact]
|
||||
public async Task Binding_failure_closes_started_endpoint_and_allows_other_host_services_to_start()
|
||||
{
|
||||
using var deadline = new CancellationTokenSource(TimeSpan.FromSeconds(15));
|
||||
using var probe = new TcpListener(IPAddress.Loopback, 0);
|
||||
probe.Start();
|
||||
var port = ((IPEndPoint)probe.LocalEndpoint).Port;
|
||||
probe.Stop();
|
||||
var pki = Path.Combine(Path.GetTempPath(), $"otopcua-binding-{Guid.NewGuid():N}");
|
||||
var deferred = new DeferredServiceLevelPublisher();
|
||||
deferred.Publish(0);
|
||||
var bindingReached = false;
|
||||
var listenerObserved = false;
|
||||
OtOpcUaServerHostedService? service = null;
|
||||
var factory = new FaultInjectingLoggerFactory(() =>
|
||||
{
|
||||
using var connected = new TcpClient();
|
||||
connected.Connect(IPAddress.Loopback, port);
|
||||
listenerObserved = connected.Connected;
|
||||
bindingReached = true;
|
||||
throw new InvalidOperationException("controlled binding factory failure");
|
||||
});
|
||||
service = new OtOpcUaServerHostedService(
|
||||
Options.Create(new OpcUaApplicationHostOptions
|
||||
{
|
||||
ApplicationName = "OtOpcUa.BindingFailure",
|
||||
ApplicationUri = $"urn:OtOpcUa.BindingFailure:{Guid.NewGuid():N}",
|
||||
OpcUaPort = port, PublicHostname = "localhost", PkiStoreRoot = pki,
|
||||
}), new DeferredAddressSpaceSink(), deferred,
|
||||
NullOpcUaUserAuthenticator.Instance, NullHistorianDataSource.Instance,
|
||||
EmptyDataPlaneAuthoritySource.Instance, new ConfigurationBuilder().Build(),
|
||||
() => throw new NotSupportedException("no actor system required"), new ActorRegistry(),
|
||||
null, Options.Create(new AclOptions()), TimeProvider.System, factory);
|
||||
var admin = new AdminBootSentinel();
|
||||
using var host = new HostBuilder().ConfigureServices(services =>
|
||||
{
|
||||
services.AddSingleton<IHostedService>(service);
|
||||
services.AddSingleton<IHostedService>(admin);
|
||||
}).Build();
|
||||
try
|
||||
{
|
||||
await host.StartAsync(deadline.Token);
|
||||
listenerObserved.ShouldBeTrue("the SDK endpoint was actually listening before replay");
|
||||
bindingReached.ShouldBeTrue();
|
||||
admin.Started.ShouldBeTrue();
|
||||
using var client = new TcpClient();
|
||||
await Should.ThrowAsync<SocketException>(async () =>
|
||||
await client.ConnectAsync(IPAddress.Loopback, port, deadline.Token));
|
||||
// After construction failure, subsequent requests are accepted into the unbound cache.
|
||||
deferred.Publish(100);
|
||||
await host.StopAsync(deadline.Token);
|
||||
}
|
||||
finally
|
||||
{
|
||||
await service.DisposeAsync();
|
||||
if (Directory.Exists(pki)) Directory.Delete(pki, recursive: true);
|
||||
}
|
||||
}
|
||||
|
||||
private sealed class FaultInjectingLoggerFactory(Action installFault) : ILoggerFactory
|
||||
{
|
||||
public ILogger CreateLogger(string categoryName)
|
||||
{
|
||||
if (categoryName == typeof(SdkServiceLevelPublisher).FullName) installFault();
|
||||
return NullLogger.Instance;
|
||||
}
|
||||
public void AddProvider(ILoggerProvider provider) { }
|
||||
public void Dispose() { }
|
||||
}
|
||||
|
||||
private sealed class AdminBootSentinel : IHostedService
|
||||
{
|
||||
public bool Started { get; private set; }
|
||||
public Task StartAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
Started = true;
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
public Task StopAsync(CancellationToken cancellationToken) => Task.CompletedTask;
|
||||
}
|
||||
}
|
||||
+71
@@ -0,0 +1,71 @@
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using Opc.Ua;
|
||||
using Opc.Ua.Client;
|
||||
using Opc.Ua.Server;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests;
|
||||
|
||||
/// <summary>Local loopback evidence that late binding replaces the SDK default on the wire.</summary>
|
||||
public sealed class ServiceLevelBindingReplayTests
|
||||
{
|
||||
[Theory]
|
||||
[InlineData((byte)100)]
|
||||
[InlineData((byte)0)]
|
||||
public async Task Late_binding_replays_latest_request_to_standard_ServiceLevel_read(byte latest)
|
||||
{
|
||||
using var deadline = CancellationTokenSource.CreateLinkedTokenSource(TestContext.Current.CancellationToken);
|
||||
deadline.CancelAfter(TimeSpan.FromSeconds(15));
|
||||
var ct = deadline.Token;
|
||||
var pki = Path.Combine(Path.GetTempPath(), $"otopcua-replay-{Guid.NewGuid():N}");
|
||||
var clientPki = TestClientSecurity.AllocatePkiRoot();
|
||||
var port = 0;
|
||||
try
|
||||
{
|
||||
var deferred = new DeferredServiceLevelPublisher();
|
||||
deferred.Publish(240);
|
||||
deferred.Publish(latest);
|
||||
var started = await OpcUaHostStart.StartAsync(p =>
|
||||
{
|
||||
port = p;
|
||||
return new OpcUaApplicationHost(new OpcUaApplicationHostOptions
|
||||
{
|
||||
ApplicationName = "OtOpcUa.Replay",
|
||||
ApplicationUri = $"urn:OtOpcUa.Replay:{Path.GetFileName(pki)}",
|
||||
OpcUaPort = p, PublicHostname = "localhost", PkiStoreRoot = pki,
|
||||
}, NullLogger<OpcUaApplicationHost>.Instance);
|
||||
}, () => new StandardServer(), ct);
|
||||
await using var host = started.Host;
|
||||
deferred.SetInner(new SdkServiceLevelPublisher(started.Server.CurrentInstance,
|
||||
NullLogger<SdkServiceLevelPublisher>.Instance));
|
||||
var config = new ApplicationConfiguration
|
||||
{
|
||||
ApplicationName = "OtOpcUa.ReplayClient",
|
||||
ApplicationUri = $"urn:OtOpcUa.ReplayClient:{Guid.NewGuid():N}",
|
||||
ApplicationType = ApplicationType.Client,
|
||||
SecurityConfiguration = TestClientSecurity.Build(clientPki),
|
||||
ClientConfiguration = new ClientConfiguration { DefaultSessionTimeout = 10_000 },
|
||||
TransportQuotas = new TransportQuotas { OperationTimeout = 5_000 },
|
||||
};
|
||||
await config.ValidateAsync(ApplicationType.Client, ct);
|
||||
config.CertificateValidator.CertificateValidation += (_, e) => e.Accept = true;
|
||||
var endpoint = await CoreClientUtils.SelectEndpointAsync(config,
|
||||
$"opc.tcp://localhost:{port}/OtOpcUa", false, DefaultTelemetry.Create(_ => { }), ct);
|
||||
using var session = await new DefaultSessionFactory(DefaultTelemetry.Create(_ => { })).CreateAsync(
|
||||
config, new ConfiguredEndpoint(null, endpoint, EndpointConfiguration.Create(config)),
|
||||
updateBeforeConnect: false, checkDomain: false, sessionName: "ServiceLevelReplay",
|
||||
sessionTimeout: 10_000, identity: new UserIdentity(new AnonymousIdentityToken()),
|
||||
preferredLocales: null, ct: ct);
|
||||
var read = await session.ReadValueAsync(new Opc.Ua.NodeId(2267u), ct);
|
||||
read.StatusCode.ShouldBe(StatusCodes.Good);
|
||||
read.Value.ShouldBe(latest);
|
||||
}
|
||||
finally
|
||||
{
|
||||
if (Directory.Exists(pki)) Directory.Delete(pki, recursive: true);
|
||||
if (Directory.Exists(clientPki)) Directory.Delete(clientPki, recursive: true);
|
||||
}
|
||||
}
|
||||
}
|
||||
+4
-4
@@ -6,9 +6,9 @@ namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests;
|
||||
|
||||
public sealed class DeferredServiceLevelPublisherTests
|
||||
{
|
||||
/// <summary>Verifies that publish before SetInner is a safe no-op.</summary>
|
||||
/// <summary>Verifies that publish before SetInner safely accepts a cached request.</summary>
|
||||
[Fact]
|
||||
public void Publish_before_SetInner_is_a_safe_noop()
|
||||
public void Publish_before_SetInner_accepts_a_cached_request()
|
||||
{
|
||||
var deferred = new DeferredServiceLevelPublisher();
|
||||
|
||||
@@ -28,9 +28,9 @@ public sealed class DeferredServiceLevelPublisherTests
|
||||
recording.LastValue.ShouldBe((byte)200);
|
||||
}
|
||||
|
||||
/// <summary>Verifies that SetInner with null reverts to the null publisher.</summary>
|
||||
/// <summary>Verifies that SetInner with null detaches the previous publisher.</summary>
|
||||
[Fact]
|
||||
public void SetInner_null_reverts_to_Null_publisher()
|
||||
public void SetInner_null_detaches_the_previous_publisher()
|
||||
{
|
||||
var recording = new RecordingPublisher();
|
||||
var deferred = new DeferredServiceLevelPublisher();
|
||||
|
||||
@@ -19,6 +19,40 @@ public sealed class SdkServiceLevelPublisherTests : IDisposable
|
||||
Path.GetTempPath(),
|
||||
$"otopcua-pki-{Guid.NewGuid():N}");
|
||||
|
||||
[Fact]
|
||||
public void Missing_ServiceLevel_is_reported_as_failure()
|
||||
{
|
||||
var server = System.Reflection.DispatchProxy.Create<IServerInternal, FaultingServer>();
|
||||
var publisher = new SdkServiceLevelPublisher(server, NullLogger<SdkServiceLevelPublisher>.Instance);
|
||||
Should.Throw<InvalidOperationException>(() => publisher.Publish(0));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Sdk_write_failure_propagates_the_original_exception()
|
||||
{
|
||||
var server = System.Reflection.DispatchProxy.Create<IServerInternal, FaultingServer>();
|
||||
var proxy = (FaultingServer)(object)server;
|
||||
proxy.Node = new Opc.Ua.ServerObjectState(null)
|
||||
{
|
||||
ServiceLevel = new Opc.Ua.PropertyState<byte>(null),
|
||||
};
|
||||
proxy.Fault = new InvalidOperationException("controlled diagnostics failure");
|
||||
var publisher = new SdkServiceLevelPublisher(server, NullLogger<SdkServiceLevelPublisher>.Instance);
|
||||
Should.Throw<InvalidOperationException>(() => publisher.Publish(0)).ShouldBeSameAs(proxy.Fault);
|
||||
}
|
||||
|
||||
public class FaultingServer : System.Reflection.DispatchProxy
|
||||
{
|
||||
public Opc.Ua.ServerObjectState? Node { get; set; }
|
||||
public Exception? Fault { get; set; }
|
||||
protected override object? Invoke(System.Reflection.MethodInfo? method, object?[]? args)
|
||||
{
|
||||
if (method?.Name == "get_ServerObject") return Node;
|
||||
if (method?.Name == "get_DiagnosticsLock") throw Fault!;
|
||||
throw new NotSupportedException(method?.Name);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Verifies that the publisher writes values to the standard Server.ServiceLevel variable.</summary>
|
||||
[Fact]
|
||||
public async Task Publish_writes_value_to_Server_ServiceLevel_variable()
|
||||
|
||||
@@ -15,6 +15,49 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.OpcUa;
|
||||
|
||||
public sealed class OpcUaPublishActorTests : RuntimeActorTestBase
|
||||
{
|
||||
[Fact]
|
||||
public void Failed_value_is_retried_and_last_level_only_tracks_success()
|
||||
{
|
||||
var sink = new RecoveringPublisher { Fail = false };
|
||||
var actor = ActorOfAsTestActorRef<OpcUaPublishActor>(OpcUaPublishActor.PropsForTests(serviceLevel: sink));
|
||||
actor.Tell(new OpcUaPublishActor.ServiceLevelChanged(100));
|
||||
sink.Fail = true;
|
||||
actor.Tell(new OpcUaPublishActor.ServiceLevelChanged(0));
|
||||
actor.UnderlyingActor.LastServiceLevel.ShouldBe((byte)100);
|
||||
sink.Fail = false;
|
||||
actor.Tell(new OpcUaPublishActor.ServiceLevelChanged(0));
|
||||
sink.Delivered.ShouldBe(new byte[] { 100, 0 });
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Requesting_prior_success_after_failure_updates_delivery_and_replay()
|
||||
{
|
||||
var sink = new RecoveringPublisher();
|
||||
var deferred = new DeferredServiceLevelPublisher();
|
||||
deferred.SetInner(sink);
|
||||
var actor = ActorOfAsTestActorRef<OpcUaPublishActor>(OpcUaPublishActor.PropsForTests(serviceLevel: deferred));
|
||||
actor.Tell(new OpcUaPublishActor.ServiceLevelChanged(100));
|
||||
sink.Fail = true;
|
||||
actor.Tell(new OpcUaPublishActor.ServiceLevelChanged(0));
|
||||
sink.Fail = false;
|
||||
actor.Tell(new OpcUaPublishActor.ServiceLevelChanged(100));
|
||||
sink.Delivered.ShouldBe(new byte[] { 100, 100 });
|
||||
var rebound = new RecoveringPublisher();
|
||||
deferred.SetInner(rebound);
|
||||
rebound.Delivered.ShouldBe(new byte[] { 100 });
|
||||
}
|
||||
|
||||
private sealed class RecoveringPublisher : IServiceLevelPublisher
|
||||
{
|
||||
public bool Fail { get; set; }
|
||||
public List<byte> Delivered { get; } = new();
|
||||
public void Publish(byte level)
|
||||
{
|
||||
if (Fail) throw new InvalidOperationException("controlled failure");
|
||||
Delivered.Add(level);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Verifies that message contracts are accepted without a pinned dispatcher in tests.</summary>
|
||||
[Fact]
|
||||
public void Accepts_message_contracts_without_pinned_dispatcher_in_tests()
|
||||
|
||||
Reference in New Issue
Block a user