I want to request to a url using okhttp in another thread (like IO thread) and get Response in the Android main thread, But I don't know how to create an Observable.
4 Answers
It's easier and safer to use Observable.defer() instead of Observable.create():
final OkHttpClient client = new OkHttpClient();
Observable.defer(new Func0<Observable<Response>>() {
@Override public Observable<Response> call() {
try {
Response response = client.newCall(new Request.Builder().url("your url").build()).execute();
return Observable.just(response);
} catch (IOException e) {
return Observable.error(e);
}
}
});
That way unsubscription and backpressure are handled for you. Here's a great post by Dan Lew about create() and defer().
If you wished to go the Observable.create() route then it should look more like in this library with isUnsubscribed() calls sprinkled everywhere. And I believe this still doesn't handle backpressure.
Comments
I realise this post is a bit old, but there's a new and more convenient way of doing this now
Observable.fromCallable {
client.newCall(Request.Builder().url("your url").build()).execute()
}
More info: https://artemzin.com/blog/rxjava-defer-execution-of-function-via-fromcallable/
7 Comments
new is missing before Request, and after that it throws an errorI came late to the discussion but, if for some reason the code need to stream the response body, then defer or fromCallable won't do it. Instead one can employ the using operator.
Single.using(() -> okHttpClient.newCall(okRequest).execute(), // 1
response -> { // 2
...
return Single.just((Consumer<OutputStream>) fileOutput -> {
try (InputStream upstreamResponseStream = response.body().byteStream();
OutputStream fileOutput = responseBodyOutput) {
ByteStreams.copy(upstreamResponseStream, output);
}
});
},
Response::close, // 3
false) // 4
.subscribeOn(Schedulers.io()) // 5
.subscribe(copier -> copier.accept(...), // 6
throwable -> ...); // 7
- The first lambda executes the response after upon subscription.
- The second lambda creates the observable type, here with
Single.just(...) - The third lambda disposes the response. With
deferone could have used the try-with-resources style. - Set the
eagertoggle tofalseto make the disposer called after the terminal event, i.e. after the subscription consumer has been executed. - Of course make the thing happen on another threadpool
- Here's the lambda that will consume the response body. Without
eagerset tofalse, the code will raise an IOException with reason 'closed' because the response will be already closed before entering this lambda. - The
onErrorlambda should handle exceptions, especially theIOExceptionthat cannot be anymore caught with theusingoperator as it was possible with a try/catch withdefer.
Comments
Okhttp3 with RxSingle background API call.
Disposable disposables = Single.fromCallable(() -> {
Log.e(TAG, "clearData: Thread[" + Thread.currentThread().getName() + "]");
OkHttpClient client = Util.getHttpClient();
Request request = new Request.Builder()
.addHeader("Authorization", "Bearer " + Util.getUserToken())
.url(BuildConfig.BASE_URL + ApiConstants.DELETE_FEEDS)
.build();
Response response = client.newCall(request).execute();
if(response.isSuccessful()) {
...
return ; // Any type
} else {
return ; // Any type
}
}).subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe((result) -> {
Log.d(TAG, "api() completed");
});
compositeDisposable.add(disposables);
3 Comments
MaybeFromUnsafeCallable call which mimicks the MaybeFromCallable, but does not throw exception when the function threw an exception after the subscription was already closed. Ditto ObservableFromUnsafeCallable etc..