RxJava 소스 모음

 

 Subscription subscription =   Observable.create(new Observable.OnSubscribe<Response>() {
        OkHttpClient client = new OkHttpClient();
          @Override
          public void call(Subscriber<? super Response> subscriber) {
            try {
              Response response = client.newCall(new Request.Builder().url("your url").build()).execute();
              subscriber.onNext(response);
              subscriber.onCompleted();
              if (!response.isSuccessful()) subscriber.onError(new Exception("error"));
            } catch (IOException e) {
              subscriber.onError(e);
            }
          }
        })
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Subscriber<Response>() {
              @Override
              public void onCompleted() {

              }

              @Override
              public void onError(Throwable e) {

              }

              @Override
              public void onNext(Response response) {

              }
            });

class ServerPollingResponse {
    boolean isJobDone;

    @Override
    public String toString() {
        return "isJobDone=" + isJobDone;
    }
}

Subscription checkJobSubscription = mDataManager.pollServer(inputData)
        .repeatWhen(new Func1<Observable<? extends Void>, Observable<?>>() {
            @Override
            public Observable<?> call(Observable<? extends Void> observable) {
                Log.v(TAG, "repeatWhen, call");
                /**
                 * This is called only once.
                 * 5 means each repeated call will be delayed by 5 seconds
                 */
                return observable.delay(5, TimeUnit.SECONDS);
            }
        })
        .takeUntil(new Func1<ServerPollingResponse, Boolean>() {
            @Override
            public Boolean call(ServerPollingResponse response) {
                /** Here we can check if the responce is correct and if we should
                 *  finish polling
                 *  We finish polling when job is done.
                 *  In other words : "We stop taking when job is done"
                 */
                Log.v(TAG, "takeUntil, call response " + response);
                return response.isJobDone;
            }
        })
        .filter(new Func1<ServerPollingResponse, Boolean>() {
            @Override
            public Boolean call(ServerPollingResponse response) {
                /**
                 * We are filtering results if we return "false".
                 * Filtering means that onNext() will not be called.
                 * But onComplete() will be delivered.
                 */
                Log.v(TAG, "filter, call response " + response);
                return response.isJobDone;
            }
        })
        .subscribe(
                new Subscriber<ServerPollingResponse>() {
                    @Override
                    public void onCompleted() {
                        Log.v(TAG, "onCompleted ");
                    }

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

                    @Override
                    public void onNext(ServerPollingResponse response) {
                        Log.v(TAG, "onNext response " + response);
                        // Do whatever you need. Server polling has been finished
                    }
                }
        );