Observable extension with generic type

user_2914850 0
Viewed 135

Context

I want to wrap the Alamofire.upload into an observable and having info regarding the upload progress.
For that I have created a custom UploadElement that is an enum representing either the progress and the value or the result. So far I have:

enum UploadElement<Result> where Result: Codable {
    case progress(Double)
    case response(Result)
}

private func buildUploadRequest(url: URL, parts: [Data]) -> Observable<UploadRequest> {
    let uploadRequest = manager.upload(
        multipartFormData: { multipartFormData in /* build multipart */ },
        to: url
    )
    return Observable.just(uploadRequest)
}

func upload<Result: Codable>(url: URL, parts: [Data]) -> Observable<UploadElement<Result>> {
    buildUploadRequest(url: url, parts: parts)
        .flatMap { request in
            Observable<UploadElement<Result>>.create { observer in
                request.response { response in
                            do {
                                observer.on(.next(.response(/* decode here */)))
                                observer.on(.completed)
                            } catch let error {
                                observer.on(.error(error))
                            }
                }.uploadProgress { progress in
                    observer.on(.next(.progress(progress.fractionCompleted)))
                }
                .resume()

              return Disposable.create { request.cancel() }
            }
        }
}

Now I would like to have an extension on an Observable<UploadEment<Result>> to have a nicer way to be notified.

Basically it would be:

service.upload(url: ..., parts: ...)
    .progress { progress in /* */ }
    .result { result in /* */ }
    .subscribe()
    .dispose(by: disposeBag)

To do that I tried:

extension ObservableType where Element == UploadElement<Resource> {
    func progress(progressCompletion: @escaping (Double) -> Void) -> Self {
        return self.do(onNext: { element in 
            switch element {
            case .progress(let progress): progressCompletion(progress)
            case .response: return
            }
        })
    }

    func result(resultCompletion: @escaping (Result) -> Void) -> Self {
        return self.do(onNext: { element in 
            switch element {
            case .response(let result): resultCompletion(result)
            case .progress: return
            }
        })
    }
}

I tried multiple variation of that but the errors that I get are:

  1. Cannot find 'Result in scope'
  2. Reference to generic type ... required argument

Is it possible to achieve something like that?

rx-swift swift
2 Answers

You just need to move the where clause from class scope down to function scope (shown below).

That said, I don't think breaking out of the monad like this in the middle of a stream is "a nicer way to be notified".

Better would be to break your Observable into two streams and subscribe to each of them:

extension ObservableType {
    func progress<Resource>() -> Observable<Double> where Element == UploadElement<Resource> {
        self.compactMap { element in
            switch element {
            case let .progress(progress):
                return progress
            case .response:
                return nil
            }
        }
    }

    func result<Resource>() -> Observable<Resource>  where Element == UploadElement<Resource> {
        self.compactMap { element in
            switch element {
            case .progress:
                return nil
            case let .response(resource):
                return resource
            }
        }
    }
}

With the above you can now do something like this:

let response = service.upload(url: ..., parts: ...)
    .share()

response
    .progress()
    .subscribe(onNext: { progress in /*...*/ })
    .disposed(by: disposeBag)

response
    .result()
    .subscribe(onNext: { result in /*...*/ })
    .dispose(by: disposeBag)

Now you don't have any empty subscribes.

user_5064410

I found something that is working:

extension ObservableType {

    func progress<O: Codable>(progressCompletion: @escaping (Double) -> Void) -> Observable<UploadElement<O>> where Element == UploadElement<O> {
        return self.do(onNext: { element in
            if case .progress(let progress) = element {
                progressCompletion(progress)
            }
        })
    }

    func response<O: Codable>(responseCompletion: @escaping (O) -> Void) -> Observable<UploadElement<O>> where Element == UploadElement<O> {
        return self.do(onNext: { element in
            if case .response(let response) = element {
                 responseCompletion(response)
            }
        })
    }
}

Now I can use the "planned" api:

service.update(data: /* ... */)
       .progress { progress in /* */ }
       .response { result in /* */ }
       .subscribe(
           onError: { error in /* */ }
       )
       .dispose(by: disposeBag)

However as Daniel mentioned this might not be the "nicer way of being notified".

user_29148500
Related