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
}
}
);