RxSwift errors dispose of subscriptions

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!

Solution

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.