Refactored flatMapSuccess(_:) to utilize flatMap(_:), and added `…

…flatMapError(_:)` to the collection. (#7)

Also:
- Added convenience computed `var error: E?` to StreamState.
- Added tests for flatMap methods
- Added tests for `Subscriber.combine(_:_:)`
- Added additional test helpers for testing Futures and Subscribers.

Sources/Propagate/PromiseAndFuture/Future+Map.swift

@@ -175,8 +175,8 @@ public extension Future {
  @discardableResult func flatMap<NewT, NewE: Error>(_ mapBlock: @escaping (Result<T,E>) -> Future<NewT,NewE>) -> Future<NewT,NewE> {
  let promise = Promise<NewT,NewE>()
 
- finally {
- mapBlock($0)
+ finally { result in
+ mapBlock(result)
  .onSuccess { promise.resolve($0) }
  .onFailure { promise.reject($0) }
  }
@@ -187,23 +187,35 @@ public extension Future {
  /**
  Operator. Maps the success value of this future to the kicking off of another future. Errors are passed through unchanged.
  
- - Parameter mapBlock: Closure which generates the new future. Takes in the `Result<T,E>` from the completion of the first future.
+ - Parameter mapBlock: Closure which generates the new future. Takes in the success value of the first future.
  - returns: The new future, as a `@discardableResult` to allow for the chaining of mutation/callback methods.
  */
  @discardableResult func flatMapSuccess<NewT>(_ mapBlock: @escaping (T) -> Future<NewT,E>) -> Future<NewT,E> {
- let promise = Promise<NewT,E>()
-
- onSuccess {
- mapBlock($0)
- .onSuccess { promise.resolve($0) }
- .onFailure { promise.reject($0) }
+ flatMap { result in
+ switch result {
+ case .success(let value):
+ return mapBlock(value)
+ case .failure(let error):
+ return .error(error)
+ }
  }
-
- onFailure { error in
- promise.reject(error)
+ }
+
+ /**
+ Operator. Chains this future such that if it fails, it kicks off another future. If it succeeds, the success value is handed back immediately.
+ 
+ - Parameter mapBlock: Closure which generates the new future. Takes in the error value of the first future.
+ - returns: The new future, as a `@discardableResult` to allow for the chaining of mutation/callback methods.
+ */
+ @discardableResult func flatMapError<NewE: Error>(_ mapBlock: @escaping (E) -> Future<T,NewE>) -> Future<T,NewE> {
+ flatMap { result in
+ switch result {
+ case .success(let value):
+ return .of(value)
+ case .failure(let error):
+ return mapBlock(error)
+ }
  }
-
- return promise.future
  }
 
 }

Sources/Propagate/PublisherAndSubscriber/StreamState.swift

@@ -31,4 +31,13 @@ public extension StreamState {
  }
  }
 
+ var error: E? {
+ switch self {
+ case .error(let error):
+ return error
+ default:
+ return nil
+ }
+ }
+
 }

Sources/Propagate/PublisherAndSubscriber/Subscriber+Combine.swift

@@ -18,12 +18,25 @@ public extension Subscriber {
  publisher.publish(tuple)
  }
  }
+ .onError {
+ publisher.publish($0)
+ }
+ .onCancelled {
+ publisher.cancelAll()
+ }
+
  sub2.onNewData {
  tupleCreator.item2 = $0
  if let tuple = tupleCreator.tuple {
  publisher.publish(tuple)
  }
  }
+ .onError {
+ publisher.publish($0)
+ }
+ .onCancelled {
+ publisher.cancelAll()
+ }
 
  return publisher.subscriber()
  .onCancelled {

Tests/PropagateTests/FlatMapTests.swift

@@ -0,0 +1,127 @@
+// FlatMapTests.swift
+// Created by Jacob Hawken on 8/13/22.
+
+import Propagate
+import XCTest
+
+class FlatMapTests: XCTestCase {
+
+ private var promise: Promise<Int,TestError>!
+ private var future: Future<Int,TestError>!
+
+ override func setUp() {
+ promise = .init()
+ future = promise.future
+ }
+
+ override func tearDown() {
+ promise = nil
+ future = nil
+ }
+
+ func testFlatMapTriggersSecondFuture() {
+ var flatMappedFuture = future.flatMap { result in
+ mapIntResultToStringFuture(result)
+ }
+ promise.resolve(5)
+ waitForCompletion(of: flatMappedFuture)
+ XCTAssertEqual(flatMappedFuture.value, "5")
+
+ promise = .init()
+ future = promise.future
+ flatMappedFuture = future.flatMap { result in
+ mapIntResultToStringFuture(result)
+ }
+ promise.reject(TestError.case1)
+ waitForCompletion(of: flatMappedFuture)
+ XCTAssertEqual(flatMappedFuture.error, OtherTestError.case1)
+ }
+
+ func testFlatMapSuccessTriggersSecondFuture() {
+ let flatMappedFuture = future.flatMapSuccess { int in
+ triggerFutureForInt(int, shouldSucceed: true)
+ }
+ promise.resolve(3)
+ waitForCompletion(of: flatMappedFuture)
+ XCTAssertEqual(flatMappedFuture.value, "3")
+ }
+
+ func testFlatMapSuccessReturnsErrorOfSecondFuture() {
+ let flatMappedFuture = future.flatMapSuccess { int in
+ triggerFutureForInt(int, shouldSucceed: false)
+ }
+ promise.resolve(3)
+ waitForCompletion(of: flatMappedFuture)
+ XCTAssertEqual(flatMappedFuture.error, TestError.case1)
+ }
+
+ func testFlatMapSuccessPassesThroughError() {
+ let flatMappedFuture = future.flatMapSuccess { int in
+ triggerFutureForInt(int, shouldSucceed: true)
+ }
+ promise.reject(TestError.case2)
+ waitForCompletion(of: flatMappedFuture)
+ XCTAssertEqual(flatMappedFuture.error, TestError.case2)
+ }
+
+ func testFlatMapErrorTriggersSecondFuture() {
+ let flatMappedFuture = future.flatMapError { error in
+ triggerFutureForError(error, shouldSucceed: true)
+ }
+ promise.reject(.case1)
+ waitForCompletion(of: flatMappedFuture)
+ XCTAssertEqual(flatMappedFuture.value, 69)
+ }
+
+ func testFlatMapErrorReturnsErrorOfSecondFuture() {
+ let flatMappedFuture = future.flatMapError { error in
+ triggerFutureForError(error, shouldSucceed: false)
+ }
+ promise.reject(TestError.case2)
+ waitForCompletion(of: flatMappedFuture)
+ XCTAssertEqual(flatMappedFuture.error, OtherTestError.case2)
+ }
+
+ func testFlatMapErrorPassesThroughSuccess() {
+ let flatMappedFuture = future.flatMapError { error in
+ triggerFutureForError(error, shouldSucceed: false)
+ }
+ promise.resolve(17)
+ waitForCompletion(of: flatMappedFuture)
+ XCTAssertEqual(flatMappedFuture.value, 17)
+ }
+
+}
+
+private func mapIntResultToStringFuture(_ result: Result<Int,TestError>) -> Future<String,OtherTestError> {
+ switch result {
+ case .success(let int):
+ return .of("\(int)")
+ case .failure(let error):
+ switch error {
+ case .case1:
+ return .error(.case1)
+ case .case2:
+ return .error(.case2)
+ }
+ }
+}
+
+private func triggerFutureForInt(_ int: Int, shouldSucceed: Bool) -> Future<String,TestError> {
+ if shouldSucceed {
+ return .of("\(int)")
+ }
+ return .error(.case1)
+}
+
+private func triggerFutureForError(_ error: TestError, shouldSucceed: Bool) -> Future<Int,OtherTestError> {
+ if shouldSucceed {
+ return .of(69)
+ }
+ switch error {
+ case .case1:
+ return .error(.case1)
+ case .case2:
+ return .error(.case2)
+ }
+}

Tests/PropagateTests/StatefulPublisherTests.swift

@@ -28,9 +28,9 @@ class StatefulPublishertests: XCTestCase {
 
  sourcePublisher.publish(5)
 
- waitForNextState {
- subject.onNewData { receivedValue = $0 }
- }
+ waitForNextState(
+ forSubscriber: subject.onNewData { receivedValue = $0 }
+ )
 
  XCTAssertNotNil(receivedValue)
  XCTAssertEqual(receivedValue, 5)

Tests/PropagateTests/SubscriberCombineTests.swift

@@ -7,15 +7,71 @@ import XCTest
 
 class SubscriberCombineTests: XCTestCase {
 
- private var publisher: Publisher<Int,TestError>!
- private var subscriber: Subscriber<Int,TestError>!
+ private var publisher1: Publisher<Int,TestError>!
+ private var subscriber1: Subscriber<Int,TestError>!
+
+ private var publisher2: Publisher<String,TestError>!
+ private var subscriber2: Subscriber<String,TestError>!
 
- override func setUpWithError() throws {
- publisher = .init()
- subscriber = publisher.subscriber()
+ override func setUp() {
+ publisher1 = .init()
+ subscriber1 = publisher1.subscriber()
+
+ publisher2 = .init()
+ subscriber2 = publisher2.subscriber()
  }
 
- override func tearDownWithError() throws {
+ override func tearDown() {
+ publisher1 = nil
+ subscriber1 = nil
+ publisher2 = nil
+ subscriber2 = nil
+ }
+
+ func testCombineTwoSubscribers() {
+ var lastStateReceived: Subscriber<(Int, String), TestError>.State?
+ let combinedSub = Subscriber.combine(subscriber1, subscriber2).subscribe {
+ lastStateReceived = $0
+ }
+
+ publisher2.publish("A")
+ confirmNextStateDoesntTrigger(forSubscriber: combinedSub)
+ XCTAssertNil(lastStateReceived)
+
+ publisher1.publish(1)
+ waitForNextState(forSubscriber: combinedSub)
+ XCTAssertNotNil(lastStateReceived)
+ XCTAssertNotNil(lastStateReceived?.value)
+ XCTAssertEqual(lastStateReceived?.value?.0, 1)
+ XCTAssertEqual(lastStateReceived?.value?.1, "A")
+
+ publisher1.publish(2)
+ waitForNextState(forSubscriber: combinedSub)
+ XCTAssertNotNil(lastStateReceived?.value)
+ XCTAssertEqual(lastStateReceived?.value?.0, 2)
+ XCTAssertEqual(lastStateReceived?.value?.1, "A")
+
+ publisher2.publish(.case2)
+ waitForNextState(forSubscriber: combinedSub)
+ XCTAssertNotNil(lastStateReceived?.error)
+ XCTAssertEqual(lastStateReceived?.error, .case2)
+
+ publisher2.publish("B")
+ waitForNextState(forSubscriber: combinedSub)
+ XCTAssertNotNil(lastStateReceived?.value)
+ XCTAssertEqual(lastStateReceived?.value?.0, 2)
+ XCTAssertEqual(lastStateReceived?.value?.1, "B")
+
+ publisher1.publish(.case1)
+ waitForNextState(forSubscriber: combinedSub)
+ XCTAssertNotNil(lastStateReceived?.error)
+ XCTAssertEqual(lastStateReceived?.error, .case1)
+
+ publisher1.publish(3)
+ waitForNextState(forSubscriber: combinedSub)
+ XCTAssertNotNil(lastStateReceived?.value)
+ XCTAssertEqual(lastStateReceived?.value?.0, 3)
+ XCTAssertEqual(lastStateReceived?.value?.1, "B")
  }
 
 }

Tests/PropagateTests/TestHelpers.swift

@@ -26,23 +26,64 @@ extension Date {
  }
 }
 
-enum TestError: Error {
+enum TestError: Error, Equatable {
  case case1
+ case case2
+}
+
+enum OtherTestError: Error, Equatable {
+ case case1
+ case case2
 }
 
 extension XCTestCase {
 
- @discardableResult func waitForNextState<T,E:Error>(timeout: TimeInterval = 0.01, subscriberBlock: () -> Subscriber<T,E>) -> Subscriber<T,E> {
- let subscriber = subscriberBlock()
-
+ func waitForNextState<T,E:Error>(
+ forSubscriber subscriber: Subscriber<T,E>,
+ timeout: TimeInterval = 0.01
+ ) {
+ var fulfilled = false
  let expectation = expectation(description: "\(subscriber) should emit within \(timeout) second\(timeout != 1 ? "s" : "").")
  subscriber.subscribe { _ in
+ guard fulfilled == false else {
+ return
+ }
  expectation.fulfill()
+ fulfilled = true
  }
 
  waitForExpectations(timeout: timeout)
+ }
+
+ func confirmNextStateDoesntTrigger<T,E:Error>(
+ forSubscriber sub: Subscriber<T,E>,
+ timeout: TimeInterval = 0.01
+ ) {
+ let start = Date()
+ let errorMessage = "Subscriber should not emit within \(timeout) seconds."
+ let expectation = expectation(description: errorMessage)
+ let timer = Timer.scheduledTimer(withTimeInterval: timeout - 0.00001, repeats: false) { timer in
+ expectation.fulfill()
+ timer.invalidate()
+ }
 
- return subscriber
+ sub.subscribe { _ in
+ if timer.isValid {
+ XCTFail(
+ "\(errorMessage) Emitted after \(Date().timeIntervalSince(start)) seconds."
+ )
+ }
+ }
+
+ waitForExpectations(timeout: timeout)
+ }
+
+ func waitForCompletion<T,E:Error>(of future: Future<T,E>, timeout: TimeInterval = 0.01) {
+ let expectation = expectation(description: "\(future) should complete within \(timeout) second\(timeout != 1 ? "s" : "").")
+ future.finally { _ in
+ expectation.fulfill()
+ }
+ waitForExpectations(timeout: timeout)
  }
 
 }