Producer/consumer, pipelines, timeouts
Build a simple producer/consumer , compose pipelines with AsyncSequence, and implement a timeout by racing a sleep task.
Producer/consumer, pipelines, timeouts is a free Swift Academy lesson on CoddyKit. This is lesson 1 of 3. Any 3 lessons of this learning path are free to read in full — after that, CoddyKit PRO unlocks every lesson, plus hands-on practice with a built-in code editor and a 24/7 AI tutor. It is part of the Swift Academy learning path, and your progress syncs across the web and the CoddyKit app. The Swift Academy course includes 3 lessons in total.
What you will build
This lesson shows three patterns:
- Producer/consumer with AsyncStream
- Pipelines using AsyncSequence transforms
- Timeouts by racing tasks cooperatively
Basic producer/consumer
AsyncStream bridges push-style producers with pull-based for await consumption.
// Producer pushes values; consumer iterates them asynchronously.
func makeNumbers(count: Int) -> AsyncStream<Int> {
AsyncStream { continuation in
Task {
for i in 1...count {
continuation.yield(i)
try? await Task.sleep(nanoseconds: 40_000_000) // ~40ms
}
continuation.finish()
}
}
}
Task {
let stream = makeNumbers(count: 5)
for await n in stream {
print("consume:", n)
}
print("done") // after finish()
}Backpressure policy
Choose a bufferingPolicy to manage backpressure (drop oldest, drop newest, or unbounded for small streams).
// Use a limited buffering policy to avoid memory spikes.
func makeBufferedNumbers(limit: Int) -> AsyncStream<Int> {
AsyncStream(bufferingPolicy: .bufferingOldest(limit)) { cont in
Task {
for i in 1...20 {
cont.yield(i)
// Producer is fast; buffer prevents unbounded growth
}
cont.finish()
}
}
}
Task {
let s = makeBufferedNumbers(limit: 5)
for await n in s {
// Simulate a slow consumer
try? await Task.sleep(nanoseconds: 60_000_000)
print("got:", n)
}
}Pipeline with AsyncSequence
AsyncSequence supports familiar operators (map, filter, reduce) so you can build streaming pipelines clearly.
// Compose a pipeline: map -> filter -> reduce over an AsyncSequence.
func numbers(_ n: Int) -> AsyncStream<Int> {
AsyncStream { cont in
Task {
for i in 1...n { cont.yield(i) }
cont.finish()
}
}
}
Task {
let evensSquaredSum = await numbers(10)
.map { $0 * $0 }
.filter { $0.isMultiple(of: 2) }
.reduce(0, +)
print("sum:", evensSquaredSum) // 220 (4+16+36+64+100)
}Parallel stage
Use task groups inside pipelines for CPU-bound transforms, then merge results back in order you prefer.
// Fan-out/fan-in: process items concurrently, then merge.
// Here we use a task group to transform elements in parallel.
func parallelUppercased(_ input: [String]) async -> [String] {
await withTaskGroup(of: String.self) { group in
for s in input {
group.addTask {
try? await Task.sleep(nanoseconds: 30_000_000)
return s.uppercased()
}
}
return await group.reduce(into: [String]()) { $0.append($1) }
}
}
Task {
let out = await parallelUppercased(["a","bb","ccc"])
print(out) // ["A","BB","CCC"]
}Timeout helper
Implement a timeout by racing the operation against a sleep task; cancel the loser to save work.
enum TimeoutError: Error { case timedOut }
func withTimeout<T>(
seconds: Double,
operation: @escaping () async throws -> T
) async throws -> T {
try await withThrowingTaskGroup(of: T.self) { group in
// Child 1: actual work
group.addTask { try await operation() }
// Child 2: the timer
group.addTask {
try await Task.sleep(nanoseconds: UInt64(seconds * 1_000_000_000))
throw TimeoutError.timedOut
}
// First child to finish wins; cancel the rest
let result = try await group.next()!
group.cancelAll()
return result
}
}
// Demo
func slowFetch() async throws -> String {
try await Task.sleep(nanoseconds: 300_000_000) // 300ms
return "OK"
}
Task {
do {
let v = try await withTimeout(seconds: 0.1) { try await slowFetch() }
print("value:", v)
} catch {
print("timeout:", error) // expected
}
}Timeout pattern (race)
Quick check: What is a good way to add a timeout?
Recap
Recap:
- Use AsyncStream for producer/consumer.
- Compose AsyncSequence pipelines with map/filter/reduce.
- Implement timeouts by racing tasks and cancelling the loser.
Learn Swift with an AI tutor — free
Write and run real code in your browser, get instant help from a 24/7 AI tutor, and pick up where you left off on the web or in the app.
- Courses
- 122
- Lessons
- 409
Frequently Asked Questions
Is the “Producer/consumer, pipelines, timeouts” lesson free?
Yes — any 3 lessons of the Swift Academy learning path, including “Producer/consumer, pipelines, timeouts”, are free to read in full here on the web. After that, CoddyKit PRO unlocks every lesson, plus interactive practice with a built-in code editor and a 24/7 AI tutor. The Swift Academy course includes 3 lessons in total.
What will I learn in “Producer/consumer, pipelines, timeouts”?
Build a simple producer/consumer , compose pipelines with AsyncSequence, and implement a timeout by racing a sleep task. You practise Swift Academy with hands-on code you run directly in the browser, and a 24/7 AI tutor answers your questions as you work through the lesson.
Do I need any experience to start Swift Academy?
No prior experience is required. Swift Academy on CoddyKit is structured for beginners through advanced learners, so you can start here or from the beginning and move at your own pace. This is lesson 1 of 3.
How long does the “Producer/consumer, pipelines, timeouts” lesson take?
Most CoddyKit lessons take about 5–10 minutes. Each one is bite-sized and interactive, so you make steady progress and pick up exactly where you left off across the web and the app.
Can I write and run code in this Swift Academy lesson?
Yes. Every Swift Academy lesson includes a built-in code editor, so you write and run real code right in your browser and get instant AI feedback — no local setup required.
All lessons in this course
- Producer/consumer, pipelines, timeouts
- Bridging legacy callbacks/Combine to async/await
- Testing async code (XCTest)