RxSwift - Observable.generate - 使用附加映射处理顺序请求

Question

目标

我正在尝试实现一个使用 Range 标头下载文件的服务。这允许我一次下载一大块文件。

实现 - Observable.generate()

为了为每个请求创建一个 observable 并保存我尝试使用的文件：

let downloadObservable = Observable.generate(initialState: 0, condition: { $0 < fileSize }, iterate: { $0 + self.defaultChunkSize })

这似乎工作得很好！除了较大的文件，它似乎有一个错误。我的请求被取消了。调试后，我发现我的工作流程并没有按照我的预期工作。这是附加到上一行的其余工作流程。

.map( { (startChunk) -> (Int64, Int64) in
    // I determine the end chunk so I can download any size file in chunks of size X
    let endChunk = (startChunk + self.defaultChunkSize > fileSize ? fileSize : startChunk + self.defaultChunkSize )

    return (startChunk, endChunk)
}).flatMap( { [unowned self] (startChunk: Int64, endChunk: Int64) -> Observable<FileChunk> in
    // I make the request via alamofire - UNEXPECTED FLOW HERE SEE NOTE #1
    return self.makeChunkRequest(url: downloadUrl, startChunk: startChunk, endChunk: endChunk)

}).flatMap( { [unowned self] (fileChunk: FileChunk) -> Observable<FileSaveChunkResult> in
    // Upon receiving chunk response save to file
    return self.saveChunkToFile(fileChunk: fileChunk, location: localDestinationUrl)

}).flatMap( { (saveResult: FileSaveChunkResult) -> Observable<Progress> in
    // Update progress if successful
    switch (saveResult) {
    case .success(let bytesSaved):
        progress.completedUnitCount += bytesSaved
    case .failure:
        break
    }

    return Observable.just(progress)
})

注意 #1

当我运行和调试它时，我的第一个 flatMap 循环直到发出 ALL 块请求。我希望这会更加连续，我们将生成一个 observable，然后通过 flatMap 进行所有转换，然后循环回到开头。

这不是我应该如何实施的吗？

我需要对我的Observable.generate() 上的merge() 施展魔法吗？

Answer 1

我想我已经找到了这个问题的解决方案。关键是通过网络请求map，然后concat 他们。这样做而不是使用flatMap。 concat 操作符将等到请求发送 onCompleted 后再开始下一个请求。代码如下：

let downloadObservable = Observable.generate(initialState: 0, condition: { $0 < fileSize }, iterate: { $0 + self.defaultChunkSize })
    .map( { (startChunk) -> (Int64, Int64) in
        let endChunk = (startChunk + self.defaultChunkSize > fileSize ? fileSize : startChunk + self.defaultChunkSize )
        return (startChunk, endChunk)
    }).map( { [unowned self] (startChunk: Int64, endChunk: Int64) -> Observable<FileChunk> in
        return self.makeChunkRequest(url: downloadUrl, startChunk: startChunk, endChunk: endChunk)
    }).concat()
    .flatMap( { [unowned self] (fileChunk: FileChunk) -> Observable<FileSaveChunkResult> in
        return self.saveChunkToFile(fileChunk: fileChunk, location: localDestinationUrl)
    }).flatMap( { (saveResult: FileSaveChunkResult) -> Observable<Progress> in
        if case .success(let bytesSaved) = saveResult {
            progress.completedUnitCount += bytesSaved
        }
        return Observable.just(progress)
    })

我想出了如何将它分成 4 批。我将它展开了一点，并在代码中制作了 cmets 以提供帮助：

let generator = Observable.generate(initialState: 0, condition: { $0 < fileSize }, iterate: { $0 + defaultChunkSize })
let chunks  = generator.map( { (startChunk) -> (Int64, Int64) in
    let endChunk = (startChunk + defaultChunkSize > fileSize ? fileSize : startChunk + defaultChunkSize )
    return (startChunk, endChunk)
})
let requests = chunks.buffer(timeSpan: 0.0, count: 4, scheduler: MainScheduler.instance)// makes batches of four item arrays.
    .map { (batch) -> Observable<FileChunk> in
        let requests = Observable.from(batch) // spreads the four items back out.
        return requests.flatMap( { (startChunk: Int64, endChunk: Int64) -> Observable<FileChunk> in
            return makeChunkRequest(url: downloadUrl, startChunk: startChunk, endChunk: endChunk)
        }) // start the four requests as normal.
    }.concat() // wait until the four requests are finished before allowing the next four to begin.

let downloadObservable = requests
    .flatMap( { (fileChunk: FileChunk) -> Observable<FileSaveChunkResult> in
        return saveChunkToFile(fileChunk: fileChunk, location: localDestinationUrl)
    }).flatMap( { (saveResult: FileSaveChunkResult) -> Observable<Progress> in
        if case .success(let bytesSaved) = saveResult {
            progress.completedUnitCount += bytesSaved
        }
        return Observable.just(progress)
    })