package me.ingeni.rxandroidexample;

import android.os.SystemClock;
import android.support.v7.app.AppCompatActivity;
import android.os.Bundle;
import android.util.Log;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

import java.util.List;
import java.util.NoSuchElementException;
import java.util.concurrent.Callable;
import java.util.logging.Logger;

import io.reactivex.BackpressureStrategy;
import io.reactivex.Flowable;
import io.reactivex.FlowableEmitter;
import io.reactivex.Maybe;
import io.reactivex.Observable;
import io.reactivex.ObservableSource;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.annotations.NonNull;
import io.reactivex.disposables.CompositeDisposable;
import io.reactivex.functions.Action;
import io.reactivex.functions.Consumer;
import io.reactivex.observers.DisposableObserver;
import io.reactivex.schedulers.Schedulers;

public class MainActivity extends AppCompatActivity {

    private final CompositeDisposable disposables = new CompositeDisposable();

    private static final String TAG = "MainActivity";

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);

        Observable.just("one", "two", "three", "four", "five")
                .subscribeOn(Schedulers.newThread())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(@NonNull String s) throws Exception {
                        Log.d("MainActivity @@@ : ", "s : " + s);
                    }
                });

        Flowable.just(1, 2, 3)
                .doOnCancel(new Action() {
                    @Override
                    public void run() throws Exception {
                        Log.d("MainActivity @@@ : ", "doOnCancel");
                    }
                })
                .take(2)
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(@NonNull Integer integer) throws Exception {
                        Log.d("MainActivity @@@ : ", "integer : " + integer);
                    }
                });

        Subscriber<Integer> subscriber = new Subscriber<Integer>() {
            @Override
            public void onSubscribe(Subscription s) {
                s.request(Long.MAX_VALUE);
            }

            public void onNext(Integer t) {
                if (t == 1) {
                    throw new IllegalArgumentException();
                }
            }

            public void onError(Throwable e) {
                if (e instanceof IllegalArgumentException) {
                    throw new UnsupportedOperationException();
                }
            }

            public void onComplete() {
                throw new NoSuchElementException();
            }
        };
        Flowable.just(1).subscribe(subscriber);

        List<Integer> list = Flowable.range(1, 100).toList().blockingGet();
        Integer i = Flowable.range(100, 100).blockingLast();

        onRunSchedulerExampleButtonClicked();
    }

    void onRunSchedulerExampleButtonClicked() {
        disposables.add(sampleObservable()
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribeWith(new DisposableObserver<String>() {
                    @Override
                    public void onComplete() {
                        Log.d(TAG, "onComplete()");
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.e(TAG, "onError()", e);
                    }

                    @Override
                    public void onNext(String string) {
                        Log.d(TAG, "onNext(" + string + ")");
                    }
                }));
    }

    static Observable<String> sampleObservable() {
        return Observable.defer(new Callable<ObservableSource<? extends String>>() {
            @Override
            public ObservableSource<? extends String> call() throws Exception {
                SystemClock.sleep(5000);
                return Observable.just("one", "two", "three", "four", "five");
            }
        });
    }

}

@@ -811,24 +811,65 @@
 s.Consumer;%0A
+import io.reactivex.functions.Predicate;%0A
 import io.re
@@ -2322,32 +2322,1049 @@
            %7D);%0A%0A
+        Flowable.just(1, 2, 3)%0A                .filter(new Predicate%3CInteger%3E() %7B%0A                    @Override%0A                    public boolean test(@NonNull Integer integer) throws Exception %7B%0A                        if (integer != 1) %7B%0A                            return false;%0A                        %7D%0A                        Log.d(%22MainActivity @@@ : %22, %22integer : %22 + integer);%0A                        return true;%0A                    %7D%0A                %7D)%0A                .doOnCancel(new Action() %7B%0A                    @Override%0A                    public void run() throws Exception %7B%0A                        Log.d(%22MainActivity @@@ : %22, %22doOnCancel%22);%0A                    %7D%0A                %7D)%0A                .take(2)%0A                .subscribe(new Consumer%3CInteger%3E() %7B%0A                    @Override%0A                    public void accept(@NonNull Integer integer) throws Exception %7B%0A                        Log.d(%22MainActivity @@@ : %22, %22integer : %22 + integer);%0A                    %7D%0A                %7D);%0A%0A
         Subscrib
@@ -5438,10 +5438,11 @@
 %0A    %7D%0A%0A
+%0A
 %7D%0A

Add example