Add operator: Merge(Observable<Observable<T>>)

src/R3/Operators/Merge.cs

@@ -0,0 +1,130 @@
+namespace R3;
+
+public static partial class Observable
+{
+    public static Observable<T> Merge<T>(this Observable<Observable<T>> sources)
+    {
+        return new MergeMany<T>(sources);
+    }
+}
+
+internal sealed class MergeMany<T>(Observable<Observable<T>> sources) : Observable<T>
+{
+    protected override IDisposable SubscribeCore(Observer<T> observer)
+    {
+        return sources.Subscribe(new _MergeMany(observer));
+    }
+
+    sealed class _MergeMany(Observer<T> observer) : Observer<Observable<T>>
+    {
+        // keep when inner is running
+        protected override bool AutoDisposeOnCompleted => false;
+
+        readonly Observer<T> observer = observer;
+        readonly object gate = new();
+        readonly CompositeDisposable subscriptions = new();
+        bool isStopped;
+
+        protected override void OnNextCore(Observable<T> value)
+        {
+            var innerObserver = new MergeInner(this);
+            var subscription = value.Subscribe(innerObserver);
+
+            lock (gate)
+            {
+                subscriptions.Add(subscription);
+            }
+        }
+
+        protected override void OnErrorResumeCore(Exception error)
+        {
+            lock (gate)
+            {
+                observer.OnErrorResume(error);
+            }
+        }
+
+        protected override void OnCompletedCore(Result result)
+        {
+            lock (gate)
+            {
+                if (result.IsFailure)
+                {
+                    PublishCompleted(result);
+                }
+                else
+                {
+                    isStopped = true;
+                    // when no running inner
+                    if (subscriptions.Count <= 0)
+                    {
+                        PublishCompleted(result);
+                    }
+                }
+            }
+        }
+
+        protected override void DisposeCore()
+        {
+            subscriptions.Dispose();
+        }
+
+        void PublishCompleted(Result result)
+        {
+            try
+            {
+                lock (gate)
+                {
+                    observer.OnCompleted(result);
+                }
+            }
+            finally
+            {
+                Dispose();
+            }
+        }
+
+        sealed class MergeInner(_MergeMany parent) : Observer<T>
+        {
+            protected override void OnNextCore(T value)
+            {
+                lock (parent.gate)
+                {
+                    parent.observer.OnNext(value);
+                }
+            }
+
+            protected override void OnErrorResumeCore(Exception error)
+            {
+                lock (parent.gate)
+                {
+                    parent.observer.OnErrorResume(error);
+                }
+            }
+
+            protected override void OnCompletedCore(Result result)
+            {
+                lock (parent.gate)
+                {
+                    if (result.IsFailure)
+                    {
+                        parent.observer.OnCompleted(result);
+                    }
+                    else
+                    {
+                        // when all sources are completed, then this observer is completed
+                        if (parent is { isStopped: true, subscriptions.Count: 1 }) // only self
+                        {
+                            parent.PublishCompleted(result);
+                        }
+                    }
+                }
+            }
+
+            protected override void DisposeCore()
+            {
+                parent.subscriptions.Remove(this);
+            }
+        }
+    }
+}

tests/R3.Tests/OperatorTests/MergeTest.cs

@@ -56,4 +56,91 @@ public void Empty()
 
         list.AssertIsCompleted();
     }
+
+    [Fact]
+    public void NestedSources()
+    {
+        var outerSubject = new Subject<Observable<int>>();
+
+        using var list = Observable.Merge(outerSubject).ToLiveList();
+
+        var innerSubject1 = new Subject<int>();
+        var innerSubject2 = new Subject<int>();
+        var innerSubject3 = new Subject<int>();
+
+        outerSubject.OnNext(innerSubject1);
+        innerSubject1.OnNext(10);
+        innerSubject1.OnNext(11);
+
+        outerSubject.OnNext(innerSubject2);
+        innerSubject2.OnNext(20);
+        innerSubject1.OnNext(12);
+        innerSubject2.OnNext(21);
+
+        outerSubject.OnNext(innerSubject3);
+        innerSubject3.OnNext(30);
+        innerSubject2.OnNext(22);
+        innerSubject1.OnNext(12);
+
+        innerSubject1.OnCompleted();
+        innerSubject2.OnCompleted();
+        innerSubject3.OnCompleted();
+
+        list.AssertIsNotCompleted();
+
+        outerSubject.OnCompleted();
+
+        list.AssertIsCompleted();
+        list.AssertEqual([10, 11, 20, 12, 21, 30, 22, 12]);
+    }
+
+    [Fact]
+    public void NestedSources_WaitForInners()
+    {
+        var outerSubject = new Subject<Observable<int>>();
+
+        using var list = Observable.Merge(outerSubject).ToLiveList();
+
+        var innerSubject1 = new Subject<int>();
+        var innerSubject2 = new Subject<int>();
+        var innerSubject3 = new Subject<int>();
+
+        outerSubject.OnNext(innerSubject1);
+        innerSubject1.OnNext(10);
+        innerSubject1.OnNext(11);
+
+        outerSubject.OnNext(innerSubject2);
+        innerSubject2.OnNext(20);
+        innerSubject1.OnNext(12);
+        innerSubject2.OnNext(21);
+
+        outerSubject.OnNext(innerSubject3);
+        innerSubject3.OnNext(30);
+        innerSubject2.OnNext(22);
+        innerSubject1.OnNext(12);
+
+        innerSubject1.OnCompleted();
+        innerSubject3.OnCompleted();
+        outerSubject.OnCompleted();
+
+        list.AssertIsNotCompleted();
+
+        innerSubject2.OnCompleted();
+
+        list.AssertIsCompleted();
+        list.AssertEqual([10, 11, 20, 12, 21, 30, 22, 12]);
+    }
+
+    [Fact]
+    public void NestedSources_Empty()
+    {
+        var outerSubject = new Subject<Observable<int>>();
+
+        using var list = Observable.Merge(outerSubject).ToLiveList();
+        list.AssertIsNotCompleted();
+        outerSubject.OnCompleted();
+
+        list.AssertIsCompleted();
+        list.AssertEqual([]);
+    }
 }