AsyncStreams & AsyncSequence

What is AsyncStream?

AsyncStream is a type that produces an asynchronous sequence of values. It bridges callback-based APIs to the structured concurrency world.

Creating an AsyncStream

Use the AsyncStream initializer to create a stream from a continuation.

func numberStream() -> AsyncStream<Int> {
    AsyncStream { continuation in
        Task {
            for i in 1...10 {
                continuation.yield(i)
                try? await Task.sleep(for: .seconds(1))
            }
            continuation.finish()
        }
    }
}

// Consume the stream
Task {
    for await number in numberStream() {
        print(number)
    }
}

Using AsyncStream.Continuation

The continuation allows you to yield values and control the stream.

func sensorStream() -> AsyncStream<Double> {
    AsyncStream { continuation in
        let sensor = TemperatureSensor()
        
        sensor.onUpdate = { temperature in
            continuation.yield(temperature)
        }
        
        continuation.onTermination = { _ in
            sensor.stop()
        }
    }
}

Buffering and Backpressure

AsyncStream supports buffering to handle fast producers and slow consumers.

let stream = AsyncStream<Int>(bufferingPolicy: .bufferingNewest(5)) { continuation in
    for i in 0..<100 {
        continuation.yield(i)
    }
    continuation.finish()
}

for await value in stream {
    try await Task.sleep(for: .milliseconds(100))
    print(value) // Only receives last 5 values if consumer is slow
}

Custom AsyncSequence

Create your own async sequence by conforming to the AsyncSequence protocol.

struct FibonacciSequence: AsyncSequence {
    typealias Element = Int
    
    struct AsyncIterator: AsyncIteratorProtocol {
        var a = 0
        var b = 1
        let maxCount: Int
        var currentCount = 0
        
        mutating func next() async throws -> Int? {
            guard currentCount < maxCount else { return nil }
            
            let result = a
            let next = a + b
            a = b
            b = next
            currentCount += 1
            
            try await Task.sleep(for: .milliseconds(100))
            return result
        }
    }
    
    let maxCount: Int
    
    func makeAsyncIterator() -> AsyncIterator {
        AsyncIterator(maxCount: maxCount)
    }
}

// Usage
Task {
    for try await fib in FibonacciSequence(maxCount: 10) {
        print(fib)
    }
}

AsyncStream from Callbacks

Wrap callback-based APIs into async sequences.

func locationUpdates() -> AsyncStream<CLLocationCoordinate2D> {
    AsyncStream { continuation in
        let manager = CLLocationManager()
        let delegate = LocationDelegate { location in
            continuation.yield(location)
        }
        manager.delegate = delegate
        manager.startUpdatingLocation()
        
        continuation.onTermination = { _ in
            manager.stopUpdatingLocation()
        }
        
        // Keep delegate alive
        withUnsafeMutablePointer(to: &delegate) { _ in }
    }
}

Merging and Combining Streams

Combine multiple async sequences using Swift concurrency.

func mergedStream() -> AsyncStream<String> {
    AsyncStream { continuation in
        Task {
            async let stream1 = fetchFromAPI1()
            async let stream2 = fetchFromAPI2()
            
            for await value in stream1 {
                continuation.yield("API1: \(value)")
            }
            for await value in stream2 {
                continuation.yield("API2: \(value)")
            }
            continuation.finish()
        }
    }
}

Consuming AsyncStreams in Views

Use task modifier to consume async sequences in SwiftUI views.

struct SensorView: View {
    @State private var temperature: Double = 0
    
    var body: some View {
        Text("Temperature: \(temperature, specifier: "%.1f") degrees")
            .task {
                for await temp in temperatureStream() {
                    temperature = temp
                }
            }
    }
}

Creating a ViewModel with AsyncStream

ViewModels can expose async streams for views to consume.

class ChatViewModel: ObservableObject {
    @Published var messages: [Message] = []
    private let messageStream: AsyncStream<Message>
    private var streamContinuation: AsyncStream<Message>.Continuation?
    
    init() {
        var continuation: AsyncStream<Message>.Continuation!
        self.messageStream = AsyncStream { cont in
            continuation = cont
        }
        self.streamContinuation = continuation
    }
    
    func startListening() async {
        for await message in messageStream {
            await MainActor.run {
                messages.append(message)
            }
        }
    }
    
    func send(_ content: String) {
        let message = Message(content: content, isOutgoing: true)
        streamContinuation?.yield(message)
    }
}

struct ChatView: View {
    @StateObject private var viewModel = ChatViewModel()
    
    var body: some View {
        List(viewModel.messages) { message in
            Text(message.content)
        }
        .task {
            await viewModel.startListening()
        }
    }
}

Task Management with AsyncStreams

Use task modifiers to manage async stream consumption lifecycle.

struct DataStreamView: View {
    @State private var items: [String] = []
    
    var body: some View {
        List(items, id: \.self) { item in
            Text(item)
        }
        .task(id: "stream") {
            for await item in dataStream() {
                items.append(item)
            }
        }
        .taskCancellable("stream") {
            // Cleanup when view disappears
        }
    }
}