70 lines
2.5 KiB
C#
70 lines
2.5 KiB
C#
using System.Text;
|
|
using System.Text.Json;
|
|
using Microsoft.Extensions.Logging.Abstractions;
|
|
using VolcanicCore.Api.Engine;
|
|
using VolcanicCore.Core;
|
|
using VolcanicCore.Nodes;
|
|
|
|
// ImplicitUsings pulls System.Threading → collision with our ExecutionContext.
|
|
using ExecutionContext = VolcanicCore.Api.Engine.ExecutionContext;
|
|
|
|
namespace VolcanicCore.Tests;
|
|
|
|
public class NatsHookNodeTests
|
|
{
|
|
private sealed class FakeSubscriber : NatsHookNode.INatsSubscriber
|
|
{
|
|
private readonly List<byte[]> _payloads;
|
|
public FakeSubscriber(params byte[][] payloads) => _payloads = payloads.ToList();
|
|
public async IAsyncEnumerable<byte[]> SubscribeAsync(string subject,
|
|
[System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken ct)
|
|
{
|
|
foreach (var p in _payloads)
|
|
{
|
|
yield return p;
|
|
await Task.Yield();
|
|
}
|
|
}
|
|
}
|
|
|
|
private sealed class FakeInstanceRegistry : IConnectionInstanceRegistry
|
|
{
|
|
private readonly object _handle;
|
|
public FakeInstanceRegistry(object h) => _handle = h;
|
|
public Task<object> GetAsync(string id, CancellationToken ct) => Task.FromResult(_handle);
|
|
}
|
|
|
|
private sealed class FakeSink : ITriggerSink
|
|
{
|
|
public List<IReadOnlyList<Item>> Batches { get; } = new();
|
|
public Task EmitAsync(IReadOnlyList<Item> items, JsonElement? td, Guid? ans, CancellationToken ct)
|
|
{ lock (Batches) Batches.Add(items); return Task.CompletedTask; }
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Subscribes_and_emits_one_execution_per_message()
|
|
{
|
|
var node = new NatsHookNode();
|
|
var sub = new FakeSubscriber(
|
|
Encoding.UTF8.GetBytes("{\"n\":1}"),
|
|
Encoding.UTF8.GetBytes("{\"n\":2}"));
|
|
var reg = new FakeInstanceRegistry(sub);
|
|
var sink = new FakeSink();
|
|
|
|
var ctx = new ExecutionContext("", "wf", "n1",
|
|
JsonDocument.Parse("""{"subject":"topic"}""").RootElement,
|
|
null, NullLogger.Instance, "conn-1", reg);
|
|
|
|
var handle = await node.StartAsync(ctx, sink, CancellationToken.None);
|
|
|
|
// Give the pump a moment to drain the fake enumerable.
|
|
for (var i = 0; i < 50 && sink.Batches.Count < 2; i++) await Task.Delay(10);
|
|
|
|
Assert.Equal(2, sink.Batches.Count);
|
|
Assert.Equal(1, sink.Batches[0][0].Json.GetProperty("n").GetInt32());
|
|
Assert.Equal(2, sink.Batches[1][0].Json.GetProperty("n").GetInt32());
|
|
|
|
await handle.DisposeAsync();
|
|
}
|
|
}
|