Adds Collection.flatMapBatches(of:). (#73)

* Adds `Collection.batchedSubscribe(by:)`.

h/t to @freak4pc and @natecook1000 for this implementation from a thread in iOS Folks. 👏🏽

* `Collection.batchedSubscribe(by:)` ⇒ `.flatMapBatches(of)`.

Author: jasdev

CombineExt.xcodeproj/project.pbxproj

@@ -31,6 +31,8 @@
 		BF3D3B67253B88E500D830ED /* IgnoreFailureTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = BF3D3B66253B88E500D830ED /* IgnoreFailureTests.swift */; };
 		BF43CC1525008B4F005AFA28 /* IgnoreOutputSetOutputType.swift in Sources */ = {isa = PBXBuildFile; fileRef = BF43CC1425008B4F005AFA28 /* IgnoreOutputSetOutputType.swift */; };
 		BF43CC1725008C45005AFA28 /* IgnoreOutputSetOutputTypeTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = BF43CC1625008C45005AFA28 /* IgnoreOutputSetOutputTypeTests.swift */; };
+		BFADDC8125BCE4C200465E9B /* FlatMapBatches.swift in Sources */ = {isa = PBXBuildFile; fileRef = BFADDC8025BCE4C200465E9B /* FlatMapBatches.swift */; };
+		BFADDC8B25BCE91E00465E9B /* FlatMapBatchesTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = BFADDC8A25BCE91E00465E9B /* FlatMapBatchesTests.swift */; };
 		C387777C24E6BBE900FAD2D8 /* Nwise.swift in Sources */ = {isa = PBXBuildFile; fileRef = C387777B24E6BBE900FAD2D8 /* Nwise.swift */; };
 		C387777F24E6BF8F00FAD2D8 /* NwiseTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = C387777D24E6BF6C00FAD2D8 /* NwiseTests.swift */; };
 		D836234824EA9446002353AC /* MergeMany.swift in Sources */ = {isa = PBXBuildFile; fileRef = D836234724EA9446002353AC /* MergeMany.swift */; };
@@ -112,6 +114,8 @@
 		BF3D3B66253B88E500D830ED /* IgnoreFailureTests.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = IgnoreFailureTests.swift; sourceTree = "<group>"; };
 		BF43CC1425008B4F005AFA28 /* IgnoreOutputSetOutputType.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = IgnoreOutputSetOutputType.swift; sourceTree = "<group>"; };
 		BF43CC1625008C45005AFA28 /* IgnoreOutputSetOutputTypeTests.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = IgnoreOutputSetOutputTypeTests.swift; sourceTree = "<group>"; };
+		BFADDC8025BCE4C200465E9B /* FlatMapBatches.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = FlatMapBatches.swift; sourceTree = "<group>"; };
+		BFADDC8A25BCE91E00465E9B /* FlatMapBatchesTests.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = FlatMapBatchesTests.swift; sourceTree = "<group>"; };
 		C387777B24E6BBE900FAD2D8 /* Nwise.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = Nwise.swift; sourceTree = "<group>"; };
 		C387777D24E6BF6C00FAD2D8 /* NwiseTests.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = NwiseTests.swift; sourceTree = "<group>"; };
 		"CombineExt::CombineExt::Product" /* CombineExt.framework */ = {isa = PBXFileReference; explicitFileType = wrapper.framework; path = CombineExt.framework; sourceTree = BUILT_PRODUCTS_DIR; };
@@ -256,6 +260,7 @@
 				OBJ_32 /* WithLatestFrom.swift */,
 				OBJ_33 /* ZipMany.swift */,
 				1970A8A925246FBD00799AB6 /* FilterMany.swift */,
+				BFADDC8025BCE4C200465E9B /* FlatMapBatches.swift */,
 			);
 			path = Operators;
 			sourceTree = "<group>";
@@ -307,6 +312,7 @@
 				OBJ_59 /* ToggleTests.swift */,
 				OBJ_60 /* WithLatestFromTests.swift */,
 				OBJ_61 /* ZipManyTests.swift */,
+				BFADDC8A25BCE91E00465E9B /* FlatMapBatchesTests.swift */,
 			);
 			path = Tests;
 			sourceTree = SOURCE_ROOT;
@@ -562,6 +568,7 @@
 				OBJ_137 /* ReplaySubjectTests.swift in Sources */,
 				OBJ_138 /* SetOutputTypeTests.swift in Sources */,
 				OBJ_139 /* ShareReplayTests.swift in Sources */,
+				BFADDC8B25BCE91E00465E9B /* FlatMapBatchesTests.swift in Sources */,
 				OBJ_140 /* ToggleTests.swift in Sources */,
 				OBJ_141 /* WithLatestFromTests.swift in Sources */,
 				BF43CC1725008C45005AFA28 /* IgnoreOutputSetOutputTypeTests.swift in Sources */,
@@ -584,6 +591,7 @@
 				OBJ_86 /* AssignToMany.swift in Sources */,
 				BF3D3B5D253B83F300D830ED /* IgnoreFailure.swift in Sources */,
 				BF330EFB24F20080001281FC /* Lock.swift in Sources */,
+				BFADDC8125BCE4C200465E9B /* FlatMapBatches.swift in Sources */,
 				OBJ_87 /* CombineLatestMany.swift in Sources */,
 				OBJ_88 /* Create.swift in Sources */,
 				OBJ_89 /* Dematerialize.swift in Sources */,

README.md

@@ -42,6 +42,7 @@ All operators, utilities and helpers respect Combine's publisher contract, inclu
 * [nwise(_:) and pairwise()](#nwise)
 * [ignoreOutput(setOutputType:)](#ignoreOutputsetOutputType)
 * [ignoreFailure](#ignoreFailure)
+* [flatMapBatches(of:)](#flatMapBatchesof)
 
 ### Publishers
 * [AnyPublisher.create](#AnypublisherCreate)
@@ -504,6 +505,8 @@ subscription = [1, 1, 2, 1, 3, 3, 4].publisher
   .sink(receiveValue: { print("removeAllDuplicates: \($0)") })
 ```
 
+#### Output:
+
 ```none
 removeAllDuplicates: 1
 removeAllDuplicates: 2
@@ -535,6 +538,8 @@ subscription2 = replayedPublisher
   .sink(receiveValue: { print("second subscriber: \($0)") })
 ```
 
+#### Output:
+
 ```none
 first subscriber: 1
 first subscriber: 2
@@ -565,6 +570,8 @@ DispatchQueue.main.asyncAfter(deadline: .now() + 1) {
 }
 ```
 
+#### Output:
+
 ```none
 1
 2
@@ -587,6 +594,8 @@ subject.send(false)
 subject.send(true)
 ```
 
+#### Output:
+
 ```none
 false
 true
@@ -611,6 +620,8 @@ subject.send(4)
 subject.send(5)
 ```
 
+#### Output:
+
 ```none
 [1, 2, 3]
 [2, 3, 4]
@@ -635,6 +646,8 @@ subject.send(4)
 subject.send(5)
 ```
 
+#### Output:
+
 ```none
 1 -> 2
 2 -> 3
@@ -676,13 +689,36 @@ subject.send(3)
 subject.send(completion: .failure(.someError))
 ```
 
+#### Output:
+
 ```none
 1
 2
 3
 .finished
 ```
 
+### flatMapBatches(of:)
+
+`Collection.flatMapBatches(of:)` subscribes to the receiver’s contained publishers in batches and returns their outputs in batches, too (while maintaining order). Subsequent batches of publishers are only subscribed to when prior batches successfully complete — any one failure is forwarded downstream.
+
+```swift
+let ints = (1...6).map(Just.init)
+
+subscription = ints
+  .flatMapBatches(of: 2)
+  .sink(receiveCompletion: { print($0) }, receiveValue: { print($0) })
+```
+
+#### Output:
+
+```none
+[1, 2]
+[3, 4]
+[5, 6]
+.finished
+```
+
 ## Publishers
 
 This section outlines some of the custom Combine publishers CombineExt provides

Sources/Operators/FlatMapBatches.swift

@@ -0,0 +1,44 @@
+//
+//  FlatMapBatches.swift
+//  CombineExt
+//
+//  Created by Shai Mishali, Nate Cook, and Jasdev Singh on 21/01/2021.
+//  Copyright © 2021 Combine Community. All rights reserved.
+//
+
+#if canImport(Combine)
+import Combine
+
+@available(OSX 10.15, iOS 13.0, tvOS 13.0, watchOS 6.0, *)
+public extension Collection where Element: Publisher {
+    /// Subscribes to the receiver’s contained publishers `size` at a time
+    /// and outputs their results in `size`-sized batches, while maintaining
+    /// order within each batch — subsequent batches of publishers are only
+    /// subscribed to when the batch before it successfully completes. Any
+    /// one failure will be forwarded downstream.
+    /// - Parameter size: The batch size.
+    /// - Returns: A publisher that subscribes to `self`’s contained publishers
+    /// `size` at a time, returning their results in-order in `size`-sized
+    /// batches, and then repeats with subsequent batches only if the ones prior
+    /// successfully completed. Any one failure is immediately forwarded downstream.
+    func flatMapBatches(of size: Int) -> AnyPublisher<[Element.Output], Element.Failure> {
+        precondition(size > 0, "Batch sizes must be positive.")
+
+        let indexBreaks = sequence(
+            first: startIndex,
+            next: {
+                $0 == endIndex ?
+                    nil :
+                    index($0, offsetBy: size, limitedBy: endIndex)
+                        ?? endIndex
+            }
+        )
+
+        return Swift.zip(indexBreaks, indexBreaks.dropFirst())
+            .publisher
+            .setFailureType(to: Element.Failure.self)
+            .flatMap(maxPublishers: .max(1)) { self[$0..<$1].zip() }
+            .eraseToAnyPublisher()
+    }
+}
+#endif

Tests/FlatMapBatchesTests.swift

@@ -0,0 +1,149 @@
+//
+//  FlatMapBatchesTests.swift
+//  CombineExtTests
+//
+//  Created by Jasdev Singh on 23/01/2021.
+//  Copyright © 2021 Combine Community. All rights reserved.
+//
+
+#if !os(watchOS)
+import XCTest
+import Combine
+import CombineExt
+
+@available(OSX 10.15, iOS 13.0, tvOS 13.0, watchOS 6.0, *)
+final class FlatMapBatchesTests: XCTestCase {
+    private var subscription: AnyCancellable!
+
+    private enum BatchedSubscribeError: Error, Equatable {
+        case anError
+    }
+
+    func testEvenBatches() {
+        let ints = (1...6).map(Just.init)
+
+        var results = [[Int]]()
+        var completed = false
+
+        subscription = ints
+            .flatMapBatches(of: 2)
+            .sink(receiveCompletion: { _ in completed = true },
+                  receiveValue: { results.append($0) })
+
+        XCTAssertEqual(results, [[1, 2], [3, 4], [5, 6]])
+        XCTAssertTrue(completed)
+    }
+
+    func testUnevenBatches() {
+        let ints = (1...5).map(Just.init)
+
+        var results = [[Int]]()
+        var completed = false
+
+        subscription = ints
+            .flatMapBatches(of: 2)
+            .sink(receiveCompletion: { _ in completed = true },
+                  receiveValue: { results.append($0) })
+
+        XCTAssertEqual(results, [[1, 2], [3, 4], [5]])
+        XCTAssertTrue(completed)
+    }
+
+    func testForwardsError() {
+        let publishers = [Fail(error: BatchedSubscribeError.anError).eraseToAnyPublisher()] +
+            (1...3).map {
+                Just($0)
+                    .setFailureType(to: BatchedSubscribeError.self)
+                    .eraseToAnyPublisher()
+            }
+
+        var results = [[Int]]()
+        var completion: Subscribers.Completion<BatchedSubscribeError>?
+
+        subscription = publishers
+            .flatMapBatches(of: 2)
+            .sink(receiveCompletion: { completion = $0 },
+                  receiveValue: { results.append($0) })
+
+        XCTAssertTrue(results.isEmpty)
+        XCTAssertEqual(completion, .failure(.anError))
+    }
+
+    func testHangsIfEarlierBatchDoesntComplete() {
+        let uncompleted = (1...2).map { number in
+            AnyPublisher<Int, Never>.create { subscriber in
+                subscriber.send(number)
+                return AnyCancellable { }
+            }
+        }
+
+        let publishers = uncompleted +
+            (3...4).map(Just.init).map(AnyPublisher.init)
+
+        var results = [[Int]]()
+        var completed = false
+
+        subscription = publishers
+            .flatMapBatches(of: 2)
+            .sink(receiveCompletion: { _ in completed = true },
+                  receiveValue: { results.append($0) })
+
+        XCTAssertEqual(results, [[1, 2]])
+        XCTAssertFalse(completed)
+    }
+
+    func testEmptyCollection() {
+        let publishers = EmptyCollection<AnyPublisher<Int, Never>>()
+
+        var results = [[Int]]()
+        var completed = false
+
+        subscription = publishers
+            .flatMapBatches(of: 2)
+            .sink(receiveCompletion: { _ in completed = true },
+                  receiveValue: { results.append($0) })
+
+        XCTAssertTrue(results.isEmpty)
+        XCTAssertTrue(completed)
+    }
+
+    func testBatchLimitLargerThanCount() {
+        let ints = [Just(1)]
+
+        var results = [[Int]]()
+        var completed = false
+
+        subscription = ints
+            .flatMapBatches(of: 2)
+            .sink(receiveCompletion: { _ in completed = true },
+                  receiveValue: { results.append($0) })
+
+        XCTAssertEqual(results, [[1]])
+        XCTAssertTrue(completed)
+    }
+
+    func testMultipleOutputsPerPublisher() {
+        let publishers = (1...2).map { number in
+            AnyPublisher<Int, Never>.create { subscriber in
+                subscriber.send(number)
+                subscriber.send(number)
+                subscriber.send(completion: .finished)
+
+                return AnyCancellable { }
+            }
+        } +
+        (3...4).map(Just.init).map(AnyPublisher.init)
+
+        var results = [[Int]]()
+        var completed = false
+
+        subscription = publishers
+            .flatMapBatches(of: 2)
+            .sink(receiveCompletion: { _ in completed = true },
+                  receiveValue: { results.append($0) })
+
+        XCTAssertEqual(results, [[1, 2], [1, 2], [3, 4]])
+        XCTAssertTrue(completed)
+    }
+}
+#endif