import io.reactivex.subscribers.DefaultSubscriber; //导入依赖的package包/类
@OnClick(R.id.btn_demo_timing_1)
public void btn1_RunSingleTaskAfter2s() {
  _log(String.format("A1 [%s] --- BTN click", _getCurrentTimestamp()));

  Flowable.timer(2, TimeUnit.SECONDS) //
      .subscribe(
          new DefaultSubscriber<Long>() {
            @Override
            public void onNext(Long number) {
              _log(String.format("A1 [%s]     NEXT", _getCurrentTimestamp()));
            }

            @Override
            public void onError(Throwable e) {
              Timber.e(e, "something went wrong in TimingDemoFragment example");
            }

            @Override
            public void onComplete() {
              _log(String.format("A1 [%s] XXX COMPLETE", _getCurrentTimestamp()));
            }
          });
}
 

import io.reactivex.subscribers.DefaultSubscriber; //导入依赖的package包/类
@OnClick(R.id.btn_demo_timing_4)
public void btn4_RunTask5Times_IntervalOf3s() {
  _log(String.format("D4 [%s] --- BTN click", _getCurrentTimestamp()));

  Flowable.interval(3, TimeUnit.SECONDS)
      .take(5)
      .subscribe(
          new DefaultSubscriber<Long>() {
            @Override
            public void onNext(Long number) {
              _log(String.format("D4 [%s]     NEXT", _getCurrentTimestamp()));
            }

            @Override
            public void onError(Throwable e) {
              Timber.e(e, "something went wrong in TimingDemoFragment example");
            }

            @Override
            public void onComplete() {
              _log(String.format("D4 [%s] XXX COMPLETE", _getCurrentTimestamp()));
            }
          });
}
 

import io.reactivex.subscribers.DefaultSubscriber; //导入依赖的package包/类
public static void main(String[] args) {
	// TODO Auto-generated method stub
	Flowable< Long> flowable=Flowable.rangeLong(2, 12);
	flowable.subscribe(new DefaultSubscriber<Long>() {

		@Override
		public void onComplete() {
			// TODO Auto-generated method stub
			System.out.println("Its Done!!!");

		}

		@Override
		public void onError(Throwable throwable) {
			// TODO Auto-generated method stub
			throwable.printStackTrace();

		}

		@Override
		public void onNext(Long value) {
			// TODO Auto-generated method stub
			if(value==4)
				cancel();
			System.out.println("value:-" + value);

		}

		@Override
		protected void onStart() {
			// TODO Auto-generated method stub
			request(5);
		}

	});

}
 

import io.reactivex.subscribers.DefaultSubscriber; //导入依赖的package包/类
@Override
    public void getDanmuDetail(String roomid, String live_type) {
        if (live_type.equals("panda")) {
            RetrofitHelper.getPandaHelper().create(LiveAPI.class)
                    .getPandaChatroom(roomid)
                    .subscribeOn(Schedulers.io())
                    .observeOn(AndroidSchedulers.mainThread())
                    .subscribe(new DefaultSubscriber<LivePandaBean>() {
                        @Override
                        public void onComplete() {

                        }

                        @Override
                        public void onError(Throwable e) {
//                            view.showError("弹幕服务器接口已过期，请刷新直播列表！");
                            view.showError(e.getMessage());
                        }

                        @Override
                        public void onNext(LivePandaBean detailPandaBean) {
                            if (detailPandaBean.getErrno() == 0) {
                                view.updateChatDetail(detailPandaBean);
                            } else {
                                view.showError(detailPandaBean.getErrmsg());
                            }
                        }
                    });
        } else {
            view.showError("直播平台：" + live_type + "！不是熊猫TV的弹幕池！");
        }
    }
 

import io.reactivex.subscribers.DefaultSubscriber; //导入依赖的package包/类
@OnClick(R.id.btn_demo_timing_5)
public void btn5_RunTask5Times_IntervalOf3s() {
  _log(String.format("D5 [%s] --- BTN click", _getCurrentTimestamp()));

  Flowable.just("Do task A right away")
      .doOnNext(input -> _log(String.format("D5 %s [%s]", input, _getCurrentTimestamp())))
      .delay(1, TimeUnit.SECONDS)
      .doOnNext(
          oldInput ->
              _log(
                  String.format(
                      "D5 %s [%s]", "Doing Task B after a delay", _getCurrentTimestamp())))
      .subscribe(
          new DefaultSubscriber<String>() {
            @Override
            public void onComplete() {
              _log(String.format("D5 [%s] XXX COMPLETE", _getCurrentTimestamp()));
            }

            @Override
            public void onError(Throwable e) {
              Timber.e(e, "something went wrong in TimingDemoFragment example");
            }

            @Override
            public void onNext(String number) {
              _log(String.format("D5 [%s]     NEXT", _getCurrentTimestamp()));
            }
          });
}