Asynchrone

Extensions and additions for Swift's async sequence.

Requirements

iOS 14.0+
macOS 12.0+
watchOS 6.0+
tvOS 14.0+

Installation

Swift Package Manager

In Xcode:

Click Project.
Click Package Dependencies.
Click +.
Enter package URL: https://github.com/reddavis/Asynchrone.
Add Asynchrone to your app target.

Documentation

Documentation can be found here.

Overview

AsyncSequence

Extensions

Assign

class MyClass {
    var value: Int = 0 {
        didSet { print("Set to \(self.value)") }
    }
}


let sequence = AsyncStream<Int> { continuation in
    continuation.yield(1)
    continuation.yield(2)
    continuation.yield(3)
    continuation.finish()
}

let object = MyClass()
sequence.assign(to: \.value, on: object)

// Prints:
// Set to 1
// Set to 2
// Set to 3

First

let sequence = AsyncStream<Int> { continuation in
    continuation.yield(1)
    continuation.yield(2)
    continuation.yield(3)
    continuation.finish()
}

print(await sequence.first())

// Prints:
// 1

Last

let sequence = AsyncStream<Int> { continuation in
    continuation.yield(1)
    continuation.yield(2)
    continuation.yield(3)
    continuation.finish()
}

print(await sequence.last())

// Prints:
// 3

Collect

let sequence = AsyncStream<Int> { continuation in
    continuation.yield(1)
    continuation.yield(2)
    continuation.yield(3)
    continuation.finish()
}

print(await sequence.collect())

// Prints:
// [1, 2, 3]

Sink

let sequence = .init { continuation in
    continuation.yield(1)
    continuation.yield(2)
    continuation.yield(3)
    continuation.finish()
}

sequence.sink { print($0) }

// Prints:
// 1
// 2
// 3

Sink with completion

let sequence = .init { continuation in
    continuation.yield(1)
    continuation.yield(2)
    continuation.yield(3)
    continuation.finish(throwing: TestError())
}

sequence.sink(
    receiveValue: { print("Value: \($0)") },
    receiveCompletion: { print("Complete: \($0)") }
)

// Prints:
// Value: 1
// Value: 2
// Value: 3
// Complete: failure(TestError())

AnyAsyncSequenceable

let sequence = Just(1)
    .map(String.init)
    .eraseToAnyAsyncSequenceable()

AnyThrowingAsyncSequenceable

let stream = Fail<Int, TestError>(error: TestError.a)
    .eraseToAnyThrowingAsyncSequenceable()

CatchErrorAsyncSequence

let sequence = Fail<Int, TestError>(
    error: TestError()
)
.catch { error in
    Just(-1)
}

for await value in sequence {
    print(value)
}

// Prints:
// -1

ChainAsyncSequence

let sequenceA = AsyncStream<Int> { continuation in
    continuation.yield(1)
    continuation.yield(2)
    continuation.yield(3)
    continuation.finish()
}

let sequenceB = AsyncStream<Int> { continuation in
    continuation.yield(4)
    continuation.yield(5)
    continuation.yield(6)
    continuation.finish()
}

let sequenceC = AsyncStream<Int> { continuation in
    continuation.yield(7)
    continuation.yield(8)
    continuation.yield(9)
    continuation.finish()
}

for await value in sequenceA.chain(with: sequenceB).chain(with: sequenceC) {
    print(value)
}

// Prints:
// 1
// 2
// 3
// 4
// 5
// 6
// 7
// 8
// 9

CombineLatestAsyncSequence

let streamA = .init { continuation in
    continuation.yield(1)
    continuation.yield(2)
    continuation.yield(3)
    continuation.yield(4)
    continuation.finish()
}

let streamB = .init { continuation in
    continuation.yield(5)
    continuation.yield(6)
    continuation.yield(7)
    continuation.yield(8)
    continuation.yield(9)
    continuation.finish()
}

for await value in streamA.combineLatest(streamB) {
    print(value)
}

// Prints:
// (1, 5)
// (2, 6)
// (3, 7)
// (4, 8)
// (4, 9)

CombineLatest3AsyncSequence

let streamA = .init { continuation in
    continuation.yield(1)
    continuation.yield(2)
    continuation.yield(3)
    continuation.yield(4)
    continuation.finish()
}

let streamB = .init { continuation in
    continuation.yield(5)
    continuation.yield(6)
    continuation.yield(7)
    continuation.yield(8)
    continuation.yield(9)
    continuation.finish()
}

let streamC = .init { continuation in
    continuation.yield(10)
    continuation.yield(11)
    continuation.finish()
}

for await value in streamA.combineLatest(streamB, streamC) {
    print(value)
}

// Prints:
// (1, 5, 10)
// (2, 6, 11)
// (3, 7, 11)
// (4, 8, 11)
// (4, 9, 11)

CurrentElementAsyncSequence

let sequence = CurrentElementAsyncSequence(0)
print(await sequence.element)

await stream.yield(1)
print(await sequence.element)

await stream.yield(2)
await stream.yield(3)
await stream.yield(4)
print(await sequence.element)

// Prints:
// 0
// 1
// 4

DebounceAsyncSequence

let stream = AsyncStream<Int> { continuation in
    continuation.yield(0)
    try? await Task.sleep(nanoseconds: 200_000_000)
    continuation.yield(1)
    try? await Task.sleep(nanoseconds: 200_000_000)
    continuation.yield(2)
    continuation.yield(3)
    continuation.yield(4)
    continuation.yield(5)
    continuation.finish()
}

for element in try await self.stream.debounce(for: 0.1) {
    print(element)
}

// Prints:
// 0
// 1
// 5

DelayAsyncSequence

let stream = AsyncStream<Int> { continuation in
    continuation.yield(0)
    continuation.yield(1)
    continuation.yield(2)
    continuation.finish()
}

let start = Date.now
for element in try await self.stream.delay(for: 0.5) {
    print("\(element) - \(Date.now.timeIntervalSince(start))")
}

// Prints:
// 0 - 0.5
// 1 - 1.0
// 2 - 1.5
>>>>>>> main

Empty

Empty<Int>().sink(
    receiveValue: { print($0) },
    receiveCompletion: { completion in
        switch completion {
        case .finished:
            print("Finished")
        case .failure:
            print("Failed")
        }
    }
)

// Prints:
// Finished

Fail

let stream = Fail<Int, TestError>(error: TestError())

do {
    for try await value in stream {
        print(value)
    }
} catch {
    print("Error!")
}

// Prints:
// Error!

Just

let stream = Just(1)

for await value in stream {
    print(value)
}

// Prints:
// 1

MergeAsyncSequence

let streamA = .init { continuation in
    continuation.yield(1)
    continuation.yield(2)
    continuation.yield(3)
    continuation.yield(4)
    continuation.finish()
}

let streamB = .init { continuation in
    continuation.yield(5)
    continuation.yield(6)
    continuation.yield(7)
    continuation.yield(8)
    continuation.yield(9)
    continuation.finish()
}

for await value in streamA.merge(with: streamB) {
    print(value)
}

// Prints:
// 1
// 5
// 2
// 6
// 3
// 7
// 4
// 8
// 9

Merge3AsyncSequence

let streamA = .init { continuation in
    continuation.yield(1)
    continuation.yield(4)
    continuation.finish()
}

let streamB = .init { continuation in
    continuation.yield(2)
    continuation.finish()
}

let streamC = .init { continuation in
    continuation.yield(3)
    continuation.finish()
}

for await value in self.streamA.merge(with: self.streamB, self.streamC) {
    print(value)
}

// Prints:
// 1
// 2
// 3
// 4

NotificationCenterAsyncSequence

let sequence = NotificationCenter.default.sequence(for: UIDevice.orientationDidChangeNotification)

for await element in sequence {
    print(element)
}

PassthroughAsyncSequence

let sequence = PassthroughAsyncSequence<Int>()
sequence.yield(0)
sequence.yield(1)
sequence.yield(2)
sequence.finish()

for await value in sequence {
    print(value)
}

// Prints:
// 0
// 1
// 2

RemoveDuplicatesAsyncSequence

let stream = .init { continuation in
    continuation.yield(1)
    continuation.yield(1)
    continuation.yield(2)
    continuation.yield(3)
    continuation.finish()
}

for await value in stream.removeDuplicates() {
    print(value)
}

// Prints:
// 1
// 2
// 3

ReplaceErrorAsyncSequence

let sequence = Fail<Int, TestError>(
    error: TestError()
)
.replaceError(with: 0)

for await value in stream {
    print(value)
}

// Prints:
// 0

SequenceAsyncSequence

let sequence = [0, 1, 2, 3].async

for await value in sequence {
    print(value)
}

// Prints:
// 1
// 2
// 3

SharedAsyncSequence

let values = [
    "a",
    "ab",
    "abc",
    "abcd"
]

let stream = AsyncStream { continuation in
    for value in values {
        continuation.yield(value)
    }
    continuation.finish()
}
.shared()

Task {
    let values = try await self.stream.collect()
    // ...
}

Task.detached {
    let values = try await self.stream.collect()
    // ...
}

let values = try await self.stream.collect()
// ...

ThrottleAsyncSequence

let stream = AsyncStream<Int> { continuation in
    continuation.yield(0)
    try? await Task.sleep(nanoseconds: 100_000_000)
    continuation.yield(1)
    try? await Task.sleep(nanoseconds: 100_000_000)
    continuation.yield(2)
    continuation.yield(3)
    continuation.yield(4)
    continuation.yield(5)
    continuation.finish()
}

for element in try await self.stream.throttle(for: 0.05, latest: true) {
    print(element)
}

// Prints:
// 0
// 1
// 2
// 5

ThrowingPassthroughAsyncSequence

let sequence = ThrowingPassthroughAsyncSequence<Int>()
sequence.yield(0)
sequence.yield(1)
sequence.yield(2)
sequence.finish(throwing: TestError())

do {
    for try await value in sequence {
      print(value)
    }
} catch {
    print("Error!")
}

// Prints:
// 0
// 1
// 2
// Error!

TimerAsyncSequence

let sequence = TimerAsyncSequence(interval: 1)

let start = Date.now
for element in await sequence {
    print(element)
}

// Prints:
// 2022-03-19 20:49:30 +0000
// 2022-03-19 20:49:31 +0000
// 2022-03-19 20:49:32 +0000

ZipAsyncSequence

let streamA = .init { continuation in
    continuation.yield(1)
    continuation.yield(2)
    continuation.finish()
}

let streamB = .init { continuation in
    continuation.yield(5)
    continuation.yield(6)
    continuation.yield(7)
    continuation.finish()
}

for await value in streamA.zip(streamB) {
    print(value)
}

// Prints:
// (1, 5)
// (2, 6)

Zip3AsyncSequence

let streamA = .init { continuation in
    continuation.yield(1)
    continuation.yield(2)
    continuation.finish()
}

let streamB = .init { continuation in
    continuation.yield(5)
    continuation.yield(6)
    continuation.yield(7)
    continuation.finish()
}

let streamC = .init { continuation in
    continuation.yield(8)
    continuation.yield(9)
    continuation.finish()
}

for await value in streamA.zip(streamB, streamC) {
    print(value)
}

// Prints:
// (1, 5, 8)
// (2, 6, 9)

Other libraries

Papyrus - Papyrus aims to hit the sweet spot between saving raw API responses to the file system and a fully fledged database like Realm.
Validate - A property wrapper that can validate the property it wraps.
Kyu - A persistent queue system in Swift.
FloatingLabelTextFieldStyle - A floating label style for SwiftUI's TextField.
Panel - A panel component similar to the iOS Airpod battery panel.

CombineLatest does not combine all values

Hello!

First: thanks for your library.

Coming from Combine, we've stumbled upon some unexpected behaviour when using CombineLatest:
StreamA's output is only processed if StreamB also provides further output.
Or if streams finish.

Expected behaviour:
Each output of both streams gets combined with each other as soon as output from either stream arrives.

Code for issue reproduction

import Asynchrone

var continuationA: AsyncStream<Int>.Continuation! = nil
var continuationB: AsyncStream<Int>.Continuation! = nil
let streamA = AsyncStream(/*bufferingPolicy: .bufferingNewest(1)*/) { continuation in
    continuationA = continuation
}

let streamB = AsyncStream(/*bufferingPolicy: .bufferingNewest(1)*/) { continuation in
    continuationB = continuation
}

let x = Task.detached(priority: .background) {
    print("START")
    for try await value in streamA.combineLatest(streamB) {
        print(value)
    }
    print("STOP")
}

Task.detached(priority: .userInitiated) {
    try! await Task.sleep(for: .seconds(1))
    
    print("A => 10")
    await continuationA.yield(10)
    
    print("B => 1")
    await continuationB.yield(1)
    
    print("A => 20")
    await continuationA.yield(20)
    print("A => 30")
    await continuationA.yield(30)
    print("A => 40")
    await continuationA.yield(40)
    
    print("B => 2")
    await continuationB.yield(2)
    
    print("B => 3")
    await continuationB.yield(3)

    print("A => 50")
    await continuationA.yield(50)
    
    print("B => 4")
    await continuationB.yield(4)
    
    print("A => finish")
    await continuationA.finish()
    
    print("B => finish")
    await continuationB.finish()
}

try! await x.value

Program's output

START
A => 10
B => 1
A => 20
(10, 1)
A => 30
A => 40
B => 2
B => 3
(20, 2)
A => 50
(30, 3)
B => 4
A => finish
(40, 4)
B => finish
(50, 4)
STOP

Expected output

START
A => 10
B => 1
A => 20
(10, 1)
(20, 1)
A => 30
(30, 1)
A => 40
(40, 1)
B => 2
(40, 2)
B => 3
(40, 3)
A => 50
(50, 3)
B => 4
(50, 4)
A => finish
B => finish
STOP

reddavis / asynchrone Goto Github PK

asynchrone's Introduction

Asynchrone

Requirements

Installation

Swift Package Manager

Documentation

Overview

AsyncSequence

Extensions

Assign

First

Last

Collect

Sink

Sink with completion

Other libraries

asynchrone's People

Contributors

Stargazers

Watchers

Forkers

asynchrone's Issues

Code for issue reproduction

Program's output

Expected output

Recommend Projects

Recommend Topics

Recommend Org