RxSwift -

Question

背景

にCONCATする小さなシーケンスにシーケンスを分割私はRangeヘッダーを使用してチャンクを経由してファイルをダウンロードして実装しようとしています。私はその後、一度に4つの要求を処理するためにCONCATできる4つの系列にhttpリクエストの大シーケンスを利用したいと思い

ゴール

。

現在の立ち

私は現在、私のシーケンスを取り、私は2番目を開始する前に、観察最初の要求が完了したことを確認するためにconcatを使用しています。これは、あまりにも多くのリクエストをAlamofireに負荷をかけすぎないようにするために行われたもので、タイムアウトしたリクエストで終了します。

Alamofireは一度に4つのホスト接続を処理するように設定されているため、理想的にはシーケンスを4つの同等のシーケンスに分割したいと考えています。私はそれが私のダウンロード速度を高めると信じているので、私はこれをしたい。

ダウンロードファイルを使用してチャンク

Observable.generate(initialState: 0, condition: { $0 < fileSize }, iterate: {$0 + self.defaultChunkSize}) 
    .map({ (startChunk) in 
     let endChunk = startChunk + self.defaultChunkSize > fileSize ? fileSize : startChunk + self.defaultChunkSize 

     return (startChunk, endChunk) 
    }) 
    .map({ (startChunk: Int, endChunk: Int) -> Observable<FileChunkResult> in 
     self.filesClient.downloadChunkOf(fileId: file.id, startChunk: Int64(startChunk), endChunk: Int64(endChunk)) 
    }) 

    .concat() // <----- This is where I am forcing the large sequence to do one observable at a time 

    .flatMap({ (result: FileChunkResult) -> Observable<FileSaveChunkResult> in 
     switch (result) { 
     case FileChunkResult.success(let chunkData): 
      return self.saveChunkToFile(fileChunk: chunkData, location: urlToSaveTo) 
     case FileChunkResult.failure: // Maybe change this to just default and return Observable.just(FileSaveChunkResult.failure) 
      break 
     case FileChunkResult.parserError: 
      break 
     } 

     return Observable.just(FileSaveChunkResult.failure) 
    }) 
    .flatMap({ (result: FileSaveChunkResult) -> Observable<Progress> in 
     switch (result) { 
     case FileSaveChunkResult.success(let bytesSaved): 
      progress.completedUnitCount += bytesSaved 
     case FileSaveChunkResult.failure: 
      break 
     } 

     return Observable.just(progress) 
    }) 

Answer 1

コードは、以下の各アレイからのセーブ一方のみが一度にアクティブであることを保証するために連結を使用して4つの等しいサイズの配列にチャンクを分割します。つまり、特定のコールがどれだけ速くても遅くても、一貫して4 saveChunkToFileコールをアクティブにすることができます。

つまり、直ちに4つの要求を開始し、前の要求の1つが完了するたびに1つの要求を開始します。

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 count = ceil(Double(fileSize)/Double(defaultChunkSize)/4) 
let requests = chunks.window(timeSpan: 0.0, count: Int(count), scheduler: MainScheduler.instance) 
    .flatMap { $0 
     .map({ (startChunk: Int64, endChunk: Int64) -> Observable<FileChunk> in 
      return makeChunkRequest(url: downloadUrl, startChunk: startChunk, endChunk: endChunk) 
     }).concat() 
} 

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) 
    }) 

_ = downloadObservable.subscribe(onNext: { print(Date(), $0) })