RxJava 2 Disposable on activity destroy

Question

I am new to RxJava/2, doing the form validation using Observable.

How can I dispose my Observable when activity destroy ?

My CODE:

     private Observable<CharSequence> passwordChangeObservable = 
RxTextView.textChanges(passwordTxt);

    passwordChangeObservable
                    .debounce(400, TimeUnit.MILLISECONDS)
                    .map(this::isValidPassword)
    //                .distinctUntilChanged()
                    .subscribeOn(Schedulers.io()) // Or Schedulers.newThread()
                    .observeOn(AndroidSchedulers.mainThread())
                    .subscribe(new Observer<Boolean>() {

                        @Override
                        public void onSubscribe(Disposable d) {

                        }

                        @Override
                        public void onNext(Boolean aBoolean) {

                        }

                        @Override
                        public void onError(Throwable e) {

                        }

                        @Override
                        public void onComplete() {

                        }
                    });

 private Boolean isValidPassword(CharSequence value) {
        return value.toString().matches("^(?=.*\\d).{4,8}$");
    } 

This one not returning anything to hold the reference and dispose on cleanup?

Answer 1

as @Blackbelt suggested, there are a few overloaded versions of subscribe() (refer to the docs).

• several that take combinations of Consumer and Action
• one that takes a Observer

the former return a Disposable instance whereby a subscription can be terminated; the latter does not. so if you'd like dispose of your stream in onDestroy(), you should change the version of subscribe() you're using.

structurally, it would look something like this:

public class Blah extends AppCompatActivity {

    private EditText passwordTxt;

    private Disposable disposable;

    @Override
    protected void onCreate(@Nullable Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);

        disposable = RxTextView.textChanges(passwordTxt)
                .debounce(400, TimeUnit.MILLISECONDS)
                .map(this::isValidPassword)
                .subscribeOn(Schedulers.io()) // Or Schedulers.newThread()
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(
                        new Consumer<Boolean>() {
                            @Override
                            public void accept(Boolean aBoolean) throws Exception {
                                // onNext
                            }
                        },
                        new Consumer<Throwable>() {
                            @Override
                            public void accept(Throwable throwable) throws Exception {
                                // onError
                            }
                        },
                        new Action() {
                            @Override
                            public void run() throws Exception {
                                // onComplete
                            }
                        },
                        new Consumer<Disposable>() {
                            @Override
                            public void accept(Disposable disposable) throws Exception {
                                // onSubscribe
                            }
                        }
                );
    }

    @Override
    protected void onDestroy() {
        super.onDestroy();

        if(disposable != null && !disposable.isDisposed()) {
            disposable.dispose();
        }
    }

    private Boolean isValidPassword(CharSequence value) {
        return value.toString().matches("^(?=.*\\d).{4,8}$");
    }

}

(although lambdas can really shorten, and thus improve the readability of your code, i've opted not to use them so as to clearly illustrate the types in use).