Using RxJava for email login validation, an observable is emitting twice

Question

I'm making a simple login form (email and password) to try and bolster my reactive programming skillset. I'm having some trouble getting the email field validation to work the way I want it.

Here's my code:

    final Observable<CharSequence> email = RxTextView.textChanges(emailView);

    Observable<Boolean> emailIsValid = email.map(new Func1<CharSequence, Boolean>() {
        @Override
        public Boolean call(CharSequence charSequence) {
            Log.d("asdf", "emailIsValid call: " + charSequence);
            return Pattern.matches(Patterns.EMAIL_ADDRESS.pattern(), charSequence);
        }
    });
    RxView.focusChanges(emailView)
            .withLatestFrom(emailIsValid, new Func2<Boolean, Boolean, Boolean>() {
                @Override
                public Boolean call(Boolean hasFocus, Boolean emailIsValid) {
                    return (!hasFocus && !emailIsValid);
                }
            })
            .subscribe(new Action1<Boolean>() {
                @Override
                public void call(Boolean showError) {
                    if (showError) {
                        emailInputLayout.setError("Enter a valid email");
                    } else {
                        emailInputLayout.setError(null);
                    }
                }
            });
    Observable<CharSequence> password = RxTextView.textChanges(passwordView);

    Observable.combineLatest(emailIsValid, password,
            new Func2<Boolean, CharSequence, Boolean>() {
                @Override
                public Boolean call(Boolean emailIsValid, CharSequence password) {
                    Log.d("asdf", "valid: " + emailIsValid + ", password: " + password);
                    return (emailIsValid && password.length() > 0);
                }
            })
            .subscribe(RxView.enabled(loginButton));

And here's the log:

emailIsValid call: emailIsValid call: valid: false, password: // I type 'j' emailIsValid call: j emailIsValid call: j valid: false, password: // I type 'a' emailIsValid call: ja emailIsValid call: ja valid: false, password:

As you can see, emailIsValid is called twice every time I type a character, which means it's doing a regex match twice, which is kind of wasteful.

I looked up how I could make emailIsValid only call once per change, no matter how many subscribers it has, and I found the share() method. Here's what happens when I add .share() to the end of emailIsValid's declaration:

emailIsValid call: // I type 'j' emailIsValid call: j valid: false, password: // I type 'a' emailIsValid call: ja valid: false, password:

That solves the problem, but it causes another: There is no initial emit by emailIsValid to the combineLatest function at the end, so the Login button starts enabled, when it should be disabled (grayed out).

What's the cleanest way to solve this? I think I want it to behave like a BehaviorSubject, but I'm not sure if that's the best way to do it.

Answer 1

I think what is going on here is the following:

• The first subscribe - the one at the end of RxView.focusChange()... - causes a Subscription to emailIsValid (and therefore also to email).

• email will then immediately emit the current content of the TextView as its first item, which in turn goes through emailIsValid and share and on to the first Subscriber (i. e. the withLatestFrom operator).

• Some time later the combineLatest causes another Subscription to emailIsValid. Since emailIsValid is shared, this Subscription does not "go through" to email and therefore each item will still be emitted only once.

• The problem is now that share behaves like a PublishSubject: It just emits any future events to all Subscribers, but it does not replay any of the past ones.

In sum this means: When the second Subscriber (the combineLatest) arrives, the initial value is already past - it was emitted right after the first subscription. And the next value will only arrive when you change the content of the TextView.

Solution: Try replay(1).refCount() instead of share() at the end of emailIsValid - that should ensure that each new Subscriber also receives the last previous evaluation result, as well as all future ones.

I hope that solves the problem and that my explanation makes sense.

Answer 2

You can use publish() and connect().

val email = RxTextView.textChanges(emailEditText)
val emailIsValid = email.map { charSequence ->
  Log.d("asdf", "emailIsValid call: " + charSequence)
  Pattern.matches(Patterns.EMAIL_ADDRESS.pattern(), charSequence)
}.publish()
RxView.focusChanges(emailEditText)
    .withLatestFrom(emailIsValid) { hasFocus, emailIsValid ->
      (!hasFocus && !emailIsValid)
    }
    .subscribe { showError ->
      if (showError) {
        Log.d("asdf", "error")
      }
    }
val password = RxTextView.textChanges(passwordEditText)
Observable.combineLatest(emailIsValid, password) { emailIsValid, password ->
  Log.d("asdf", "valid: $emailIsValid, password: $password")
  (emailIsValid && password.length > 0)
}.subscribe(RxView.enabled(button))
emailIsValid.connect()

Or just switch the order of your subscribe because it causes the issue.

val email = RxTextView.textChanges(emailEditText)
val emailIsValid = email.map { charSequence ->
  Log.d("asdf", "emailIsValid call: " + charSequence)
  Pattern.matches(Patterns.EMAIL_ADDRESS.pattern(), charSequence)
}.share()
val password = RxTextView.textChanges(passwordEditText)
Observable.combineLatest(emailIsValid, password) { emailIsValid, password ->
  Log.d("asdf", "valid: $emailIsValid, password: $password")
  (emailIsValid && password.length > 0)
}.subscribe(RxView.enabled(button))
RxView.focusChanges(emailEditText)
    .withLatestFrom(emailIsValid) { hasFocus, emailIsValid ->
      (!hasFocus && !emailIsValid)
    }
    .subscribe { showError ->
      if (showError) {
        Log.d("asdf", "error")
      }
    }

Note: code is in Kotlin and more info about publish/connect is at http://www.introtorx.com/content/v1.0.10621.0/14_HotAndColdObservables.html#PublishAndConnect

Your problem is similar to what mention there in PublishAndConnect section.:

The second subscription subscribes late and misses the first publication. We could move the invocation of the Connect() method until after all subscriptions have been made. That way, even with the call to Thread.Sleep we will not really subscribe to the underlying until after both subscriptions are made.