8. Operator Pengumpulan dan Buffering

chunks(ofCount:)

Mengumpulkan tepat N elemen sebelum emit sebagai koleksi. Chunk terakhir mungkin berisi lebih sedikit elemen jika sequence selesai sebelum penuh.

swift
// chunks: proses dalam batch N elemen
let records = databaseStream()  // AsyncSequence<Record>

for await batch in records.chunks(ofCount: 100) {
    // batch: ArraySlice<Record> dengan maksimum 100 elemen
    try await bulkInsert(Array(batch))
    print("Inserted \(batch.count) records")
}

AsyncChannel sebagai Buffer

AsyncChannel dari package ini menyediakan back-pressure aware channel — cocok untuk producer-consumer dengan kecepatan berbeda.

swift
// AsyncChannel: channel dengan back-pressure
let channel = AsyncChannel<Data>()

// Producer task
Task {
    for url in urlList {
        let data = try await download(url)
        await channel.send(data)  // menunggu jika consumer sibuk
    }
    channel.finish()
}

// Consumer task
Task {
    for await data in channel {
        try await process(data)  // consumer mengontrol kecepatan
    }
}

AsyncThrowingChannel

Seperti AsyncChannel tapi bisa mengirim error untuk mengakhiri sequence.

swift
// AsyncThrowingChannel: channel yang bisa throw error
let channel = AsyncThrowingChannel<URLSessionDataTask, Error>()

Task {
    do {
        for request in pendingRequests {
            let task = try await executeRequest(request)
            await channel.send(task)
        }
        channel.finish()
    } catch {
        channel.fail(error)  // tutup channel dengan error
    }
}

// Consumer
Task {
    do {
        for try await task in channel {
            await handleTask(task)
        }
    } catch {
        print("Channel error: \(error)")
    }
}