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
Showing
6 changed files
with
255 additions
and
10 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
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,80 @@ | ||
| // | ||
| // Dematerialize.swift | ||
| // CombineExt | ||
| // | ||
| // Created by Shai Mishali on 14/03/2020. | ||
| // | ||
|
|
||
| import Combine | ||
|
|
||
| public extension Publisher where Output: EventConvertible, Failure == Never { | ||
| /// Converts any previously-materialized publisher into its original form | ||
| /// | ||
| /// - returns: A publisher dematerializing the materialized events | ||
| func dematerialize() -> Publishers.Dematerialize<Self> { | ||
| Publishers.Dematerialize(upstream: self) | ||
| } | ||
| } | ||
|
|
||
| // MARK: - Publisher | ||
| public extension Publishers { | ||
| /// A publisher which takes a materialized upstream publisher and converts | ||
| /// the wrapped events back into their original form | ||
| class Dematerialize<Upstream: Publisher>: Publisher where Upstream.Output: EventConvertible { | ||
| public typealias Output = Upstream.Output.Output | ||
| public typealias Failure = Upstream.Output.Failure | ||
|
|
||
| private let upstream: Upstream | ||
|
|
||
| public init(upstream: Upstream) { | ||
| self.upstream = upstream | ||
| } | ||
|
|
||
| public func receive<S: Subscriber>(subscriber: S) where Failure == S.Failure, Output == S.Input { | ||
| subscriber.receive(subscription: Subscription(upstream: upstream, downstream: subscriber)) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| // MARK: - Subscrription | ||
| private extension Publishers.Dematerialize { | ||
| class Subscription<Downstream: Subscriber>: Combine.Subscription | ||
| where Downstream.Input == Upstream.Output.Output, Downstream.Failure == Upstream.Output.Failure { | ||
| private var sink: Sink<Downstream>? | ||
|
|
||
| init(upstream: Upstream, | ||
| downstream: Downstream) { | ||
| self.sink = Sink(upstream: upstream, | ||
| downstream: downstream) | ||
| } | ||
|
|
||
| func request(_ demand: Subscribers.Demand) { | ||
| sink?.demand(demand) | ||
| } | ||
|
|
||
| func cancel() { | ||
| sink = nil | ||
| } | ||
| } | ||
| } | ||
|
|
||
| // MARK: - Sink | ||
| private extension Publishers.Dematerialize { | ||
| class Sink<Downstream: Subscriber>: CombineExt.Sink<Upstream, Downstream> | ||
| where Downstream.Input == Upstream.Output.Output, Downstream.Failure == Upstream.Output.Failure { | ||
| override func receive(_ input: Upstream.Output) -> Subscribers.Demand { | ||
| /// We have to override the default mechanism here to convert a | ||
| /// materialized failure into an actual failure | ||
| switch input.event { | ||
| case .value(let value): | ||
| return buffer.buffer(value: value) | ||
| case .failure(let failure): | ||
| buffer.complete(completion: .failure(failure)) | ||
| return .none | ||
| case .finished: | ||
| buffer.complete(completion: .finished) | ||
| return .none | ||
| } | ||
| } | ||
| } | ||
| } |
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
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,134 @@ | ||
| // | ||
| // DematerializeTests.swift | ||
| // CombineExtTests | ||
| // | ||
| // Created by Shai Mishali on 14/03/2020. | ||
| // | ||
|
|
||
| import XCTest | ||
| import Combine | ||
| import CombineExt | ||
|
|
||
| class DematerializeTests: XCTestCase { | ||
| var subscription: AnyCancellable? | ||
| var values = [String]() | ||
| var completion: Subscribers.Completion<MyError>? | ||
| var subject = PassthroughSubject<Event<String, MyError>, Never>() | ||
|
|
||
| override func setUp() { | ||
| values = [] | ||
| completion = nil | ||
| subject = PassthroughSubject<Event<String, MyError>, Never>() | ||
| } | ||
|
|
||
| override func tearDown() { | ||
| subscription?.cancel() | ||
| } | ||
|
|
||
| enum MyError: Swift.Error { | ||
| case someError | ||
| } | ||
|
|
||
| func testEmpty() { | ||
| subscription = subject | ||
| .dematerialize() | ||
| .sink(receiveCompletion: { self.completion = $0 }, | ||
| receiveValue: { self.values.append($0) }) | ||
|
|
||
| subject.send(.finished) | ||
|
|
||
| XCTAssertTrue(values.isEmpty) | ||
| XCTAssertEqual(completion, .finished) | ||
| } | ||
|
|
||
| func testFail() { | ||
| subscription = subject | ||
| .dematerialize() | ||
| .sink(receiveCompletion: { self.completion = $0 }, | ||
| receiveValue: { self.values.append($0) }) | ||
|
|
||
| subject.send(.failure(.someError)) | ||
|
|
||
| XCTAssertTrue(values.isEmpty) | ||
| XCTAssertEqual(completion, .failure(.someError)) | ||
| } | ||
|
|
||
| func testFinished() { | ||
| subscription = subject | ||
| .dematerialize() | ||
| .sink(receiveCompletion: { self.completion = $0 }, | ||
| receiveValue: { self.values.append($0) }) | ||
|
|
||
| subject.send(.value("Hello")) | ||
| subject.send(.value("There")) | ||
| subject.send(.value("World!")) | ||
| subject.send(.finished) | ||
|
|
||
| XCTAssertEqual(values, ["Hello", "There", "World!"]) | ||
| XCTAssertEqual(completion, .finished) | ||
| } | ||
|
|
||
| func testFinishedLimitedDemand() { | ||
| let subscriber = makeSubscriber(demand: .max(2), expectation: nil) | ||
|
|
||
| subject | ||
| .dematerialize() | ||
| .subscribe(subscriber) | ||
|
|
||
| subject.send(.value("Hello")) | ||
| subject.send(.value("There")) | ||
| subject.send(.value("World!")) | ||
| subject.send(.finished) | ||
|
|
||
| XCTAssertEqual(values, ["Hello", "There"]) | ||
| XCTAssertEqual(completion, nil) | ||
| } | ||
|
|
||
| func testError() { | ||
| subscription = subject | ||
| .dematerialize() | ||
| .sink(receiveCompletion: { self.completion = $0 }, | ||
| receiveValue: { self.values.append($0) }) | ||
|
|
||
| subject.send(.value("Hello")) | ||
| subject.send(.value("There")) | ||
| subject.send(.value("World!")) | ||
| subject.send(.failure(.someError)) | ||
|
|
||
| XCTAssertEqual(values, ["Hello", "There", "World!"]) | ||
| XCTAssertEqual(completion, .failure(.someError)) | ||
| } | ||
|
|
||
| func testErrorLimitedDemand() { | ||
| let subscriber = makeSubscriber(demand: .max(2), expectation: nil) | ||
|
|
||
| subject | ||
| .dematerialize() | ||
| .subscribe(subscriber) | ||
|
|
||
| subject.send(.value("Hello")) | ||
| subject.send(.value("There")) | ||
| subject.send(.value("World!")) | ||
| subject.send(.failure(.someError)) | ||
|
|
||
| XCTAssertEqual(values, ["Hello", "There"]) | ||
| XCTAssertEqual(completion, nil) | ||
| } | ||
| } | ||
|
|
||
| private extension DematerializeTests { | ||
| func makeSubscriber(demand: Subscribers.Demand, expectation: XCTestExpectation?) -> AnySubscriber<String, MyError> { | ||
| return AnySubscriber( | ||
| receiveSubscription: { subscription in | ||
| subscription.request(demand) | ||
| }, | ||
| receiveValue: { value in | ||
| self.values.append(value) | ||
| return .none | ||
| }, | ||
| receiveCompletion: { finished in | ||
| self.completion = finished | ||
| expectation?.fulfill() | ||
| }) | ||
| } | ||
| } |