feat(nodes): NatsPublishNode — publishes each item as its JSON bytes

src/VolcanicCore.Nodes/NatsPublishNode.cs

@@ -0,0 +1,51 @@
+using System.Text;
+using System.Text.Json;
+using NATS.Client.Core;
+using VolcanicCore.Core;
+
+namespace VolcanicCore.Nodes;
+
+public sealed class NatsPublishNode : IActionNode
+{
+    public interface INatsPublisher
+    {
+        Task PublishAsync(string subject, byte[] payload, CancellationToken ct);
+    }
+
+    private sealed class Adapter : INatsPublisher
+    {
+        private readonly INatsConnection _c;
+        public Adapter(INatsConnection c) => _c = c;
+        public async Task PublishAsync(string subject, byte[] payload, CancellationToken ct)
+            => await _c.PublishAsync(subject, payload, cancellationToken: ct);
+    }
+
+    public string Id => "nats.publish";
+    public string DisplayName => "NATS Publish";
+    public NodeCategory Category => NodeCategory.Terminal;
+    public int InputCount => 1;
+    public int OutputCount => 0;
+    public string? ConnectionProviderId => "nats";
+    public JsonSchema ConfigSchema { get; } = new(JsonDocument.Parse("""
+        {
+          "type": "object",
+          "required": ["subject"],
+          "properties": { "subject": { "type": "string", "title": "Subject" } }
+        }
+        """).RootElement);
+
+    public async Task<IReadOnlyList<IReadOnlyList<Item>>> ExecuteAsync(
+        INodeContext ctx, IReadOnlyList<IReadOnlyList<Item>> inputsByPort, CancellationToken ct)
+    {
+        var subject = ctx.Config.GetProperty("subject").GetString()!;
+        var raw = ctx.GetConnection<object>();
+        var pub = raw as INatsPublisher ?? new Adapter((INatsConnection)raw);
+
+        foreach (var item in inputsByPort[0])
+        {
+            var payload = Encoding.UTF8.GetBytes(item.Json.GetRawText());
+            await pub.PublishAsync(subject, payload, ct);
+        }
+        return Array.Empty<IReadOnlyList<Item>>();
+    }
+}

tests/VolcanicCore.Tests/NatsPublishNodeTests.cs

@@ -0,0 +1,50 @@
+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 NatsPublishNodeTests
+{
+    private sealed class FakePublisher : NatsPublishNode.INatsPublisher
+    {
+        public List<(string subject, byte[] payload)> Sent { get; } = new();
+        public Task PublishAsync(string subject, byte[] payload, CancellationToken ct)
+        { Sent.Add((subject, payload)); return Task.CompletedTask; }
+    }
+    private sealed class FakeReg : IConnectionInstanceRegistry
+    {
+        private readonly object _h;
+        public FakeReg(object h) => _h = h;
+        public Task<object> GetAsync(string id, CancellationToken ct) => Task.FromResult(_h);
+    }
+
+    [Fact]
+    public async Task Publishes_each_item_as_its_json_bytes()
+    {
+        var pub = new FakePublisher();
+        var reg = new FakeReg(pub);
+        var node = new NatsPublishNode();
+        var ctx = new ExecutionContext("e", "wf", "n",
+            JsonDocument.Parse("""{"subject":"out"}""").RootElement,
+            null, NullLogger.Instance, "c1", reg);
+
+        var input = new IReadOnlyList<Item>[]
+        {
+            new [] {
+                new Item(JsonDocument.Parse("""{"a":1}""").RootElement.Clone()),
+                new Item(JsonDocument.Parse("""{"a":2}""").RootElement.Clone()),
+            }
+        };
+        var outputs = await node.ExecuteAsync(ctx, input, CancellationToken.None);
+        Assert.Empty(outputs);
+        Assert.Equal(2, pub.Sent.Count);
+        Assert.All(pub.Sent, s => Assert.Equal("out", s.subject));
+        Assert.Equal("""{"a":1}""", System.Text.Encoding.UTF8.GetString(pub.Sent[0].payload));
+    }
+}