feat: add Union pipe to combine the output of many pipes as a single …

…pipe (#34)

feat: add Union pipe to combine the output of many pipes and present them as single pipe

Streamistry.Core/Union.cs

@@ -0,0 +1,31 @@
+using System;
+using System.Collections.Generic;
+using System.IO;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+using Streamistry.Observability;
+
+namespace Streamistry;
+
+/// <summary>
+/// Represents a pipeline element that merges two or more upstream streams into a single downstream stream by emetting each values from each upstream.
+/// </summary>
+/// <typeparam name="TInput">The type of the elements in any input stream.</typeparam>
+public class Union<TInput> : ChainablePipe<TInput>
+{
+    public Union(IChainablePipe<TInput>[] upstreams)
+    : base(upstreams[0]?.GetObservabilityProvider())
+    {
+        foreach (var upstream in upstreams)
+            upstream.RegisterDownstream(Emit, Complete);
+    }
+
+    [Meter]
+    public void Emit(TInput? obj)
+        => PushDownstream(Invoke(obj));
+
+    [Trace]
+    protected TInput? Invoke(TInput? obj)
+        => obj;
+}

Streamistry.Testing/UnionTests.cs

@@ -0,0 +1,32 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Security.Cryptography.X509Certificates;
+using System.Text;
+using System.Threading.Tasks;
+using NUnit.Framework;
+using Streamistry.Pipes.Sinks;
+using Streamistry.Pipes.Sources;
+
+namespace Streamistry.Testing;
+public class UnionTests
+{
+    [Test]
+    public void Union_ManyPipes_AsSinglePipe()
+    {
+
+        var firstSource = new EnumerableSource<int>([1, 2, 3]);
+        var secondSource = new EnumerableSource<int>([10, 20, 30]);
+        var union = new Union<int>([firstSource, secondSource]);
+        var sink = new MemorySink<int>(union);
+        var pipeline = new Pipeline([firstSource, secondSource]);
+        pipeline.Start();
+
+        Assert.Multiple(() =>
+        {
+            Assert.That(sink.State, Has.Count.EqualTo(6));
+            Assert.That(sink.State.First(), Is.EqualTo(1));
+            Assert.That(sink.State.Last(), Is.EqualTo(30));
+        });
+    }
+}