DataProvider using AsyncStream

DataProvider.swift

import Foundation

struct DataProvider<Element: Sendable> {
    let stream: AsyncStream<Element>
    private let continuation: AsyncStream<Element>.Continuation

    init(bufferingPolicy: AsyncStream<Element>.Continuation.BufferingPolicy = .unbounded) {
        (stream, continuation) = AsyncStream.makeStream(bufferingPolicy: bufferingPolicy)
    }

    func receiveNext(item: Element) {
        continuation.yield(item)
    }

    func finish() {
        continuation.finish()
    }
}

processEachItem.swift

func processEachItem<ItemType, ResultType>(_ items: [ItemType], operation: @escaping @Sendable (ItemType) -> ResultType) async -> [ResultType] {
	await withTaskGroup(of: ResultType.self) { group in
		for item in items {
			group.addTask {
				operation(item)
			}
		}
		return await group.reduce(into: []) { $0.append($1) }
	}
}