RxJava- RxAndroid form validation on dynamic EditText

Question

I have form that can have variable number of EditText that needs to be validated before form submission. I can perform validation check if EditTexts are fixed in number like following -

Observable<CharSequence> emailObservable = RxTextView.textChanges(editEmail).skip(1);
Observable<CharSequence> passwordObservable = RxTextView.textChanges(editPassword).skip(1);

mFormValidationSubscription = Observable.combineLatest(emailObservable, passwordObservable,
                (newEmail, newPassword) -> {                   
                    boolean emailValid = !TextUtils.isEmpty(newEmail) && android.util.Patterns.EMAIL_ADDRESS.matcher(newEmail).matches();
                    if(!emailValid) {
                        emailInputLayout.setError(getString(R.string.error_invalid_email));
                        emailInputLayout.setErrorEnabled(true);
                    }else {
                        emailInputLayout.setError(null);
                        emailInputLayout.setErrorEnabled(false);
                    }

                    boolean passValid = !TextUtils.isEmpty(newPassword) && newPassword.length() > 4;
                    if (!passValid) {
                        passwordInputLayout.setError(getString(R.string.error_invalid_password));
                        passwordInputLayout.setErrorEnabled(true);
                    } else {
                        passwordInputLayout.setError(null);
                        passwordInputLayout.setErrorEnabled(true);
                    }

                    return emailValid && passValid;
                }).subscribe(isValid ->{
                    mSubmitButton.setEnabled(isValid);
                });

But now as there are variable number of inputs I tried creating a list of Observable<CharSequence> and Observable.combineLatest() but I'm stuck as to proceed with that.

List<Observable<CharSequence>> observableList = new ArrayList<>();

        for(InputRule inputRule : mMaterial.getRules()) {
            View vInputRow = inflater.inflate(R.layout.item_material_input_row, null, false);

            StyledEditText styledEditText = ((StyledEditText)vInputRow.findViewById(R.id.edit_input));
            styledEditText.setHint(inputRule.getName());

            Observable<CharSequence> observable = RxTextView.textChanges(styledEditText).skip(1);
            observableList.add(observable);

            linearLayout.addView(vInputRow);
        }

        Observable.combineLatest(observableList,......); // What should go in place of these "......" 

How can I perform checks for a valid charsequence for each input field. I looked into flatMap(), map(), filter() methods but I don't know how to use them.

Answer 1

I used R. Zagórski's answer as guide on how to do this with Kotlin

This is what worked for me in the end.

        val ob1 = RxTextView.textChanges(field1).skip(1)
        val ob2 = RxTextView.textChanges(field2).skip(1)
        val ob3 = RxTextView.textChanges(field3).skip(1)

        val observableList = arrayListOf<Observable<CharSequence>>()

        observableList.add(ob1)
        observableList.add(ob3)

        val formValidator = Observable.combineLatest(observableList, {
            var isValid = true
            it.forEach {
                val string = it.toString()
                if (string.isEmpty()) {
                    isValid = false
                }
            }
            return@combineLatest isValid
        })

        formValidator.subscribe { isValid ->
            if (isValid) {
                //do something
            } else {
                //do something
            }
        }

Answer 2

Yes, you process abitrary number of Observables in .combineLatest(), but there is still workaround. I got interested in this problem and came up with following solution- we can store information about some data source - last value and source ID (String and resource id) and tunnell all data into some common pipe. For that we can use PublishSubject. We also need to track connection state, for that we should save Subscription to each source on subscription and sever it when we unsubscribe from that source. We store last data from each source, so we can tell user what source just emitted new value, callback will only contain source id. User can get last value of any source by source id. I came up with the following code:

import android.util.Log;
import android.widget.EditText;

import com.jakewharton.rxbinding.widget.RxTextView;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.Map;
import rx.Observable;
import rx.Subscription;
import rx.functions.Action1;
import rx.subjects.PublishSubject;

public class MultiSourceCombinator {
    String LOG_TAG = MultiSourceCombinator.class.getSimpleName();
    /**
     * We can't handle arbitrary number of sources by CombineLatest, but we can pass data along
     * with information about source (sourceId)
     */
    private static class SourceData{
        String data = "";
        Integer sourceId = 0;
    }

    /**
     * Keep id of source, subscription to that source and last value emitted
     * by source. This value is passed when source is attached
     */
    private class SourceInfo{
        Subscription sourceTracking;
        Integer sourceId;
        SourceData lastData;

        SourceInfo(int sourceId, String data){
            this.sourceId = sourceId;
            // initialize last data with empty value
            SourceData d = new SourceData();
            d.data = data;
            d.sourceId = sourceId;
            this.lastData = d;
        }
    }

    /**
     * We can tunnel data from all sources into single pipe. Subscriber can treat it as
     * Observable<SourceData>
     */
    private PublishSubject<SourceData> dataDrain;

    /**
     * Stores all sources by their ids.
     */
    Map<Integer, SourceInfo> sources;

    /**
     * Callback, notified whenever source emit new data. it receives source id.
     * When notification is received by client, it can get value from source by using
     * getLastSourceValue(sourceId) method
     */
    Action1<Integer> sourceUpdateCallback;

    public MultiSourceCombinator(){
        dataDrain = PublishSubject.create();
        sources = new HashMap<>();
        sourceUpdateCallback = null;
        // We have to process data, ccoming from common pipe
        dataDrain.asObservable()
                .subscribe(newValue -> {
                    if (sourceUpdateCallback == null) {
                        Log.w(LOG_TAG, "Source " + newValue.sourceId + "emitted new value, " +
                                "but used did't set callback ");
                    } else {
                        sourceUpdateCallback.call(newValue.sourceId);
                    }
                });
    }

    /**
     * Disconnect from all sources (sever Connection (s))
     */
    public void stop(){
        Log.i(LOG_TAG, "Unsubscribing from all sources");
        // copy references to aboid ConcurrentModificatioinException
        ArrayList<SourceInfo> t = new ArrayList(sources.values());
        for (SourceInfo si : t){
            removeSource(si.sourceId);
        }
        // right now there must be no active sources
        if (!sources.isEmpty()){
            throw new RuntimeException("There must be no active sources");
        }
    }

    /**
     * Create new source from edit field, subscribe to this source and save subscription for
     * further tracking.
     * @param editText
     */
    public void addSource(EditText editText, int sourceId){
        if (sources.containsKey(sourceId)){
            Log.e(LOG_TAG, "Source with id " + sourceId + " already exist");
            return;
        }
        Observable<CharSequence> source = RxTextView.textChanges(editText).skip(1);
        String lastValue = editText.getText().toString();
        Log.i(LOG_TAG, "Source with id " + sourceId + " has data " + lastValue);
        // Redirect data coming from source to common pipe, to do that attach source id to
        // data string
        Subscription sourceSubscription = source.subscribe(text -> {
            String s = new String(text.toString());
            SourceData nextValue = new SourceData();
            nextValue.sourceId = sourceId;
            nextValue.data = s;
            Log.i(LOG_TAG, "Source " + sourceId + "emits new value: " + s);
            // save vlast value
            sources.get(sourceId).lastData.data = s;
            // pass new value down pipeline
            dataDrain.onNext(nextValue);
        });
        // create SourceInfo
        SourceInfo sourceInfo = new SourceInfo(sourceId, lastValue);
        sourceInfo.sourceTracking = sourceSubscription;
        sources.put(sourceId, sourceInfo);
    }

    /**
     * Unsubscribe source from common pipe and remove it from list of sources
     * @param sourceId
     * @throws IllegalArgumentException
     */
    public void removeSource(Integer sourceId) throws IllegalArgumentException {
        if (!sources.containsKey(sourceId)){
            throw new IllegalArgumentException("There is no source with id: " + sourceId);
        }
        SourceInfo si = sources.get(sourceId);
        Subscription s = si.sourceTracking;
        if (null != s && !s.isUnsubscribed()){
            Log.i(LOG_TAG, "source " + sourceId + " is active, unsubscribing from it");
            si.sourceTracking.unsubscribe();
            si.sourceTracking = null;
        }
        // source is disabled, remove it from list
        Log.i(LOG_TAG, "Source " + sourceId + " is disabled ");
        sources.remove(sourceId);
    }

    /**
     * User can get value from any source by using source ID.
     * @param sourceId
     * @return
     * @throws IllegalArgumentException
     */
    public String getLastSourceValue(Integer sourceId) throws IllegalArgumentException{
        if (!sources.containsKey(sourceId)){
            throw new IllegalArgumentException("There is no source with id: " + sourceId);
        }
        String lastValue = sources.get(sourceId).lastData.data;
        return lastValue;
    }

    public void setSourceUpdateCallback(Action1<Integer> sourceUpdateFeedback) {
        this.sourceUpdateCallback = sourceUpdateFeedback;
    }
}

And we can use it in UI like this:

import android.app.Activity; 
import android.os.Bundle; 
import android.util.Log; 
import android.widget.EditText; 
import android.widget.Toast;

import butterknife.BindView; 
import butterknife.ButterKnife;

public class EdiTextTestActivity extends Activity {

    @BindView(R.id.aet_et1)
    public EditText et1;
    @BindView(R.id.aet_et2)
    public EditText et2;
    @BindView(R.id.aet_et3)
    public EditText et3;

    private MultiSourceCombinator multiSourceCombinator;

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_edit_text_test);
        ButterKnife.bind(this);

        multiSourceCombinator = new MultiSourceCombinator();
        multiSourceCombinator.setSourceUpdateCallback(id -> {
            Toast.makeText(EdiTextTestActivity.this, "New value from source: " + id  + " : " +
            multiSourceCombinator.getLastSourceValue(id), Toast.LENGTH_SHORT).show();
        });
    }

    @Override
    protected void onPause() {
        // stop tracking all fields
        multiSourceCombinator.stop();
        super.onPause();
    }

    @Override
    protected void onResume() {
        super.onResume();
        // Register fields
        multiSourceCombinator.addSource(et1, R.id.aet_et1);
        multiSourceCombinator.addSource(et2, R.id.aet_et2);
        multiSourceCombinator.addSource(et3, R.id.aet_et3);
    } 
}