apple / apple/swift-async-algorithms

Multi-consumption of an `AsyncSequence`

Open
#110 50 comments 62 reactions 0 assignees View on GitHub
Future
Dominant language
Swift
Stars
3.7k
Forks
226
Avg merge
10d 3h
Merged PRs (30d)
1

Description

Hi, I have been reading the documentation for this repository and it left me wondering: is this planned to introduce a feature to allow several tasks to consume the same `AsyncSequence`?

_If this is not the right place to offer ideas, please let me know. I don't think this would fit in a proposal._

## Context
I recently had to stream a file using `URLSession` and two pieces of code were interested in the streamed values. Rather than starting a stream twice which would not be efficient, I wanted to allow the stream to be consumed by several tasks. The data that were already streamed would be sent when a new consumption is set up.

## Basic implementation
I tried to implement such a solution (in a [gist](https://gist.github.com/ABridoux/913dfb95fabc22048b0a1c566059105a)) that will use a reduce function on the already emitted values and emit the result when a new task starts consuming the sequence. It's far from being perfect but I think it might help understanding the idea.

Implementation

```swift
struct ReducedReplayAsyncStream {

typealias Reduce = (_ partialResult: inout Element, _ nextResult: Element) -> Void

private let storage: _Storage
private var originalStream: AsyncStream

init(
bufferingPolicy limit: AsyncStream.Continuation.BufferingPolicy = .unbounded,
initialResult: Element,
reduce: @escaping Reduce,
build: (AsyncStream.Continuation) -> Void
) {
originalStream = AsyncStream(Element.self, bufferingPolicy: limit, build)
storage = _Storage(stored: initialResult, reduce: reduce)
}

private func makeStream() -> AsyncStream {
AsyncStream { continuation in
Task {
var isFirst = false
if await !storage.didStart {
await storage.setDidStart(true)
isFirst = true
startConsumingOriginalStream()
}

if !isFirst {
await continuation.yield(storage.stored)
}
await storage.appendContinuation(continuation)
}
}
}

private func startConsumingOriginalStream () {
Task {
for await value in originalStream {
await storage.updateWith(value: value)
}
await storage.continuations.forEach { $0.finish() }
}
}
}

extension ReducedReplayAsyncStream {

private actor _Storage {
private let reduce: ReducedReplayAsyncStream.Reduce

var didStart = false
var stored: Element
var continuations: [AsyncStream.Continuation] = []

init(stored: Element, reduce: @escaping Reduce) {
self.stored = stored
self.reduce = reduce
}

func updateWith(value: Element) {
reduce(&stored, value)
continuations.forEach { $0.yield(value) }
}

func setDidStart(_ value: Bool) {
didStart = value
}

func appendContinuation(_ continuation: AsyncStream.Continuation) {
continuations.append(continuation)
}
}
}

extension ReducedReplayAsyncStream: AsyncSequence {
typealias AsyncIterator = AsyncStream.AsyncIterator

func makeAsyncIterator() -> AsyncIterator {
let stream = makeStream()
return stream.makeAsyncIterator()
}
}
```

Usage

```swift
var subscriptions: Set = []
var continuation: Stream.Continuation!

let replayStream = ReducedReplayAsyncStream(
initialResult: 0,
reduce: { partialResult, nextResult in partialResult = partialResult + nextResult },
build: { continuation = $0 }
)

var counter = 0
Timer.publish(every: 0.4, on: .main, in: .default)
.autoconnect()
.sink { _ in
if counter == 10 {
continuation.finish()
}
continuation.yield(counter)
counter += 1
}
.store(in: &subscriptions)

Task {
for await value in replayStream {
print("[A]", value)
}
}

Task {
try await Task.sleep(nanoseconds: 3_000_000_000)
for await value in replayStream {
print("[B]", value)
}
}
```

Some considerations about efficiency can be found in the gist.

Contributor guide

Open the contributing guide

Research direction

Start by reading the AsyncSequence and AsyncStream usage shown in the issue, then compare the linked ReducedReplayAsyncStream gist with the stated multi-consumption and replay requirements. Review the existing comment discussion before deciding whether the project wants this feature; done would require an agreed scope and design rather than a small isolated edit.

Written by the indexing model from the issue text.

Assessment

Tech stack
swift
Domain
api
Issue type
Feature
Difficulty
5/5
Estimated time
Over a week
Activity status
Stale
Clarity
Needs clarification
Newbie friendliness
20/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.