EDITED: with @Alrid’s comment tl;dr public abstract class Subscriber<T> implements Observer<T>, Subscription So a Subscriber is an implementation of the Observer, with additional semantics on subscription (it’s more about un-subscription). The code in your question just shows that it passes the Observer interface, instead of the implementation (usual programming practice). Also this code returns a … Read more
You can use the retryWhen() operator to add retry logic to any Observable. The following class contains the retry logic: RxJava 2.x public class RetryWithDelay implements Function<Observable<? extends Throwable>, Observable<?>> { private final int maxRetries; private final int retryDelayMillis; private int retryCount; public RetryWithDelay(final int maxRetries, final int retryDelayMillis) { this.maxRetries = maxRetries; this.retryDelayMillis = … Read more
SubscribeOn specify the Scheduler on which an Observable will operate. ObserveOn specify the Scheduler on which an observer will observe this Observable. So basically SubscribeOn is mostly subscribed (executed) on a background thread ( you do not want to block the UI thread while waiting for the observable) and also in ObserveOn you want to … Read more
You can use flatMap in case you have dynamic tasks composition. Something like this: public Observable<Boolean> whenAll(List<Observable<Boolean>> tasks) { return Observable.from(tasks) //execute in parallel .flatMap(task -> task.observeOn(Schedulers.computation())) //wait, until all task are executed //be aware, all your observable should emit onComplete event //otherwise you will wait forever .toList() //could implement more intelligent logic. eg. check … Read more
In Kotlin, the return@label syntax is used for specifying which function among several nested ones this statement returns from. It works with function literals (lambdas) and local functions. Non-labeled return statements return from the nearest (i.e. innermost) enclosing fun (ignoring lambdas). Consider this function: fun foo(ints: List<Int>) { ints.forEach { if (it == 0) return … Read more
You need to tell Retrofit that you want to use RxJava 2, using: addCallAdapterFactory(RxJava2CallAdapterFactory.create()) So, for creating your Retrofit object, you will have something like: Retrofit retrofit = new Retrofit.Builder() .baseUrl(SERVICE_ENDPOINT) .addCallAdapterFactory(RxJava2CallAdapterFactory.create()) .build();
Instead of declaring the API call like you did: Observable<MyResponseObject> apiCall(@Body body); You can also declare it like this: Observable<Response<MyResponseObject>> apiCall(@Body body); You will then have a Subscriber like the following: new Subscriber<Response<StartupResponse>>() { @Override public void onCompleted() {} @Override public void onError(Throwable e) { Timber.e(e, “onError: %”, e.toString()); // network errors, e. g. UnknownHostException, … Read more
Observable is the generic ReactiveX building block, of event source that emits values over time. (and thus exists in every language ReactiveX extended to) in short Observable events are: onNext* (onCompleted | onError)? /(* zero or more ? – zero or 1) Single and Completable are new types introduced exclusively at RxJava that represent reduced … Read more
What backpressure manifests in practice is bounded buffers, Flowable.observeOn has a buffer of 128 elements that gets drained as fast as the dowstream can take it. You can increase this buffer size individually to handle bursty source and all the backpressure-management practices still apply from 1.x. Observable.observeOn has an unbounded buffer that keeps collecting the … Read more
Short answer All sequence/stream processing libs are offering very similar API for pipeline building. The differences are in API for handling multi-threading and composition of pipelines. Long answer RxJava is quite different from Stream. Of all JDK things, the closest to rx.Observable is perhaps java.util.stream.Collector Stream + CompletableFuture combo (which comes at a cost of … Read more