fix: Merge immediately finishes when empty array of bases

Author: twittemb

Sources/Combiners/Merge/AsyncMergeSequence.swift

@@ -34,15 +34,19 @@ public struct AsyncMergeSequence<Base: AsyncSequence>: AsyncSequence {
   }
 
   public struct Iterator: AsyncIteratorProtocol {
+    private let isEmpty: Bool
     let mergeStateMachine: MergeStateMachine<Element>
 
     init(bases: [Base]) {
+      isEmpty = bases.isEmpty
       self.mergeStateMachine = MergeStateMachine(
         bases
       )
     }
 
     public mutating func next() async rethrows -> Element? {
+      guard !self.isEmpty else { return nil }
+
       let mergedElement = await self.mergeStateMachine.next()
       switch mergedElement {
         case .element(let result):

Sources/Combiners/Merge/MergeStateMachine.swift

@@ -98,7 +98,8 @@ struct MergeStateMachine<Element>: Sendable {
     var regulators = [Regulator<Base>]()
 
     for base in bases {
-      let regulator = Regulator<Base>(base, onNextRegulatedElement: { [state] in Self.onNextRegulatedElement($0, state: state) })
+      let regulator = Regulator<Base>(base, onNextRegulatedElement: { [state] in Self.onNextRegulatedElement($0, state: state) }
+      )
       regulators.append(regulator)
     }
 

Tests/Combiners/Merge/AsyncMergeSequenceTests.swift

@@ -76,7 +76,6 @@ final class AsyncMergeSequenceTests: XCTestCase {
 
     let expectedElements = asyncSequence1 + asyncSequence2 + asyncSequence3 + asyncSequence4
 
-
     let sut = merge(asyncSequence1.async, asyncSequence2.async, asyncSequence3.async, asyncSequence4.async)
 
     var receivedElements = [Int]()
@@ -247,7 +246,7 @@ final class AsyncMergeSequenceTests: XCTestCase {
       for try await element in sut {
         firstElement = element
         canCancelExpectation.fulfill()
-        wait(for: [hasCancelExceptation], timeout: 5)
+        await fulfillment(of: [hasCancelExceptation], timeout: 5)
       }
       XCTAssertEqual(firstElement, 10)
       taskHasFinishedExpectation.fulfill()
@@ -289,4 +288,22 @@ final class AsyncMergeSequenceTests: XCTestCase {
 
     wait(for: [hasCancelExceptation], timeout: 1)
   }
+
+  func testMerge_finishes_when_empty_array_of_base() {
+    let sut = AsyncMergeSequence<AsyncStream<Int>>([])
+    let hasFinishedExpectation = expectation(description: "Merge has finished")
+
+    let task = Task {
+      var received = [Int]()
+      for try await element in sut {
+        received.append(element)
+      }
+      XCTAssertTrue(received.isEmpty)
+      hasFinishedExpectation.fulfill()
+    }
+
+    wait(for: [hasFinishedExpectation], timeout: 1)
+
+    task.cancel()
+  }
 }