Skip to content

Commit

Permalink
operators: add fromAsync, fromThrowingAsync, fromAsyncSequence to bri…
Browse files Browse the repository at this point in the history 
…dge async/await with Combine
  • Loading branch information
Thibault Wittemberg committed Sep 26, 2021
1 parent fc3e405 commit 8c97c8c
Show file tree
Hide file tree
Showing 5 changed files with 609 additions and 24 deletions.
36 changes: 13 additions & 23 deletions .github/workflows/tests.yml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -5,32 +5,31 @@ on: [push, pull_request, workflow_dispatch]
jobs:
xcode-tests:
name: "Test"
runs-on: macOS-latest
runs-on: macOS-11

strategy:
matrix:
platform: [macOS, iOS, tvOS]
# platform: [macOS, iOS, tvOS]
platform: [iOS, tvOS]
include:
- platform: macOS
sdk: macosx
destination: "arch=x86_64"

# - platform: macOS
# sdk: macosx11.3
# destination: "arch=x86_64"
#
- platform: iOS
sdk: iphonesimulator
destination: "name=iPhone 11"
sdk: iphoneos15.0
destination: "name=iPhone 13"

- platform: tvOS
sdk: appletvsimulator
sdk: appletvsimulator15.0
destination: "name=Apple TV"

steps:
- uses: actions/checkout@v2
- name: Select Xcode 12 (beta)
run: sudo xcode-select -s /Applications/Xcode_12_beta.app
- name: Generate project
run: make project
- name: Select Xcode 13
run: sudo xcode-select -s /Applications/Xcode_13.0.app
- name: Run tests
run: set -o pipefail && xcodebuild -project CombineExt.xcodeproj -scheme CombineExt-Package -enableCodeCoverage YES -sdk ${{ matrix.sdk }} -destination "${{ matrix.destination }}" test | xcpretty -c -r html --output logs/${{ matrix.platform }}.html
run: set -o pipefail && xcodebuild -scheme CombineExt-Package -enableCodeCoverage YES -sdk ${{ matrix.sdk }} -destination "${{ matrix.destination }}" test | xcpretty -c -r html --output logs/${{ matrix.platform }}.html
- uses: codecov/[email protected]
with:
token: 1519d58c-6fb9-483f-af6c-7f6f0b384345
Expand All @@ -39,12 +38,3 @@ jobs:
with:
name: build-logs-${{ github.run_id }}
path: logs

SPM:
name: "Test (SPM)"
runs-on: macOS-latest

steps:
- uses: actions/checkout@v2
- name: Run tests
run: set -o pipefail && swift test
2 changes: 1 addition & 1 deletion CombineExt.podspec
View file
Original file line number Diff line number Diff line change
Expand Up @@ -18,5 +18,5 @@ Pod::Spec.new do |s|
s.source = { :git => "https://github.com/CombineCommunity/CombineExt.git", :tag => s.version }
s.source_files = 'Sources/**/*.swift'
s.frameworks = ['Combine']
s.swift_version = '5.1'
s.swift_version = '5.5'
end
95 changes: 95 additions & 0 deletions README.md
View file
Original file line number Diff line number Diff line change
Expand Up @@ -755,6 +755,101 @@ subscription = ints
.finished
```

------

### fromAsync(priority:_:)

Creates a Combine Publisher from an async function. The Publisher emits a value and then completes when the async function returns its result.
The task that supports the async function is canceled when the publisher's subscription is canceled.
An optional priority indicates the priority of the Task supporting the execution of the async function.

```swift
var value: Int {
get async {
3
}
}

Publishers
.fromAsync {
await value
}.sink {
print($0)
} receiveValue: {
print($0)
}
```

#### Output:

```none
3
finished
```

------

### fromThrowingAsync(priority:_:)

Creates a Combine Publisher from a throwing async function
The Publisher emits a value and completes or fail according the the async function execution result.
The task that supports the async function is canceled when the publisher's subscription is canceled.
An optional priority indicates the priority of the Task supporting the execution of the async function.

```swift
struct MyError: Error, CustomStringConvertible {
var description: String {
"Async Error"
}
}

Publishers
.fromAsync { () async throws -> String in
throw MyError()
}.sink {
print($0)
} receiveValue: {
print($0)
}
```

#### Output:

```none
failure(Async Error)
```

### fromAsyncSequence(priority:_:)

Creates a Combine Publisher from an async sequence.
The Publisher emits values or fail according the the async sequence execution result.
An optional priority indicates the priority of the Task supporting the execution of the async sequence.

```swift
let sequence = AsyncStream(Int.self) { continuation in
continuation.yield(1)
continuation.yield(2)
continuation.yield(3)
continuation.finish()
}

Publishers
.fromAsyncSequence(sequence).sink {
print($0)
} receiveValue: {
print($0)
}
```

#### Output:

```none
1
2
3
finished
```

## Publishers

This section outlines some of the custom Combine publishers CombineExt provides
Expand Down
191 changes: 191 additions & 0 deletions Sources/Operators/FromAsync.swift
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,191 @@
//
// FromAsync.swift
// CombineExt
//
// Created by Thibault Wittemberg on 2021-06-15.
// Copyright © 2021 Combine Community. All rights reserved.
//

#if canImport(Combine)
import Combine

@available(macOS 12.0, iOS 15.0, tvOS 15.0, watchOS 8.0, *)
public extension Publishers {
/// Creates a Combine Publisher from an async function
/// The Publisher emits a value and then completes when the async function returns its result.
/// The task that supports the async function is canceled when the publisher's subscription is canceled.
/// ```
/// var value: Int {
/// get async {
/// 3
/// }
/// }
///
/// Publishers
/// .fromAsync {
/// await value
/// }.sink {
/// print($0)
/// } receiveValue: {
/// print($0)
/// }
///
/// // will print:
/// // 3
/// // finished
/// ```
/// - parameter priority: Optional value indicating the priority of the Task supporting the execution of the async function
/// - Returns: The Combine Publisher wrapping the async function execution
static func fromAsync<Output>(priority: TaskPriority? = nil,
_ asyncFunction: @escaping () async -> Output) -> AnyPublisher<Output, Never> {
AnyPublisher<Output, Never>.create { subscriber in
let task = Task(priority: priority) {
let result = await asyncFunction()
subscriber.send(result)
subscriber.send(completion: .finished)
}

return AnyCancellable {
task.cancel()
}
}
}

/// Creates a Combine Publisher from a throwing async function
/// The Publisher emits a value or fail according the the async function execution result.
/// The task that supports the async function is canceled when the publisher's subscription is canceled.
///
/// ```
/// var value: Int {
/// get async {
/// 3
/// }
/// }
///
/// Publishers
/// .fromAsync {
/// await value
/// }.sink {
/// print($0)
/// } receiveValue: {
/// print($0)
/// }
///
/// // will print:
/// // 3
/// // finished
/// ```
///
/// Whenever the async function throws an error, the stream will faile:
///
/// ```
/// struct MyError: Error, CustomStringConvertible {
/// var description: String {
/// "Async Error"
/// }
/// }
///
/// Publishers
/// .fromAsync { () async throws -> String in
/// throw MyError()
/// }.sink {
/// print($0)
/// } receiveValue: {
/// print($0)
/// }
///
/// // will print:
/// // failure(Async Error)
///```
/// - parameter priority: Optional value indicating the priority of the Task supporting the execution of the async function
/// - Returns: The Combine Publisher wrapping the async function execution
static func fromThrowingAsync<Output>(priority: TaskPriority? = nil,
_ asyncThrowingFunction: @escaping () async throws -> Output) -> AnyPublisher<Output, Error> {
AnyPublisher<Output, Error>.create { subscriber in
let task = Task(priority: priority) {
do {
let result = try await asyncThrowingFunction()
subscriber.send(result)
subscriber.send(completion: .finished)
} catch {
subscriber.send(completion: .failure(error))
}
}

return AnyCancellable {
task.cancel()
}
}
}

/// Creates a Combine Publisher from an async sequence.
/// The Publisher emits values or fail according the the async sequence execution result.
///
/// ```
/// let sequence = [1, 2, 3].publisher.values
///
/// Publishers
/// .fromAsyncSequence(sequence).sink {
/// print($0)
/// } receiveValue: {
/// print($0)
/// }
///
/// // will print:
/// // 1
/// // 2
/// // 3
/// // finished
/// ```
///
/// If the asyncSequence faild:
///
/// ```
/// struct MyError: Error, CustomStringConvertible {
/// var description: String {
/// "Async Error"
/// }
/// }
///
/// let sequence = AsyncThrowingStream(Int.self) { continuation in
/// continuation.yield(1)
/// continuation.yield(2)
/// continuation.finish(throwing: MockError(value: Int.random(in: 1...100)))
/// }
///
/// Publishers
/// .fromAsyncSequence(sequence).sink {
/// print($0)
/// } receiveValue: {
/// print($0)
/// }
///
/// // will print:
/// // 1
/// // 2
/// // failure(Async Error)
///```
/// - parameter priority: Optional value indicating the priority of the Task supporting the async sequence execution
/// - Returns: The Combine Publisher wrapping the async sequence iteration
static func fromAsyncSequence<Output, AsyncSequenceType>(priority: TaskPriority? = nil,
_ asyncSequence: AsyncSequenceType) -> AnyPublisher<Output, Error>
where AsyncSequenceType: AsyncSequence, AsyncSequenceType.Element == Output {
AnyPublisher<Output, Error>.create { subscriber in
let task = Task(priority: priority) {
do {
for try await result in asyncSequence {
subscriber.send(result)
}
subscriber.send(completion: .finished)
} catch {
subscriber.send(completion: .failure(error))
}
}

return AnyCancellable {
task.cancel()
}
}
}
}
#endif
Loading

0 comments on commit 8c97c8c

Please sign in to comment.