forked from CombineCommunity/CombineExt
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
1 parent
547d11a
commit 02d113a
Showing
3 changed files
with
227 additions
and
0 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,95 @@ | ||
| // | ||
| // RetryWhen.swift | ||
| // CombineExt | ||
| // | ||
| // Created by Daniel Tartaglia on 3/21/20. | ||
| // | ||
|
|
||
| import Combine | ||
|
|
||
| public extension Publisher { | ||
|
|
||
| /// Repeats the source publisher on error when the notifier emits a next value. If the source publisher errors and the notifier completes, it will complete the source sequence. | ||
| /// | ||
| /// - Parameter notificationHandler: A handler that is passed a publisher of errors raised by the source publisher and returns a publisher that either continues, completes or errors. This behavior is then applied to the source publisher. | ||
| /// - Returns: A publisher producing the elements of the given sequence repeatedly until it terminates successfully or is notified to error or complete. | ||
| func retryWhen<Trigger>(_ notificationHandler: @escaping (AnyPublisher<Self.Failure, Never>) -> Trigger) | ||
| -> Publishers.RetryWhen<Self, Trigger, Output, Failure> where Trigger: Publisher { | ||
| .init(upstream: self, notificationHandler: notificationHandler) | ||
| } | ||
| } | ||
|
|
||
| public extension Publishers { | ||
| class RetryWhen<Upstream, Trigger, Output, Failure>: Publisher where Upstream: Publisher, Upstream.Output == Output, Upstream.Failure == Failure, Trigger: Publisher { | ||
|
|
||
| typealias Handler = (AnyPublisher<Upstream.Failure, Never>) -> Trigger | ||
|
|
||
| private let upstream: Upstream | ||
| private let handler: Handler | ||
|
|
||
| init(upstream: Upstream, notificationHandler: @escaping Handler) { | ||
| self.upstream = upstream | ||
| self.handler = notificationHandler | ||
| } | ||
|
|
||
| public func receive<S>(subscriber: S) where S: Subscriber, Failure == S.Failure, Output == S.Input { | ||
| subscriber.receive(subscription: Subscription(upstream: upstream, downstream: subscriber, handler: handler)) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| extension Publishers.RetryWhen { | ||
| class Subscription<Downstream>: Combine.Subscription where Downstream: Subscriber, Downstream.Input == Upstream.Output, Downstream.Failure == Upstream.Failure { | ||
|
|
||
| private let upstream: Upstream | ||
| private let downstream: Downstream | ||
| private let errorSubject = PassthroughSubject<Upstream.Failure, Never>() | ||
| private var sink: Sink<Upstream, Downstream>? | ||
| private var cancellable: AnyCancellable? | ||
|
|
||
| init(upstream: Upstream, downstream: Downstream, handler: @escaping (AnyPublisher<Upstream.Failure, Never>) -> Trigger) { | ||
| self.upstream = upstream | ||
| self.downstream = downstream | ||
| self.sink = Sink( | ||
| downstream: downstream, | ||
| transformOutput: { $0 }, | ||
| transformFailure: { [errorSubject] in | ||
| errorSubject.send($0) | ||
| return nil | ||
| } | ||
| ) | ||
| self.cancellable = handler(errorSubject.eraseToAnyPublisher()) | ||
| .sink( | ||
| receiveCompletion: { [sink] completion in | ||
| switch completion { | ||
| case .finished: | ||
| sink?.buffer.complete(completion: Subscribers.Completion<Downstream.Failure>.finished) | ||
| case .failure(let error): | ||
| if let error = error as? Downstream.Failure { | ||
| sink?.buffer.complete(completion: Subscribers.Completion<Downstream.Failure>.failure(error)) | ||
| } | ||
| } | ||
| }, | ||
| receiveValue: { [upstream, sink] _ in | ||
| guard let sink = sink else { return } | ||
| upstream.subscribe(sink) | ||
| } | ||
| ) | ||
| upstream.subscribe(sink!) | ||
| } | ||
|
|
||
| func request(_ demand: Subscribers.Demand) { | ||
| sink?.demand(demand) | ||
| } | ||
|
|
||
| func cancel() { | ||
| sink = nil | ||
| } | ||
| } | ||
| } | ||
|
|
||
| extension Publishers.RetryWhen.Subscription: CustomStringConvertible { | ||
| var description: String { | ||
| return "RetryWhen.Subscription<\(Output.self), \(Failure.self)>" | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,124 @@ | ||
| // | ||
| // RetryWhenTests.swift | ||
| // CombineExtTests | ||
| // | ||
| // Created by Daniel Tartaglia on 8/28/21. | ||
| // | ||
|
|
||
| #if !os(watchOS) | ||
| import XCTest | ||
| import Combine | ||
| import CombineExt | ||
|
|
||
| @available(OSX 10.15, iOS 13.0, tvOS 13.0, watchOS 6.0, *) | ||
| class RetryWhenTests: XCTestCase { | ||
| var subscription: AnyCancellable! | ||
|
|
||
| func testPassthroughNextAndComplete() { | ||
| let source = PassthroughSubject<Int, MyError>() | ||
|
|
||
| var expectedOutput: Int? | ||
|
|
||
| var completion: Subscribers.Completion<MyError>? | ||
|
|
||
| subscription = source | ||
| .retryWhen { error in | ||
| error.filter { _ in false } | ||
| } | ||
| .sink( | ||
| receiveCompletion: { completion = $0 }, | ||
| receiveValue: { expectedOutput = $0 } | ||
| ) | ||
|
|
||
| source.send(2) | ||
| source.send(completion: .finished) | ||
|
|
||
| XCTAssertEqual( | ||
| expectedOutput, | ||
| 2 | ||
| ) | ||
| XCTAssertEqual(completion, .finished) | ||
|
|
||
| } | ||
|
|
||
| func testSuccessfulRetry() { | ||
| var times = 0 | ||
|
|
||
| var expectedOutput: Int? | ||
|
|
||
| var completion: Subscribers.Completion<RetryWhenTests.MyError>? | ||
|
|
||
| subscription = Deferred(createPublisher: { () -> AnyPublisher<Int, MyError> in | ||
| defer { times += 1 } | ||
| if times == 0 { | ||
| return Fail<Int, MyError>(error: MyError.someError).eraseToAnyPublisher() | ||
| } | ||
| else { | ||
| return Just(5).setFailureType(to: MyError.self).eraseToAnyPublisher() | ||
| } | ||
| }) | ||
| .retryWhen { error in | ||
| error.map { _ in } | ||
| } | ||
| .sink( | ||
| receiveCompletion: { completion = $0 }, | ||
| receiveValue: { expectedOutput = $0 } | ||
| ) | ||
|
|
||
| XCTAssertEqual( | ||
| expectedOutput, | ||
| 5 | ||
| ) | ||
| XCTAssertEqual(completion, .finished) | ||
| XCTAssertEqual(times, 2) | ||
| } | ||
|
|
||
| func testRetryFailure() { | ||
| var expectedOutput: Int? | ||
|
|
||
| var completion: Subscribers.Completion<RetryWhenTests.MyError>? | ||
|
|
||
| subscription = Fail<Int, MyError>(error: MyError.someError) | ||
| .retryWhen { error in | ||
| error.tryMap { _ in throw MyError.retryError } | ||
| } | ||
| .sink( | ||
| receiveCompletion: { completion = $0 }, | ||
| receiveValue: { expectedOutput = $0 } | ||
| ) | ||
|
|
||
| XCTAssertEqual( | ||
| expectedOutput, | ||
| nil | ||
| ) | ||
| XCTAssertEqual(completion, .failure(MyError.retryError)) | ||
| } | ||
|
|
||
| func testRetryComplete() { | ||
| var expectedOutput: Int? | ||
|
|
||
| var completion: Subscribers.Completion<RetryWhenTests.MyError>? | ||
|
|
||
| subscription = Fail<Int, MyError>(error: MyError.someError) | ||
| .retryWhen { error in | ||
| error.prefix(1) | ||
| } | ||
| .sink( | ||
| receiveCompletion: { completion = $0 }, | ||
| receiveValue: { expectedOutput = $0 } | ||
| ) | ||
|
|
||
| XCTAssertEqual( | ||
| expectedOutput, | ||
| nil | ||
| ) | ||
| XCTAssertEqual(completion, .finished) | ||
| } | ||
|
|
||
| enum MyError: Swift.Error { | ||
| case someError | ||
| case retryError | ||
| } | ||
|
|
||
| } | ||
| #endif |