RxSwift errors dispose of subscriptions

Question

I have been experimenting with some new swift architectures and patterns and I have noticed a strange issue with RxSwift where it seems if I am making a service call and an error occurs - e.g. user enters wrong password - then it seems to dispose of my subscriptions so I cannot make the service call again

I am unsure as to why this happening. I made a quick mini project demonstrating the issue with a sample login app.

My ViewModel looks like this

import RxSwift
import RxCocoa
import RxCoordinator
import RxOptional
extension LoginModel : ViewModelType {
    struct Input {
        let loginTap : Observable<Void>
        let password : Observable<String>
    }

    struct Output {
        let validationPassed : Driver<Bool>
        let loginActivity : Driver<Bool>
        let loginServiceError : Driver<Error>
        let loginTransitionState : Observable<TransitionObservables>
    }

    func transform(input: LoginModel.Input) -> LoginModel.Output {
        // check if email passes regex
        let isValid = input.password.map{(val) -> Bool in
            UtilityMethods.isValidPassword(password: val)
        }

        // handle response
        let loginResponse = input.loginTap.withLatestFrom(input.password).flatMapLatest { password in
            return self.service.login(email: self.email, password: password)
        }.share()

        // handle loading
        let loginServiceStarted = input.loginTap.map{true}
        let loginServiceStopped = loginResponse.map{_ in false}
        let resendActivity = Observable.merge(loginServiceStarted, loginServiceStopped).materialize().map{$0.element}.filterNil()

        // handle any errors from service call
        let serviceError = loginResponse.materialize().map{$0.error}.asDriver(onErrorJustReturn: RxError.unknown).filterNil()

        let loginState = loginResponse.map { _ in
            return self.coordinator.transition(to: .verifyEmailController(email : self.email))
        }

        return Output(validationPassed : isValid.asDriver(onErrorJustReturn: false), loginActivity: resendActivity.asDriver(onErrorJustReturn: false), loginServiceError: serviceError, loginTransitionState : loginState)
    }
}

class LoginModel {
    private let coordinator: AnyCoordinator<WalkthroughRoute>
    let service : LoginService
    let email : String
    init(coordinator : AnyCoordinator<WalkthroughRoute>, service : LoginService, email : String) {
        self.service = service
        self.email = email
        self.coordinator = coordinator
    } 
}

And my ViewController looks like this

import UIKit
import RxSwift
import RxCocoa
class TestController: UIViewController, WalkthroughModuleController, ViewType {

    // password
    @IBOutlet var passwordField : UITextField!

    // login button
    @IBOutlet var loginButton : UIButton!

    // disposes of observables
    let disposeBag = DisposeBag()

    // view model to be injected
    var viewModel : LoginModel!

    // loader shown when request is being made
    var generalLoader : GeneralLoaderView?

    override func viewDidLoad() {
        super.viewDidLoad()

    }
    // bindViewModel is called from route class
    func bindViewModel() {
        let input = LoginModel.Input(loginTap: loginButton.rx.tap.asObservable(), password: passwordField.rx.text.orEmpty.asObservable())

        // transforms input into output
        let output = transform(input: input)

        // fetch activity
        let activity = output.loginActivity

        // enable/disable button based on validation
        output.validationPassed.drive(loginButton.rx.isEnabled).disposed(by: disposeBag)

        // on load
        activity.filter{$0}.drive(onNext: { [weak self] _ in
            guard let strongSelf = self else { return }
            strongSelf.generalLoader = UtilityMethods.showGeneralLoader(container: strongSelf.view, message: .Loading)
        }).disposed(by: disposeBag)

        // on finish loading
        activity.filter{!$0}.drive(onNext : { [weak self] _ in
            guard let strongSelf = self else { return }
            UtilityMethods.removeGeneralLoader(generalLoader: strongSelf.generalLoader)
        }).disposed(by: disposeBag)

        // if any error occurs
        output.loginServiceError.drive(onNext: { [weak self] errors in
            guard let strongSelf = self else { return }

            UtilityMethods.removeGeneralLoader(generalLoader: strongSelf.generalLoader)

            print(errors)
        }).disposed(by: disposeBag)

        // login successful
        output.loginTransitionState.subscribe().disposed(by: disposeBag)
    }
}

My service class

import RxSwift
import RxCocoa

struct LoginResponseData : Decodable {
    let msg : String?
    let code : NSInteger
}

    class LoginService: NSObject {
        func login(email : String, password : String) -> Observable<LoginResponseData> {
            let url = RequestURLs.loginURL

            let params = ["email" : email,
                          "password": password]

            print(params)

            let request = AFManager.sharedInstance.setupPostDataRequest(url: url, parameters: params)
            return request.map{ data in
                return try JSONDecoder().decode(LoginResponseData.self, from: data)
            }.map{$0}
        }
    }

If I enter valid password, request works fine. If I remove the transition code for testing purposes, I could keep calling the login service over and over again as long as password is valid. But as soon as any error occurs, then the observables relating to the service call get disposed of so user can no longer attempt the service call again

So far the only way I have found to fix this is if any error occurs, call bindViewModel again so subscriptions are setup again. But this seems like very bad practice.

Any advice would be much appreciated!

Answer 1

At the place where you make the login call:

let loginResponse = input.loginTap
    .withLatestFrom(input.password)
    .flatMapLatest { [unowned self] password in
        self.service.login(email: self.email, password: password)
    }
    .share()

You can do one of two things. Map the login to a Result<T> type.

let loginResponse = input.loginTap
    .withLatestFrom(input.password)
    .flatMapLatest { [unowned self] password in
        self.service.login(email: self.email, password: password)
            .map(Result<LoginResponse>.success)
            .catchError { Observable.just(Result<LoginResponse>.failure($0)) }
    }
    .share()

Or you can use the materialize operator.

let loginResponse = input.loginTap
    .withLatestFrom(input.password)
    .flatMapLatest { [unowned self] password in
        self.service.login(email: self.email, password: password)
            .materialize()
    }
    .share()

Either method changes the type of your loginResponse object by wrapping it in an enum (either a Result<T> or an Event<T>. You can then deal with errors differently than you do with legitimate results without breaking the Observable chain and without loosing the Error.

Another option, as you have discovered is to change the type of loginResponse to an optional but then you loose the error object.

Answer 2

The behavior is not strange, but works as expected: As stated in the official RxSwift documentation documentation: "When a sequence sends the completed or error event all internal resources that compute sequence elements will be freed." For your example that means, a failed login attempt, will cause method func login(email : String, password : String) -> Observable<LoginResponseData> to return an error, i.e. return Observable<error>, which will:

• on the one hand fast forward this error to all its subscribers (which will be done your VC)
• on the other hand dispose the observable

To answer your question, what you can do other than subscribing again, in order to maintain the subscription: You could just make use of .catchError(), so the observable does not terminate and you can decide yourself what you want to return after an error occurs. Note, that you can also check the error for a specific error domain and return errors only for certain domains.

I personally see the responsibility of the error handling in the hand of the respective subscribers, i.e. in your case your TestController (so you could use .catchError() there), but if you want to be sure the observable returned from from func login(email : String, password : String) -> Observable<LoginResponseData> does not even fast forward any errors for all subscriptions, you could also use .catchError() here, although I'd see issues for potential misbehaviors.