AsyncStream & AsyncThrowingStream in Swift - Real-Time Data Sync Explained

Modern apps are no longer single-device experiences.

Users expect their data to stay perfectly in sync across iPhone, iPad, and Mac, with updates appearing instantly—no refresh button, no manual sync. This expectation becomes even stronger in productivity apps like invoicing, accounting, or note-taking, where consistency and correctness matter.

In my invocing app Reckord, users can sign in on multiple devices and continue their work seamlessly. To support this, I needed a real-time data synchronization system that:

• Reacts immediately to remote changes
• Integrates cleanly with Swift’s async/await model
• Cancels automatically when views disappear
• Avoids callback-heavy, hard-to-maintain code

To achieve this, I built the entire real-time sync layer using AsyncStream and AsyncThrowingStream.

This article explains:

• What AsyncStream is
• What AsyncThrowingStream is
• When to use each
• How they work internally
• And how I used them to implement real-time multi-device syncing in Reckord

The Problem with Callback-Based Real-Time APIs

Most real-time systems—including Firebase Firestore—are built around callbacks.

A typical Firestore listener looks like this:

listener = query.addSnapshotListener { snapshot, error in
    // Called every time data changes
}

While functional, this approach has drawbacks in modern SwiftUI apps:

• Hard to integrate with async/await
• Manual listener cleanup
• Complex cancellation logic
• Business logic scattered across closures

What we want is something like this:

for await update in updates {
    apply(update)
}

This is exactly what AsyncSequence enables—and AsyncStream is how we create one.

AsyncSequence: The Foundation

Both AsyncStream and AsyncThrowingStream conform to AsyncSequence.

An AsyncSequence:

• Produces multiple values over time
• Is consumed using for await
• Integrates naturally withTask and SwiftUI lifecycles
• Supports structured cancellation

You can think of it as:

The async/await equivalent of a data stream.

What is AsyncStream?

AsyncStream is a Swift type that allows you to create an asynchronous sequence of values that are produced over time, rather than all at once.

In simple terms:

AsyncStream lets you manually feed values into an async for await loop.

Why AsyncStream Exists

Before Swift Concurrency, when values arrived over time, we usually relied on:

• Closures
• Delegates
• Notifications
• Combine publishers

These approaches don’t integrate naturally with async/await.

AsyncStream exists to bridge non-async, callback-based APIs into Swift’s structured concurrency world.

How AsyncStream Works (Conceptually)

An AsyncStream has two sides:

• Producer: This is where values are generated using a Continuation.
• Consumer: This is where values are consumed using for await.

The producer and consumer can live in completely different parts of your app.

Anatomy of AsyncStream

AsyncStream<Element> { continuation in
    continuation.yield(value)
    continuation.finish()
}

• Element → type of values emitted
• continuation.yield(_:) → sends a value to consumers
• continuation.finish() → ends the stream

Once finished, no more values can be emitted.

A Very Simple Example

Let’s say you want to emit numbers over time.

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

Consume it like this:

Task {
    for await number in numberStream() {
        print(number)
    }
}

Output:

1
2
3
4
5

This is not a loop returning values — it’s a stream pushing values asynchronously.

Important Characteristics of AsyncStream

• Can emit multiple values
• Works naturally with async/await
• Cannot throw errors
• Must be manually finished
• Supports cancellation

Because it cannot fail, AsyncStream is best suited for:

• UI events
• Timers
• State changes
• In-memory signals

What is AsyncThrowingStream?

AsyncThrowingStream is the error-capable sibling of AsyncStream.

It represents an async sequence that:

• Emits values over time
• Can fail with an error

In other words:

AsyncThrowingStream is AsyncStream + error handling.

Why AsyncThrowingStream Exists

Most real-world asynchronous systems can fail:

• Network requests
• Databases
• File IO
• Permissions
• Authentication

Using AsyncStream for such systems would mean:

• Errors must be hidden
• Or sent as values (bad idea)
• Or handled outside the stream (messy)

AsyncThrowingStream solves this cleanly.

Anatomy of AsyncThrowingStream

AsyncThrowingStream<Element, Error> { continuation in
    continuation.yield(value)
    continuation.finish(throwing: error)
}

Key differences:

• Can call finish(throwing:)
• Consumers must use for try await

Simple Example with Errors

Imagine a stream that emits random numbers but fails sometimes.

enum RandomError: Error {
    case unlucky
}

func randomNumberStream() -> AsyncThrowingStream<Int, Error> {
    AsyncThrowingStream { continuation in
        Task {
            for _ in 1...10 {
                let number = Int.random(in: 1...10)

                if number == 7 {
                    continuation.finish(throwing: RandomError.unlucky)
                    return
                }

                continuation.yield(number)
                try await Task.sleep(for: .seconds(1))
            }

            continuation.finish()
        }
    }
}

Consume it like this:

Task {
    do {
        for try await number in randomNumberStream() {
            print(number)
        }
    } catch {
        print("Stream failed:", error)
    }
}

Output:

3
2
4
Stream failed: unlucky

It generates random numbers between 1 and 10 and prints them. If the generated number is 7, it stops generating numbers and throws an error.

Key Characteristics of AsyncThrowingStream

• Emits multiple values
• Can terminate with an error
• Natural error propagation
• Structured cancellation
• Ideal for networked systems

This makes it perfect for real-time databases like Firestore.

AsyncStream vs AsyncThrowingStream (Mental Model)

If you’re unsure, ask yourself:

Can this realistically fail at runtime?
If yes → use AsyncThrowingStream.

How I Implemented Real-Time Multi-Device Sync in Reckord

In Reckord, users can sign in on multiple devices and expect their data to stay perfectly in sync.

This means:

• Any change on one device
• Must instantly appear on all other devices
• Without refresh, polling, or manual sync

Firestore already provides real-time updates via snapshot listeners, but those APIs are callback-based.

My goal was to:

• Convert Firestore listeners into async/await
• Automatically manage lifecycle and cancellation
• Keep UI code clean and predictable

Step 1: Model Firestore as an AsyncThrowingStream

Firestore snapshot listeners:

• Emit initial data
• Emit updates continuously
• Can fail
• Must be removed manually

That maps exactly to AsyncThrowingStream.

func invoicesStream(userId: String) -> AsyncThrowingStream<[Invoice], Error> {
    AsyncThrowingStream { continuation in
        let listener = firestore
            .collection("invoices")
            .whereField("userId", isEqualTo: userId)
            .addSnapshotListener { snapshot, error in
                if let error {
                    continuation.finish(throwing: error)
                    return
                }

                guard let snapshot else { 
                    continuation.finish(throwing: FirestoreServiceError.invalidPath)
                    return 
                }

                do {
                    var invoices: [Invoice] = []
                    for document in snapshot.documents {
                        let parsedInvoice = try FirestoreParser.parse(document.data(), type: Invoice.self)
                        invoices.append(parsedInvoice)
                    }
                    continuation.yield(invoices)
                } catch {
                    continuation.finish(throwing: FirestoreServiceError.parseError)
                }
            }

        continuation.onTermination = { _ in
            listener.remove()
        }
    }
}

This single function:

• Turns Firestore into an async sequence
• Handles real-time updates
• Cleans up listeners automatically

Step 2: Consume the Stream in SwiftUI

@MainActor
func observeInvoices() async {
    do {
        for try await invoices in invoicesStream(userId: userId) {
            self.invoices = invoices
        }
    } catch {
        self.errorMessage = error.localizedDescription
    }
}

And attach it to the view lifecycle:

.task {
    await observeInvoices()
}

Step 3: Automatic Multi-Device Sync

Now the flow looks like this:

• Device A updates data
• Firestore emits a snapshot update
• AsyncThrowingStream yields new values
• Device B receives changes instantly
• SwiftUI view updates automatically

No polling.
No manual refresh.
No Combine.

Just structured concurrency.

Where AsyncStream Fits in Reckord

While Firestore uses AsyncThrowingStream, I use AsyncStream internally for:

• Authentication status changes
• Subscription state changes

This keeps error handling where it belongs and avoids overusing throwing streams.

Final Thoughts

AsyncStream and AsyncThrowingStream are foundational tools for modern Swift development.

They allow you to:

• Model real-time systems cleanly
• Embrace structured concurrency
• Write predictable, maintainable code
• Build true multi-device experiences

They form the backbone of Reckord ’s real-time sync system—and once you adopt them, callback-heavy designs start to feel outdated.

Thank you for reading. If you have any questions, feel free to follow me on X and send me a DM. If you enjoyed this article and would like to support my work, Buy me a coffee ☕️