RxSwift - SCISSION séquence en séquences plus petites à concaténer sur

Question

Contexte

Je suis en train de mettre en œuvre le téléchargement d'un fichier par morceaux en utilisant l'en-tête Range.

Objectif

Je souhaite prendre une grande séquence de requêtes HTTP en quatre séquences que je peux ensuite concat pour traiter les demandes de 4 à la fois.

Courant permanent

Je suis actuellement ma séquence et en utilisant concat pour faire en sorte que la première demande observable est terminée avant que je commence la deuxième. Cela a été fait pour s'assurer que je ne surcharge pas Alamofire avec trop de demandes et que je finisse par avoir des demandes qui expirent.

Idéalement, je voudrais diviser ma séquence en quatre séquences assez égales puisque Alamofire est configuré pour gérer quatre connexions à l'hôte à la fois. Je veux faire cela parce que je crois que cela augmenterait mes vitesses de téléchargement.

Téléchargement du fichier utilisant Chunks

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

Le code ci-dessous briser les morceaux en quatre tableaux de taille égale qui utilisent concat pour faire en sorte que seule sauvegarde de chaque tableau est actif à la fois. Cela signifie que vous aurez toujours 4 appels saveChunkToFile actifs à tout moment, peu importe la vitesse ou la lenteur d'un appel particulier. En d'autres termes, il lance immédiatement quatre requêtes, puis lance une requête à chaque fois qu'une des demandes précédentes est terminée.

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