Управление несколькими добавлениями с помощью ReactiveX (на iOS с Swift и Alamofire)

Question

Я пытаюсь загрузить несколько фотографий на сервер с помощью ReactiveX (RxSwift), собирать ответы от каждого запроса и затем сделать один окончательный запрос для завершения подачи.

Все, кажется, работает достаточно хорошо, пока я не попытаюсь ответить reduce на все ответы. Финал subscribeNext никогда не вызывается. (Возможно, я не понимаю, как работает flatMap или reduce?)

В частности, я пытаюсь выполнить эту процедуру.

• Приготовьте наблюдаемый для кодирования каждой фотографии (self.imageMgr является экземпляром PHCachingImageManager())

  func getPhotoDataObservable(asset: PHAsset) -> Observable<NSData> { 
      return create { observer in 
       self.imageMgr.requestImageForAsset(asset, 
        targetSize: PHImageManagerMaximumSize, 
        contentMode: .AspectFit, 
        options: nil, 
        resultHandler: { (myImage, myInfo) -> Void in 
         let data = UIImageJPEGRepresentation(myImage!, 1.0)! 
         NSLog("Encoded photo") 
         observer.onNext(data) 
         self.converts += 1 
         if self.converts == self.userReview.photos.count { 
          NSLog("Completed encoding photos") 
          observer.onCompleted() 
         } 
        }) 
       return NopDisposable.instance 
      } 
  } 
  

• Подготовка наблюдаемого для загрузки каждой фотографии когда-то закодированная (с Alamofire и RxAlamofire)

  func getPostPhotoObservable(photoData: NSData) -> Observable<ReviewPhotoObject> { 
      return create { observer in 
       NSLog("Uploading Photo") 
  
       upload(.POST, 
        urlRequest.URLString, 
        headers: nil, 
        multipartFormData: { mfd in 
         mfd.appendBodyPart(data: photoData, name: "image", fileName: "image", mimeType: "image/jpeg") 
        }, 
        encodingMemoryThreshold: Manager.MultipartFormDataEncodingMemoryThreshold, 
        encodingCompletion: { encodingResult in 
         switch encodingResult { 
         case .Success(let upload, _, _): 
          upload.responseJSON(completionHandler: { (myResponse) -> Void in 
           if let photoResponse = myResponse.result.value { 
            let photoObject = photoResponse.objectForKey("photo")! 
            let photo = ReviewPhotoObject() 
            photo.photoID = photoObject.objectForKey("id")! as! NSNumber 
            NSLog("Uploaded Photo") 
            observer.onNext(photo) 
           } 
  
           self.uploads += 1 
           if self.uploads == self.userReview.photos.count { 
            NSLog("Completed uploading photos") 
            observer.onCompleted() 
           } 
          }) 
  
         case .Failure(let encodingError): 
          observer.onError(encodingError) 
          print(encodingError) 
         } 
        }) 
  
       return NopDisposable.instance 
      } 
  } 
  

• И наконец, все вместе

  func postReview(review: MyReview) { 
      self.userReview = review 
  
      _ = review.photos.toObservable().flatMap { photos in 
       return self.getPhotoDataObservable(photos) 
      }.flatMap { photoData in 
       return self.getPostPhotoObservable(photoData) 
      }.reduce([], { var accumulator, photo: ReviewPhotoObject) -> [Int] in 
       accumulator.append(Int(photo.photoID)) 
       return accumulator 
      }).subscribeNext({ (photoIds) -> Void in 
       print(photoIds) // Never called 
      }) 
  } 
  

При запуске (с 2-х фотографий, например), это выход:

Encoded photo 
Uploading photo 
Encoded photo 
Uploading photo 
Completed encoding photos 
Uploaded photo 
Uploaded photo 
Completed uploading photos 

Но subscribeNext никогда не вызывается. Поскольку документация на RxSwift специально по-прежнему немного тонкая, я надеялся, что кто-то здесь может понять меня в том, что я недопонимаю.

Answer 1

Идея здесь в том, что после того, как будет выполнено наблюдение, отправив все элементы, которые он собирается отправить, он должен завершиться. Вы создаете наблюдаемый для каждого PHAsset, и наблюдаемый только отправляет один элемент, чтобы он после этого завершался. То, как у вас был код, только последний был завершен, поэтому оператор reduce просто сидел, ожидая, пока все остальное закончится, прежде чем оно сможет закончить работу.

Вот как я бы написал первую функцию (в Swift 3 вместо 2)

extension PHImageManager { 

    func requestMaximumSizeImage(for asset: PHAsset) -> Observable<UIImage> { 
     return .create { observer in 
      let request = self.requestImage(for: asset, targetSize: PHImageManagerMaximumSize, contentMode: .aspectFit, options: nil, resultHandler: { image, info in 
       if let image = image { 
        observer.onNext(image) 
        observer.onCompleted() 
       } 
       else if let info = info, let error = info[PHImageErrorKey] as? Error { 
        observer.onError(error) 
       } 
      }) 
      return Disposables.create { self.cancelImageRequest(request) } 
     } 
    } 
} 

Вы увидите, что я бы сделал это расширение на PHImageManager вместо свободной функции, но это просто разница в стиле. Функциональные различия заключаются в том, что мой код испустит ошибку, если ошибки основного запроса не будут устранены, и отменит запрос, если подписчики будут освобождены под залог до завершения запроса. Кроме того, он не выполняет преобразование JPEG. Держите эти операции малого и сделать преобразование JPEG внутри карты, как это:

let imagesData = review.photos.toObservable().flatMap { 
     self.imageMgr.requestMaximumSizeImage(for: $0) 
    }.map { 
     UIImageJPEGRepresentation($0, 1.0) 
    }.filter { $0 != nil }.map { $0! } 

Приведенный выше код запрашивает изображения из диспетчера, а затем преобразует их в данные JPEG, отфильтровывая не прошедшие преобразование. imagesData - тип Observable<Data>.

Ваш getPostPhotoObservable в порядке, за исключением завершенной эмиссии, и тот факт, что он не обрабатывает отмену в одноразовой упаковке. Кроме того, вы можете просто вернуть функцию сообщения Observable вместо того, чтобы обернуть результат в ReviewPhotoObject.

Другие предупреждения:

1. , как вы помещаете все это вместе не гарантирует, что ReviewPhotoObject s будет находиться в том же порядке, что и фотографии собираются в (потому что вы не можете гарантировать порядок что загрузка будет завершена.) Чтобы исправить это, при необходимости вам нужно будет использовать concat вместо flatMap.

2. Если какая-либо из загрузок завершилась неудачно, весь конвейер отключится и прервет последующие загрузки. Вероятно, вы должны настроить что-то, чтобы поймать ошибки и сделать что-то подходящее. Либо catchErrorJustReturn, либо catchError в зависимости от ваших требований.