全部完成后 ReactiveSwift 管道计数失败

Question

我在ReactiveSwift 中有一个用于上传的管道。我想确保即使其中一个上传失败，其余的也不会被中断。

在所有这些都成功或失败后，我应该从performUploads 方法返回成功，或者如果有任何失败，我应该返回一个错误，所以下一步，下载部分不会开始。即使有错误，所有上传都应该有机会上传，不应该停止。

有没有办法判断上传完成后是否有任何错误？ 方法见这里：

let pendingUploadItemsArray: [Items] = ...
func performUploads() -> SignalProducer<(), MyError> {
    return upload(pendingUploadItemsArray)
        .then(doAnything())
}

private func upload(_ items: [Items]) -> SignalProducer<Signal<(), MyError>.Event, Never> {
    let producers = items
        .filter { item in
            return item.readyForUpload
        }
        .map { self.upload($0).materialize() }
    
    return SignalProducer.merge(producers)
}

private func upload(_ item: Item) -> SignalProducer<(), MyError> {
    return internalUploader.upload(item)
        .on(failed: failedToUpload(item),
            value: successfullyUploaded(item))
        .ignoreValues()
}

internalUploader上传方法在哪里：

func upload(_ item: Item) -> SignalProducer<Any, MyError>

然后在另一个班级中，您将调用此上传者：

let sync = self.uploader.performUploads()
        .then(startDownloads())

startDownloads 只有在所有上传都成功完成后才能运行。 感谢您提供任何见解。

这可能应该以完全不同的方式完成。

Answer 1

我不确切知道successfullyUploaded 和failedToUpload 在您的代码中做了什么，但大概您正在跟踪成功和失败以提供某种实时进度UI。这就是我的结构：

struct UploadResult {
    let item: Item
    let error: Error? // nil if the upload succeeded

    var succeeded: Bool { error == nil }
    var failed: Bool { !succeeded }
}

...

static func upload(_ items: [Item]) -> SignalProducer<[UploadResult], Never> {
    SignalProducer(items)
        .filter(\.readyForUpload)
        .flatMap(.merge) { item in
            Self.internalUploader(item)
                .map { UploadResult(item: item, error: nil) }
                .flatMapError { error in
                    SignalProducer(value: UploadResult(item: item, error: error))
                }
        }
        .scan(into: [UploadResult]()) { ( results: inout [UploadResult], nextResult) in
            results.append(nextResult)
        }
}

1. 我创建了一个 UploadResult 结构体，它表示上传成功或失败的项目。
2. 在upload 函数中，我没有创建生产者数组然后合并它们，而是将项目数组转换为具有SignalProducer(items) 的项目的信号生产者，然后使用flatMap(.merge) 将上传合并到一个单一信号发生器。
3. 我没有使用materialize，而是使用map 将成功的上传转换为UploadResult，并使用flatMapError 将失败的上传转换为UploadResult。
4. 我使用scan 在每次上传完成时累积结果。每次上传完成（成功或出错）时，scan 将发送更新的上传结果数组，可用于更新 UI。

那么你可以这样使用它：

Uploader.upload(someItems)
    .on(value: { resultsSoFar in
        // Update UI here
    })
    .take(last: 1)
    .attempt { results in
        if !results.allSatisfy(\.succeeded) {
            // At least one of the uploads failed, so send an error
            throw MyError()
        }
    }
    .then(startDownloads)

1. 我使用on(value:) 运算符根据当前结果更新 UI。每次下载成功或失败时，都会使用更新的结果调用此闭包。
2. 我使用take(last: 1)过滤掉所有中间结果；它只会在所有上传完成后发送最终结果。
3. 我使用attempt 检查是否有任何上传失败，如果失败则抛出错误。这可确保只有在所有上传成功后才会开始下载。

希望这能处理您的用例，但如果我错过了有关更广泛背景的某些内容，请在评论中告诉我！

编辑

如果你只关心一次处理一个结果而不是作为一个运行数组，你可以去掉scan，然后用collect替换take(last: 1)：

static func upload(_ items: [Item]) -> SignalProducer<UploadResult, Never> {
    SignalProducer(items)
        .filter(\.readyForUpload)
        .flatMap(.merge) { item in
            Self.internalUploader(item)
                .map { UploadResult(item: item, error: nil) }
                .flatMapError { error in
                    SignalProducer(value: UploadResult(item: item, error: error))
                }
        }
}

...

Uploader.upload(someItems)
    .on(value: { latestResult in
        // Do something with the latest result
    })
    .collect()
    .attempt { results in
        if !results.allSatisfy(\.succeeded) {
            // At least one of the uploads failed, so send an error
            throw MyError()
        }
    }
    .then(startDownloads)