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 — lesson 1 of 3. You can read the complete lesson below for free — then practise it hands-on in the browser with a built-in code editor and a 24/7 AI tutor. It is part of the Swift Academy learning path, one of 3 lessons in the course, and your progress syncs across the web and the CoddyKit app.
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.
Frequently asked questions
Is the “Producer/consumer, pipelines, timeouts” lesson free?
Yes — the full text of “Producer/consumer, pipelines, timeouts” is free to read here on the web, and the Swift Academy course includes 3 lessons in total. To practise it interactively (a built-in code editor and a 24/7 AI tutor) and unlock the rest of the Swift Academy course, upgrade to CoddyKit PRO.
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; this is — lesson 1 of 3, so you can start here or from the beginning and move at your own pace.
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)