pyadav

public void doubleClickDetect(View view){
        Observable<Void> observable = RxView.clicks(view).share();
        observable.buffer(observable.debounce(200, TimeUnit.MILLISECONDS))
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Action1<List<Void>>() {
                    @Override
                    public void call(List<Void> voids) {
                        if(voids.size() >= 2){
                        	//double click detected
                        }

pyadav

package com.example.rx_okhttp;

import com.squareup.okhttp.OkHttpClient;
import org.apache.http.*;
import org.apache.http.entity.InputStreamEntity;
import org.apache.http.message.BasicHeader;
import org.apache.http.message.BasicHttpResponse;
import rx.Observable;
import rx.Observer;
import rx.Scheduler;

pyadav

A small collection of rx-java resources

##Blog posts Intro to rxjava

• https://gist.github.com/staltz/868e7e9bc2a7b8c1f754
• http://joluet.github.io/blog/2014/07/07/rxjava-retrofit/
• http://www.node.mu/2014/07/02/using-retrofit-and-rxjava-to-interact-with-web-services-on-android/
• http://techblog.netflix.com/2013/02/rxjava-netflix-api.html
• http://nerds.weddingpartyapp.com/tech/2015/01/05/debouncedbuffer-used-in-rxbus-example/ EventBus + rxjava

pyadav

public class FacebookHomeProvider {
	protected final PublishSubject<FbPost> behaviorSubject;
	private Request request;

	public FacebookHomeProvider() {
		behaviorSubject = PublishSubject.create();
		behaviorSubject.subscribeOn(Schedulers.io());
	}

    /**

pyadav

//返回一个Obserable,侦听发给heat-sensor的信息
Observable<Double> observable = vertx.eventBus().
    <Double>consumer("heat-sensor").
    bodyStream().
    toObservable();

//这个obserable持续1秒钟，并对数据进行map reduce操作
observable.
    buffer(1, TimeUnit.SECONDS).
    map(samples -> samples.

pyadav

public class HomePresenterImpl implements HomePresenter {
	private static final String TAG = HomePresenterImpl.class.getSimpleName();
	private final HomeScreen screen;
	private final DataProvider provider;
	private Subscription subscription1;

	public HomePresenterImpl(HomeScreenImpl screen, DataProvider dataProvider) {
		this.screen = screen;
		this.provider = dataProvider;
	}

pyadav

public class GcmRegistrationIntentService extends IntentService {

    private static final String TAG = "RegIntentService";
    private static final String[] TOPICS = {"global"};

    private PushNotificationManager pushNotificationManager;
    private GcmRequestManager gcmRequestManager;
    private InstanceID instanceID;

    public GcmRegistrationIntentService() {

pyadav

final Scheduler.Worker worker = Schedulers.newThread().createWorker();
        worker.schedule(new Action0() {
            @Override
            public void call() {
                if (!mAnimationSet.isStarted()) {
                    removeViewAt(0);
                    worker.unsubscribe();
                } else {
                    worker.schedule(this, 500, TimeUnit.MILLISECONDS);
                }

pyadav

#build.gradle
#
#    compile 'io.reactivex:rxandroid:1.0.1'
#    compile 'io.reactivex:rxjava:1.0.14'
#    compile 'io.reactivex:rxjava-math:1.0.0'
#    compile 'com.jakewharton.rxbinding:rxbinding:0.2.0'

# rxjava
-keep class rx.schedulers.Schedulers {
    public static <methods>;

pyadav

// slide 75 of this: http://www.slideshare.net/Couchbase/reactive-programmingrxjavaefficientdata-benchristensenmichaelnitschinger

Observable.
  .defer(() -> bucket.get("id")) //create new
  .retryWhen(attempts -> attempts
      .zipWith(Observable.range(1,3), (n, i) -> i) // retry maximum of 3 times
      .flatMap(i -> {
        System.out.println("Delaying retry by " + i + " second(s)");
        return Observable.timer(i, TimeUnit.SECONDS); //delay the resubscribe
      })