Java 类rx.functions.Action1 实例源码

private void loadData(final boolean refresh) {
    Observable.timer(5, TimeUnit.SECONDS).compose(new Observable.Transformer<Long, List<String>>() {
        @Override
        public Observable<List<String>> call(Observable<Long> longObservable) {
            return Observable.create(new Observable.OnSubscribe<List<String>>() {
                @Override
                public void call(Subscriber<? super List<String>> subscriber) {
                    subscriber.onNext(pageNo == 0 ? new ArrayList<String>() : AnalogData.analogString(pageNo));
                    subscriber.onCompleted();
                }
            });
        }
    }).compose(RxSchedulers.<List<String>>io2main())
            .subscribe(new Action1<List<String>>() {
                @Override
                public void call(List<String> section2Models) {
                    mBaseAdapter.addItems(section2Models, refresh);
                    stopRefresh(refresh);
                }
            });
}

@Override
public void getData(boolean isRefresh) {
    mDbDao.queryBuilder()
            .where(VideoInfoDao.Properties.DownloadStatus.eq(DownloadStatus.COMPLETE))
            .rx()
            .list()
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Action1<List<VideoInfo>>() {
                @Override
                public void call(List<VideoInfo> videoList) {
                    if (ListUtils.isEmpty(videoList)) {
                        mView.noData();
                    } else {
                        mView.loadData(videoList);
                    }
                }
            });
}

/**
 * Start an async task which can do things beforehand, in background and callback when the job is done on the main thread, and handle the exception with the given action.
 *
 * @param preExecute     action to do beforehand.
 * @param doInBackground action to do in the background.
 * @param doOnFinish     action to do when the job is done.(this is called on main thread)
 * @param onError        action to do when exceptions are thrown.
 * @return the subscription of the task.
 */
public static Subscription asyncTask(final Action0 preExecute, @NonNull final Action0 doInBackground, final Action0 doOnFinish, Action1<Throwable> onError) {
    return Observable.just("Hey nerd! This is an async task.")
            .subscribeOn(Schedulers.io())
            .doOnSubscribe(new Action0() {
                @Override
                public void call() {
                    if (preExecute != null) preExecute.call();
                }
            })
            .observeOn(Schedulers.io())
            .doOnNext(Actions.toAction1(doInBackground))
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Action1<String>() {
                @Override
                public void call(String s) {
                    if (doOnFinish != null) doOnFinish.call();
                }
            }, onError == null ? RxActions.onError() : onError);
}

@Override
public void getData(boolean isRefresh) {
    mDbDao.queryBuilder().where(VideoInfoDao.Properties.IsCollect.eq(true))
            .rx()
            .list()
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Action1<List<VideoInfo>>() {
                @Override
                public void call(List<VideoInfo> videoList) {
                    if (ListUtils.isEmpty(videoList)) {
                        mView.noData();
                    } else {
                        mView.loadData(videoList);
                    }
                }
            });
}

/**
 * 重写onTouchEvent， 当用户的手在MyScrollView上面的时候，
 * 直接将MyScrollView滑动的Y方向距离回调给onScroll方法中，当用户抬起手的时候，
 * MyScrollView可能还在滑动，所以当用户抬起手我们隔5毫秒给handler发送消息，在handler处理
 * MyScrollView滑动的距离
 */
@Override
public boolean onTouchEvent(MotionEvent ev) {
    if(onScrollListener != null){
        lastScrollX = this.getScrollX();
        onScrollListener.onScroll(this.getScrollX());
    }
    switch(ev.getAction()){
        case MotionEvent.ACTION_UP:
            Observable.timer(500, TimeUnit.MILLISECONDS, AndroidSchedulers.mainThread())
                    .subscribe(new Action1<Long>() {
                        @Override
                        public void call(Long aLong) {
                            if(onScrollListener != null){
                                onScrollListener.onScrollStop();
                            }
                        }
                    });
            break;
    }
    return super.onTouchEvent(ev);
}

private void initRxBus(){
    addSubscription(RemoveItemEvent.class, new Action1<RemoveItemEvent>() {
        @Override
        public void call(RemoveItemEvent removeItemEvent) {
            ProductEntity removeEntity = removeItemEvent.getRemoveEntity();
            if(DataLayer.getInstance().getDataPool().getSelectedProductEntityList().contains(removeEntity)){
                DataLayer.getInstance().getDataPool().getSelectedProductEntityList().remove(removeEntity);
            }
            mCategoryProductAdapter.updateData(removeEntity);
        }
    }, new Action1<Throwable>() {
        @Override
        public void call(Throwable throwable) {
            throwable.printStackTrace();
        }
    });
}

private void checkRandCodeAndNext() {
    final HttpService service = RetrofitManager.getInstance().getService();
    service.checkRandCode2(randCode, "randp", "", orderParam.getREPEAT_SUBMIT_TOKEN())
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Action1<CheckRandCodeResponse>() {
                @Override
                public void call(CheckRandCodeResponse checkRandCodeResponse) {
                    if (checkRandCodeResponse.getData().getResult().equals("1")) {
                        RxBus.getDefault().post("pass");
                        submitOrderNext(service, orderParam.getREPEAT_SUBMIT_TOKEN());
                    }
                    else {
                        RxBus.getDefault().post("验证码错误");
                        RxBus.getDefault().post("clear");
                        refreshPassCode();
                    }
                }
            });
}

@Override
public void getNewsData(int channelId, final boolean refresh) {
    if (refresh) pageNum = 0;
    mModel.getNewsData(channelId, pageNum, new Action1<NewsList>() {
        @Override
        public void call(NewsList newsList) {
            mView.onNewsData(newsList.getList(), refresh);
            mView.stopLoading();
            if ("True".equals(newsList.getNext())) {
                pageNum = Integer.parseInt(newsList.getNextpage());
            }
            if (refresh) {
                SharedPreManager.getInstance().putNewsList(mContext, newsList);
            }
        }
    });
}

public static <T> Observable.Transformer<T, T> rxCacheBeanHelper(final String key) {
    return new Observable.Transformer<T, T>() {
        @Override
        public Observable<T> call(Observable<T> observable) {
            return observable
                    .subscribeOn(Schedulers.io())//指定doOnNext执行线程是新线程
                    .doOnNext(new Action1<T>() {
                        @Override
                        public void call(final T data) {
                            Schedulers.io().createWorker().schedule(new Action0() {
                                @Override
                                public void call() {
                                    LogUtils.d("get data from network finish ,start cache...");
                                    ACache.get(ReaderApplication.getsInstance())
                                            .put(key, new Gson().toJson(data, data.getClass()));
                                    LogUtils.d("cache finish");
                                }
                            });
                        }
                    })
                    .observeOn(AndroidSchedulers.mainThread());
        }
    };
}

private void throttleFirst() {
   Observable.create(new Observable.OnSubscribe<Integer>() {
        @Override
        public void call(Subscriber<? super Integer> subscriber) {
              for(int i=0;i<10;i++){
                  subscriber.onNext(i);
                  try {
                      Thread.sleep(100);
                  } catch (InterruptedException e) {
                      e.printStackTrace();
                  }
              }
            subscriber.onCompleted();
        }
    }).throttleFirst(200, TimeUnit.MILLISECONDS).subscribe(new Action1<Integer>() {
       @Override
       public void call(Integer integer) {
           Log.d(TAG, "throttleFirst:"+integer);
       }
   });
}

@Override
protected void onLocationPermissionGranted() {
    compositeSubscription = new CompositeSubscription();
    compositeSubscription.add(reactiveLocationProvider.getPlaceById(placeId)
            .subscribe(new Action1<PlaceBuffer>() {
                @Override
                public void call(PlaceBuffer buffer) {
                    Place place = buffer.get(0);
                    if (place != null) {
                        placeNameView.setText(place.getName());
                        placeLocationView.setText(place.getLatLng().latitude + ", " + place.getLatLng().longitude);
                        placeAddressView.setText(place.getAddress());
                    }
                    buffer.release();
                }
            }));
}

private void addGeofence() {
    final GeofencingRequest geofencingRequest = createGeofencingRequest();
    if (geofencingRequest == null) return;

    final PendingIntent pendingIntent = createNotificationBroadcastPendingIntent();
    reactiveLocationProvider
            .removeGeofences(pendingIntent)
            .flatMap(new Func1<Status, Observable<Status>>() {
                @Override
                public Observable<Status> call(Status pendingIntentRemoveGeofenceResult) {
                    return reactiveLocationProvider.addGeofences(pendingIntent, geofencingRequest);
                }
            })
            .subscribe(new Action1<Status>() {
                @Override
                public void call(Status addGeofenceResult) {
                    toast("Geofence added, success: " + addGeofenceResult.isSuccess());
                }
            }, new Action1<Throwable>() {
                @Override
                public void call(Throwable throwable) {
                    toast("Error adding geofence.");
                    Log.d(TAG, "Error adding geofence.", throwable);
                }
            });
}

protected void configurationImp(@NonNull Action1<byte[]> action) {
    mConn.flatMap(rxBleConnection -> rxBleConnection
            .writeCharacteristic(uuidConf, baConf))
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(action, this::onConnectionFailure);

    // Speed up period for compass calibration
    // For some reason temperature sensor does not work at this speed
    if (!uuidConf.equals(UUID.fromString(TempertureProfile.GattConf))) {
        // Period 0x14 = 200ms
        byte[] periodConf = new byte[]{(byte) 0x14};
        mConn.flatMap(rxBleConnection -> rxBleConnection
                .writeCharacteristic(uuidPeri, periodConf))
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(action, this::onConnectionFailure);
    }
}

@Test
public void testPagingCapabilities() {
    PublishSubject<Object> view = PublishSubject.create();
    BehaviorSubject<Integer> nextPageRequests = BehaviorSubject.create();
    final TestObserver<Delivery<Object, String>> testObserver = new TestObserver<>();

    nextPageRequests
        .concatMap(new Func1<Integer, Observable<Integer>>() {
            @Override
            public Observable<Integer> call(Integer targetPage) {
                return targetPage <= requestedPageCount ?
                    Observable.<Integer>never() :
                    Observable.range(requestedPageCount, targetPage - requestedPageCount);
            }
        })
        .doOnNext(new Action1<Integer>() {
            @Override
            public void call(Integer it) {
                requestedPageCount = it + 1;
            }
        })
        .startWith(Observable.range(0, requestedPageCount))
        .concatMap(new Func1<Integer, Observable<String>>() {
            @Override
            public Observable<String> call(final Integer page) {
                return requestPage(page, PAGE_SIZE);
            }
        })
        .compose(new DeliverReplay<Object, String>(view))
        .subscribe(testObserver);

    ArrayList<Delivery<Object, String>> onNext = new ArrayList<>();

    testObserver.assertReceivedOnNext(onNext);

    view.onNext(999);
    addOnNext(onNext, 999, 0, 1, 2);

    testObserver.assertReceivedOnNext(onNext);

    nextPageRequests.onNext(2);
    addOnNext(onNext, 999, 3, 4, 5);

    testObserver.assertReceivedOnNext(onNext);

    view.onNext(null);

    assertEquals(0, testObserver.getOnCompletedEvents().size());
    testObserver.assertReceivedOnNext(onNext);

    nextPageRequests.onNext(3);

    assertEquals(0, testObserver.getOnCompletedEvents().size());
    testObserver.assertReceivedOnNext(onNext);

    view.onNext(9999);
    addOnNext(onNext, 9999, 0, 1, 2, 3, 4, 5, 6, 7, 8);

    assertEquals(0, testObserver.getOnCompletedEvents().size());
    testObserver.assertReceivedOnNext(onNext);
}

@Override
public void onViewAttached(View view) {
    if (!hasAttach) {
        hasAttach = true;
        RxTask.asyncMap(new Func1<String, String>() {
            @Override
            public String call(String s) {
                return extractContent(prefix + id);
            }
        }, new Action1<String>() {
            @Override
            public void call(String summary) {
                if (summary.length() == 2) {
                    getSelfView().getBinding().tvSummary.setText(getString(R.string.douban_staff_no_introduce));
                    getSelfView().getBinding().ivToggle.setVisibility(View.GONE);
                } else {
                    initText(summary);
                }
                RxBus.getInstance().send(true, Constant.LOADING_COMPLETE_SIGNAL);       //通知已经加载完毕
            }
        });
    }
}

public void connect(Action1<? super Subscription> connection) {
    PublishSubscriber<T> ps;
    PublishSubscriber<T> u;
    boolean doConnect;
    do {
        ps = (PublishSubscriber) this.current.get();
        if (ps != null && !ps.isUnsubscribed()) {
            break;
        }
        u = new PublishSubscriber(this.current);
        u.init();
    } while (!this.current.compareAndSet(ps, u));
    ps = u;
    if (ps.shouldConnect.get() || !ps.shouldConnect.compareAndSet(false, true)) {
        doConnect = false;
    } else {
        doConnect = true;
    }
    connection.call(ps);
    if (doConnect) {
        this.source.unsafeSubscribe(ps);
    }
}

@Override
public StringBuffer fill(final StringBuffer sb) {
  query.fill(sb);
  if (fields.length > 0) {
    sb.append(" ORDER BY ");
    visit(fields, new Action1<String>() {
      @Override public void call(String s) {
        sb.append(s);
      }
    }, new Runnable() {
      @Override public void run() {
        sb.append(",");
      }
    });

    sb.append(desc ? " DESC" : " ASC");
  }
  return sb;
}

@Override
protected void initView() {
    swipeRefreshLayout.setColorSchemeResources(R.color.app_bar_bg_dark);
    swipeRefreshLayout.setOnRefreshListener(new SwipeRefreshLayout.OnRefreshListener() {
        @Override
        public void onRefresh() {
            Observable.timer(600, TimeUnit.MILLISECONDS, AndroidSchedulers.mainThread())
                    .subscribe(new Action1<Long>() {
                        @Override
                        public void call(Long aLong) {
                            Bundle bundle = new Bundle();
                            bundle.putSerializable(Constant.FILE_DATA_TYPE, FileDataType.refresh);
                            bundle.putString(Constant.BACK_PATH, adapter.getParentPath());
                            getLoaderManager().destroyLoader(DataType.fileData.ordinal());
                            getLoaderManager().initLoader(DataType.fileData.ordinal(), bundle, adapter);
                        }
                    });
        }
    });
}

@Override
public void loadResolvers(@NonNull final String mimeType) {
    mSubscription.clear();

    Subscription subscription = loadResolversInternal(mimeType)
            .subscribe(new Action1<List<Resolver>>() {
                @Override
                public void call(List<Resolver> resolvers) {
                    if (resolvers == null || resolvers.size() == 0) {
                        loadMediaTypes();
                    } else if (resolvers.size() == 1) {
                        try {
                            mView.showFileContent(resolvers.get(0).loadActivityInfo(mimeType), mFile, mRequestCode);
                        } catch (AppChooserException e) {
                            loadMediaTypes();
                        }
                    } else {
                        mView.showResolvers(resolvers);
                    }
                }
            });
    mSubscription.add(subscription);
}

public void connect(Action1<? super Subscription> connection) {
    ReplaySubscriber<T> ps;
    ReplaySubscriber<T> u;
    boolean doConnect;
    do {
        ps = (ReplaySubscriber) this.current.get();
        if (ps != null && !ps.isUnsubscribed()) {
            break;
        }
        u = new ReplaySubscriber(this.current, (ReplayBuffer) this.bufferFactory.call());
        u.init();
    } while (!this.current.compareAndSet(ps, u));
    ps = u;
    if (ps.shouldConnect.get() || !ps.shouldConnect.compareAndSet(false, true)) {
        doConnect = false;
    } else {
        doConnect = true;
    }
    connection.call(ps);
    if (doConnect) {
        this.source.unsafeSubscribe(ps);
    }
}

@Override
public Observable<T> adapt(final Call<T> call) {

    return Observable.<T>create(SyncOnSubscribe.createSingleState(new Func0<Call<T>>() {
        @Override
        public Call<T> call() {
            return call.clone();
        }
    }, new Action2<Call<T>, Observer<? super T>>() {
        @Override
        public void call(Call<T> callClone, Observer<? super T> observer) {
            try {
                observer.onNext(callClone.execute());
                observer.onCompleted();
            } catch (Throwable throwable) {
                observer.onError(throwable);
            }
        }
    }, new Action1<Call<T>>() {
        @Override
        public void call(Call<T> tCall) {
            tCall.cancel();
        }
    }));
}

@Override
protected void onResume() {
    super.onResume();
    RxBus.getInstance().toObserverable(RxBean.class)
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Action1<RxBean>() {
                @Override
                public void call(RxBean rxBean) {
                    if (rxBean.getPwd() == "") {
                        return;
                    } else {
                        et_password.setText(rxBean.getPwd());
                    }
                }
            });
}

@Override
public void requestData(RequestParamsBean params) {
    Log.i("tag", "KaienPresenter requestData()");
    mSubscription = ApiManager.getInstance().getNewsApi()
            .getIplayCommunityData()
            .compose(this.<IplayCommunity>applySchedulers())
            .subscribe(new Action1<IplayCommunity>() {
                @Override
                public void call(IplayCommunity iplayCommunity) {
                    Log.i("tag", "KaienPresenter call()"+iplayCommunity.getInfo().getDiscuzList());
                    mView.onLoadSuccesed();
                    mView.refreshRecyclerView(iplayCommunity.getInfo().getDiscuzList());

                }
            },mOnError);
}

private void getCommentList(Action1<List<Comment>> onNext) {
    service.getCommentList(id, start, perPage)
            .compose(new ThreadDispatcher<DoubanCommentList>())
            .map(new Func1<DoubanCommentList, List<Comment>>() {
                @Override
                public List<Comment> call(DoubanCommentList list) {
                    return list.comments;
                }
            })
            .subscribe(onNext, new ActionHttpError() {
                @Override
                protected void onError() {
                    setIsLoadingMore(false);
                    hideRefreshing();
                }
            });
}

@Override
public void getVerifyCode() {
    Subscription subscription = Observable.just("")
            .subscribeOn(AndroidSchedulers.mainThread())
            .doOnSubscribe(new Action0() {
                @Override
                public void call() {
                    mLoginView.showLoading("获取验证码...");
                }
            })
            .observeOn(Schedulers.io())
            .subscribe(new Action1<Object>() {
                @Override
                public void call(Object o) {
                    mDataRepository.getVerifyCode("3", getIMEI(mLoginView.getViewContext()), mCallbackImp);
                }
            });

    mSubscriptions.add(subscription);
}

@Override
protected void onProcessLogic() {
    Observable.create(new Observable.OnSubscribe<List<NormalModel>>() {
        @Override
        public void call(Subscriber<? super List<NormalModel>> subscriber) {
            subscriber.onNext(AnalogData.analogNormalModel());
            subscriber.onCompleted();
        }
    }).compose(RxSchedulers.<List<NormalModel>>io2main())
            .subscribe(new Action1<List<NormalModel>>() {
                @Override
                public void call(List<NormalModel> normalModels) {
                    mBaseAdapter.addItems(normalModels);
                    mOnePlusNLayoutAdapter.addItems(normalModels);
                }
            });
}

public void runCode() {

//        window操作符会在时间间隔内缓存结果，类似于buffer缓存一个list集合，区别在于window将这个结果集合封装成了observable

//        http://blog.csdn.net/axuanqq/article/details/50756530
        Observable
                .sequenceEqual(
                 Observable.just("A","B")
                ,Observable.just("A","B"))
                .subscribe(new Action1<Boolean>() {
            @Override
            public void call(Boolean aBoolean) {
                println(aBoolean);
            }
        });

    }

@Override
public void onSuccess(boolean isTop, GankBean data) {
    Log.i(TAG, "onSuccess: ----------------------------------");
    if (isTop) {
        mView.setRefreshData(data);
    } else {
        mView.setMoreGankData(data);
    }
    Observable.timer(2, TimeUnit.SECONDS)
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Action1<Long>() {
                @Override
                public void call(Long aLong) {
                    mView.hideLoading();
                }
            });
}

/**
 * Sign out from both Google and Facebook and disable auto sign in for Smart Lock Password.
 * Require a FragmentActivity
 *
 * @param activity the activity
 * @return a PublishSubject<RxStatus>
 */
public PublishSubject<RxStatus> signOut(FragmentActivity activity) {
    mStatusObserver = PublishSubject.create();

    Observable<RxStatus> rxGoogleSignOut = new RxGoogleAuth.Builder(activity).build().signOut();
    Observable<RxStatus> rxFacebookSignOut = new RxFacebookAuth.Builder(activity).build().signOut();
    Observable<RxStatus> rxSmartLockDisableAutoSignin =
            new RxSmartLockPasswords.Builder(activity).build().disableAutoSignIn();

    Observable.merge(rxGoogleSignOut, rxFacebookSignOut, rxSmartLockDisableAutoSignin)
            .subscribe(new Action1<RxStatus>() {
                @Override
                public void call(RxStatus rxStatus) {
                    mStatusObserver.onNext(rxStatus);
                    mStatusObserver.onCompleted();
                }
            });

    return mStatusObserver;
}

@Override
public Response intercept(Chain chain) throws IOException {
    Response originalResponse = chain.proceed(chain.request());
    if (!originalResponse.headers(KEY_COOKIE).isEmpty()) {
        final StringBuffer cookieBuffer = new StringBuffer();
        Observable.from(originalResponse.headers(KEY_COOKIE))
                .map(new Func1<String, String>() {
                    @Override
                    public String call(String s) {
                        String[] cookieArray = s.split(";");
                        return cookieArray[0];
                    }
                })
                .subscribe(new Action1<String>() {
                    @Override
                    public void call(String cookie) {
                        cookieBuffer.append(cookie);
                    }
                });
        if (cookieBuffer.toString().contains("session_id")&&chain.request().url().uri().getPath().contains("/gongfu/v2/authenticate")){
            PreferencesHelper.setCookie(cookieBuffer.toString());
            Log.i("session_id",cookieBuffer.toString());
        }
    }
    return originalResponse;
}

private void getTrains() {
    progressBar.setVisibility(View.VISIBLE);
    HttpService service = RetrofitManager.getInstance().getService();
    SubscriptionUtil.unSubscribe(querySubscription);
    QueryParam queryParam = getQueryParam();
    querySubscription = service.queryTrains(leftTicketUrl, queryParam.getTrain_date(), queryParam.getFrom_station_code(), queryParam.getTo_station_code(), queryParam.getPurpose_codes())
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Action1<QueryTrainsResponse>() {
                @Override
                public void call(QueryTrainsResponse queryTrainsResponse) {
                    RxBus.getDefault().post("stop");
                    trains.clear();
                    List<QueryTrainsResponse.Data> datas = queryTrainsResponse.getData();
                    for (QueryTrainsResponse.Data data : datas) {
                        trains.add(data.getQueryLeftNewDTO().getStation_train_code());
                    }
                }
            });
}

public void runCode() {
    Observable.create(new Observable.OnSubscribe<Integer>() {

        @Override
        public void call(Subscriber<? super Integer> subscriber) {
            subscriber.onNext(1);
            subscriber.onCompleted();
        }
    }).subscribeOn(AndroidSchedulers.mainThread())  // 在doOnSubscribe()之前，不会影响上面的doOnSubscribe()
            .doOnSubscribe(new Action0() {
                @Override
                public void call() {
                    println(Thread.currentThread().getName()); // 在子线程中执行
                }
            }).subscribeOn(Schedulers.io()) // 会影响上面的doOnSubscribe()  .subscribe(new Observer<Integer>() {
            .subscribeOn(AndroidSchedulers.mainThread()) // 不会影响上面的doOnSubscribe()
            .subscribe(new Action1<Integer>() {
                @Override
                public void call(Integer integer) {
                    println(Thread.currentThread().getName());
                }
            });


}

public void runCode() {

        Integer[] arr = new Integer[]{1, 2, 3, 4, 5, 6, 7, 8};
        Observable.from(arr)
                .flatMap(new Func1<Integer, Observable<String>>() {
                    @Override
                    public Observable<String> call(final Integer integer) {
                        return Observable.just(String.valueOf(integer));
                    }

                })
                .subscribe(new Action1<String>() {
                    @Override
                    public void call(String s) {
                        println(s);
                    }
                });


    }

private void defaultIfEmpty() {
    Observable.create(new Observable.OnSubscribe<Integer>() {
        @Override
        public void call(Subscriber<? super Integer> subscriber) {
            subscriber.onCompleted();
        }
    }).defaultIfEmpty(3).subscribe(new Action1<Integer>() {
        @Override
        public void call(Integer integer) {
            Log.d(TAG, "defaultIfEmpty:"+integer);
        }
    });
}

@Override
protected void afterComponentsBound() {
    changePbColor(R.color.colorPrimary);
    getRecyclerView().setBackgroundColor(getContext().getResources().getColor(R.color.gray_light_translucent));

    onLoadMore = new Action1<List<Review>>() {
        @Override
        public void call(List<Review> reviews) {
            setIsLoadingMore(false);
            addItems(reviews);
            start += perPage;
            hideRefreshing();
        }
    };

    onRefresh = new Action1<List<Review>>() {
        @Override
        public void call(List<Review> reviews) {
            getAdapter().clear();
            addItems(reviews);
            start += perPage;
            hideRefreshing();
            enableLoadMore();
        }
    };

    onItemClick = new Action1<Review>() {
        @Override
        public void call(Review review) {
            Intent intent = DoubanMovieReviewActivity.intentFor(getContext());
            intent.putExtra(Constant.EXTRA_DOUBAN_MOVIE_REVIEW, review);
            getContext().startActivity(intent);
        }
    };

}

public static <T> AsyncSubject<T> create() {
    final SubjectSubscriptionManager<T> state = new SubjectSubscriptionManager();
    state.onTerminated = new Action1<SubjectObserver<T>>() {
        public void call(SubjectObserver<T> o) {
            Object v = state.getLatest();
            NotificationLite<T> nl = state.nl;
            o.accept(v, nl);
            if (v == null || !(nl.isCompleted(v) || nl.isError(v))) {
                o.onCompleted();
            }
        }
    };
    return new AsyncSubject(state, state);
}

public final Subscription connect() {
    final Subscription[] out = new Subscription[1];
    connect(new Action1<Subscription>() {
        public void call(Subscription t1) {
            out[0] = t1;
        }
    });
    return out[0];
}

@AfterViews
void afterViews(){
    setToolbar("강의녹음", 0, R.drawable.ic_arrow_back_white_24dp, 0, 0);

    df.setTimeZone(tz);

    mAudioRecorder = AudioRecorder.getInstance();
    mRxAudioPlayer = RxAudioPlayer.getInstance();
    mAudioRecorder.setOnErrorListener(this);

    boolean isPermissionsGranted = new RxPermissions(this)
            .isGranted(Manifest.permission.WRITE_EXTERNAL_STORAGE) &&
            new RxPermissions(this)
                    .isGranted(Manifest.permission.RECORD_AUDIO);
    if (!isPermissionsGranted) {
        new RxPermissions(this)
                .request(Manifest.permission.WRITE_EXTERNAL_STORAGE,
                        Manifest.permission.RECORD_AUDIO)
                .subscribe(new Action1<Boolean>() {
                    @Override
                    public void call(Boolean granted) {
                        // not record first time to request permission
                        if (granted) {
                            Toast.makeText(getApplicationContext(), "Permission granted",
                                    Toast.LENGTH_SHORT).show();
                        } else {
                            Toast.makeText(getApplicationContext(), "Permission not granted",
                                    Toast.LENGTH_SHORT).show();
                        }
                    }
                }, new Action1<Throwable>() {
                    @Override
                    public void call(Throwable throwable) {
                        throwable.printStackTrace();
                    }
                });
    } else {
        initRecorder();
    }
}

@Override
public Observable<StorageAccount> updateResourceAsync() {
    this.networkRulesHelper.setDefaultActionIfRequired();
    updateParameters.withTags(this.inner().getTags());
    return this.manager().inner().storageAccounts().updateAsync(
            resourceGroupName(), name(), updateParameters)
            .map(innerToFluentMap(this))
            .doOnNext(new Action1<StorageAccount>() {
                @Override
                public void call(StorageAccount storageAccount) {
                    clearWrapperProperties();
                }
            });
}

<T> Observable<CacheResult<T>> loadRemote(final ApiCache apiCache, final String key, Observable<T> source) {
    return source.map(new Func1<T, CacheResult<T>>() {
        @Override
        public CacheResult<T> call(T t) {
            ViseLog.i("loadRemote result=" + t);
            apiCache.put(key, t).subscribeOn(Schedulers.io()).subscribe(new Action1<Boolean>() {
                @Override
                public void call(Boolean status) {
                    ViseLog.i("save status => " + status);
                }
            });
            return new CacheResult<T>(false, t);
        }
    });
}