From b9a3926be29e47bc7fcf24ba6016acf5510972e6 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 28 Aug 2026 15:11:49 +0000 Subject: [PATCH] =?UTF-8?q?feat(nodes):=20NatsHookNode=20=E2=80=94=20trigg?= =?UTF-8?q?er=20emits=20one=20execution=20per=20message?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/VolcanicCore.Nodes/NatsHookNode.cs | 86 +++++++++++++++++++ tests/VolcanicCore.Tests/NatsHookNodeTests.cs | 69 +++++++++++++++ 2 files changed, 155 insertions(+) create mode 100644 src/VolcanicCore.Nodes/NatsHookNode.cs create mode 100644 tests/VolcanicCore.Tests/NatsHookNodeTests.cs diff --git a/src/VolcanicCore.Nodes/NatsHookNode.cs b/src/VolcanicCore.Nodes/NatsHookNode.cs new file mode 100644 index 0000000..3aae4bd --- /dev/null +++ b/src/VolcanicCore.Nodes/NatsHookNode.cs @@ -0,0 +1,86 @@ +using System.Text; +using System.Text.Json; +using NATS.Client.Core; +using VolcanicCore.Core; + +namespace VolcanicCore.Nodes; + +public sealed class NatsHookNode : ITriggerNode +{ + // Testable seam over the tiny slice of INatsConnection we actually use. + public interface INatsSubscriber + { + IAsyncEnumerable SubscribeAsync(string subject, CancellationToken ct); + } + + private sealed class NatsSubscriberAdapter : INatsSubscriber + { + private readonly INatsConnection _conn; + public NatsSubscriberAdapter(INatsConnection c) => _conn = c; + public async IAsyncEnumerable SubscribeAsync(string subject, + [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken ct) + { + await foreach (var msg in _conn.SubscribeAsync(subject, cancellationToken: ct)) + if (msg.Data is not null) yield return msg.Data; + } + } + + public string Id => "nats.hook"; + public string DisplayName => "NATS Hook"; + public NodeCategory Category => NodeCategory.Trigger; + public int InputCount => 0; + public int OutputCount => 1; + public string? ConnectionProviderId => "nats"; + public JsonSchema ConfigSchema { get; } = new(JsonDocument.Parse(""" + { + "type": "object", + "required": ["subject"], + "properties": { "subject": { "type": "string", "title": "Subject" } } + } + """).RootElement); + + public Task StartAsync(INodeContext ctx, ITriggerSink sink, CancellationToken ct) + { + var subject = ctx.Config.GetProperty("subject").GetString()!; + var raw = ctx.GetConnection(); + var sub = raw as INatsSubscriber ?? new NatsSubscriberAdapter((INatsConnection)raw); + + var pumpCts = CancellationTokenSource.CreateLinkedTokenSource(ct); + var pump = Task.Run(async () => + { + try + { + await foreach (var payload in sub.SubscribeAsync(subject, pumpCts.Token)) + { + JsonElement json; + try + { + using var doc = JsonDocument.Parse(payload); + json = doc.RootElement.Clone(); + } + catch (JsonException) + { + json = JsonSerializer.SerializeToElement(new { raw = Encoding.UTF8.GetString(payload) }); + } + await sink.EmitAsync(new[] { new Item(json) }, ct: pumpCts.Token); + } + } + catch (OperationCanceledException) { /* stop */ } + }, pumpCts.Token); + + return Task.FromResult(new Handle(pumpCts, pump)); + } + + private sealed class Handle : IAsyncDisposable + { + private readonly CancellationTokenSource _cts; + private readonly Task _pump; + public Handle(CancellationTokenSource c, Task p) { _cts = c; _pump = p; } + public async ValueTask DisposeAsync() + { + _cts.Cancel(); + try { await _pump; } catch { } + _cts.Dispose(); + } + } +} diff --git a/tests/VolcanicCore.Tests/NatsHookNodeTests.cs b/tests/VolcanicCore.Tests/NatsHookNodeTests.cs new file mode 100644 index 0000000..d140e89 --- /dev/null +++ b/tests/VolcanicCore.Tests/NatsHookNodeTests.cs @@ -0,0 +1,69 @@ +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 _payloads; + public FakeSubscriber(params byte[][] payloads) => _payloads = payloads.ToList(); + public async IAsyncEnumerable 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 GetAsync(string id, CancellationToken ct) => Task.FromResult(_handle); + } + + private sealed class FakeSink : ITriggerSink + { + public List> Batches { get; } = new(); + public Task EmitAsync(IReadOnlyList 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(); + } +}