|
|
import Foundation
|
|
|
|
|
|
/// Verarbeitet mehrere Dateien über `ConversionPipeline`, begrenzt auf eine
|
|
|
/// feste Anzahl gleichzeitiger Exporte (README: "1–2 parallele Exports").
|
|
|
/// Ergebnisse kommen fortlaufend rein, sobald ein Job fertig ist, nicht erst
|
|
|
/// am Ende des gesamten Batches — die UI kann so live mitschreiben.
|
|
|
public enum JobQueue {
|
|
|
public struct Job: Sendable {
|
|
|
public let id: UUID
|
|
|
public let sourceURL: URL
|
|
|
|
|
|
public init(id: UUID = UUID(), sourceURL: URL) {
|
|
|
self.id = id
|
|
|
self.sourceURL = sourceURL
|
|
|
}
|
|
|
}
|
|
|
|
|
|
public struct JobResult: Sendable {
|
|
|
public let job: Job
|
|
|
public let outcome: ConversionPipeline.Outcome
|
|
|
}
|
|
|
|
|
|
public static func run(
|
|
|
jobs: [Job],
|
|
|
outputRootDirectory: URL,
|
|
|
options: ConversionPipeline.Options = ConversionPipeline.Options(),
|
|
|
maxConcurrent: Int = 2,
|
|
|
progress: (@Sendable (Job, Double) -> Void)? = nil
|
|
|
) -> AsyncStream<JobResult> {
|
|
|
AsyncStream { continuation in
|
|
|
let task = Task {
|
|
|
await withTaskGroup(of: JobResult.self) { group in
|
|
|
var iterator = jobs.makeIterator()
|
|
|
|
|
|
func startNext() {
|
|
|
guard let job = iterator.next() else { return }
|
|
|
group.addTask {
|
|
|
let outcome = await ConversionPipeline.process(
|
|
|
sourceURL: job.sourceURL,
|
|
|
outputRootDirectory: outputRootDirectory,
|
|
|
options: options
|
|
|
) { value in
|
|
|
progress?(job, value)
|
|
|
}
|
|
|
return JobResult(job: job, outcome: outcome)
|
|
|
}
|
|
|
}
|
|
|
|
|
|
let concurrencyLimit = max(1, maxConcurrent)
|
|
|
for _ in 0..<min(concurrencyLimit, jobs.count) {
|
|
|
startNext()
|
|
|
}
|
|
|
|
|
|
while let result = await group.next() {
|
|
|
continuation.yield(result)
|
|
|
startNext()
|
|
|
}
|
|
|
}
|
|
|
continuation.finish()
|
|
|
}
|
|
|
continuation.onTermination = { _ in task.cancel() }
|
|
|
}
|
|
|
}
|
|
|
}
|
