Comparing Combine's subjects with AsyncStream in Swift

Combine's subjects such as PassthroughSubject or CurrentValueSubject have long been the go-to way to implement reactive observer patterns across an app. They provide a clear way to publish and observe values across different layers of an architecture.

With Swift concurrency, we now have another way to represent values that arrive over time: AsyncStream. In this article, we'll look at how it works, how it compares to Combine, and when it makes sense to switch.

Architecture & Design Patterns for iOS

This book is a practical guide to essential architectural principles and design patterns for iOS development. It covers strategies for building maintainable apps with Swift and SwiftUI, including dependency injection, navigation, common patterns, and modularization.

LEARN MORE

The Combine way

A PassthroughSubject is a simple Combine publisher. It emits values to all current subscribers, often used to send updates, for example from a service:

final class DataService {
    let values = PassthroughSubject<String, Never>()

    func load() {
        someAsyncCall() { [weak self] result in
            self?.didUpdate.send("New data loaded \(result)")
        }
    }
}

To observe updates, we subscribe using Combine’s .sink method:

class DataViewModel {
    private var cancellables = Set<AnyCancellable>()
    private let service = DataService()

    init() {
        service.values
            .receive(on: DispatchQueue.main)
            .sink { [weak self] value in
                // received a new value
            }
            .store(in: &cancellables)
    }
}

This setup works great in apps already built around Combine. The framework takes care of threading and synchronization for us, automatically delivering updates on the main queue when needed. Each subscription is cancellable through AnyCancellable, making cleanup predictable and safe. Combine also provides a rich set of operators like map, debounce, and combineLatest, allowing us to transform, coordinate, and throttle asynchronous values with minimal code. And through SwiftUI's .onReceive, Combine integrates naturally into the view layer.

The concurrency-native alternative: AsyncStream

AsyncStream provides a lightweight, concurrency-native way to emit values over time. It fits naturally into Swift's structured concurrency model and can be used as an alternative to Combine's subjects in systems which does not need the full reactive possibilities Combine provides.

When using AsyncStream, we need to be aware of that it does not handle thread safety or subscription management automatically, and cancellation behaves differently than in Combine. Used with care, it can be an elegant replacement for simple "publish and listen" flows.

Let's look at a simple example:

actor DataService {
    private var continuations: [UUID: AsyncStream<String>.Continuation] = [:]

    var updates: AsyncStream<String> {
        let id = UUID()
        return AsyncStream { continuation in
            continuations[id] = continuation
            continuation.onTermination = { [weak self] _ in
                Task { await self?.removeContinuation(id) }
            }
        }
    }

    func load() async {
        let result = await someAsyncCall()
        for continuation in continuations.values {
            continuation.yield("New data loaded \(result)")
        }
    }

    private func removeContinuation(_ id: UUID) {
        continuations[id] = nil
    }
}

To consume values, we can use for await:

let service = DataService()

Task {
    for await update in service.updates {
        print("Received:", update)
    }
}

Task {
    await service.load()
}

Let's unpack what's happening:

• When we create an AsyncStream we pass in a closure which gets called once per consumer. Inside that closure, Swift gives us an AsyncStream.Continuation. We can think of that continuation as a pipe into the consumer's for await loop. We can send (yield) values into it, and the consumer will receive them one by one.
• Since we can have multiple consumers, we save these continuations in an array: one continuation per active subscriber.
• When we later call yield we're broadcasting the new value to every consumer who’s currently listening. Each continuation represents an active for await loop somewhere in the app.
• If one of those subscribers stops consuming the stream (for example, its task is cancelled), the stream’s onTermination closure is called. That's when we remove that continuation from the array. That keeps our list clean and prevents sending values into closed streams.

We use an actor here because actors serialize access to shared mutable state. Without it, accessing the continuations collection from multiple concurrent tasks could lead to race conditions.

Reusing AsyncStream

If we find ourselves creating many small streams, we can wrap this pattern in a lightweight helper, similar to a PassthroughSubject, but based on AsyncStream:

actor AsyncPassthrough<Value> {
    private var continuations: [UUID: AsyncStream<Value>.Continuation] = [:]

    func stream() -> AsyncStream<Value> {
        let id = UUID()
        return AsyncStream { continuation in
            continuations[id] = continuation
            continuation.onTermination = { [weak self] _ in
                Task { await self?.removeContinuation(id) }
            }
        }
    }

    func send(_ value: Value) {
        for continuation in continuations.values {
            continuation.yield(value)
        }
    }

    private func removeContinuation(_ id: UUID) {
        continuations[id] = nil
    }
}

Usage can look as follows:

let messages = AsyncPassthrough<String>()

Task {
    for await msg in await messages.stream() {
        print("Received:", msg)
    }
}

Task {
    await messages.send("Hello")
    await messages.send("World")
}

What about CurrentValueSubject?

If we need the behaviour of a CurrentValueSubject to hold and publish the latest value, we could build something similar as follows:

actor CurrentValue<Value> {
    private var value: Value
    private let passthrough = AsyncPassthrough<Value>()

    init(_ initial: Value) { value = initial }

    func updates() async -> AsyncStream<Value> {
        await passthrough.stream()
    }

    func set(_ newValue: Value) async {
        value = newValue
        await passthrough.send(newValue)
    }

    func get() -> Value { value }
}

Here, we store the latest value in a property and expose a stream that emits it whenever it changes. This pattern gives us both current state (get()) and an async sequence of updates (updates).

Using AsyncSequence operators

Since AsyncStream conforms to AsyncSequence, we can also use the same sequence operators Swift provides, such as map, filter or prefix. For example:

Task {
    for await update in await service.updates
        .filter { !$0.isEmpty }
        .map { $0.uppercased() }
        .prefix(3)
    {
        print("Received:", update)
    }
}

This applies basic transformations directly to the stream: we filter out empty updates, convert them to uppercase, and only take the first three results.

While these basic operators cover many common cases, they don’t replace the depth of Combine’s reactive toolset. Features like debounce, merge, or combineLatest are often essential when coordinating multiple streams or handling user-driven events. In those scenarios, Combine still provides a richer and more expressive way to work with asynchronous data flows.

Note on cancellation

When a task consuming the stream is cancelled, Swift automatically calls the continuation's onTermination handler. This means cancellation happens for free in many structured contexts, such as SwiftUI's .task modifier. However, when we manually create tasks outside of SwiftUI, we need to manage cancellation ourselves.

We'll explore this distinction in an upcoming article.

Wrapping up

AsyncStream doesn't replace Combine. For complex reactive data flows, Combine currently remains the more powerful framework. For lightweight "publish and listen" patterns, AsyncStream can be used as an alternative.