Управление несколькими загрузками с помощью ReactiveX (на iOS с помощью Swift и Alamofire)

Я пытаюсь загрузить несколько фотографий на сервер с помощью 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 все еще немного скудна, я надеялся, что кто-нибудь здесь сможет подсказать мне, что я неправильно понимаю.