For example you can use Observable.fromCallable to create observable with your data. public Observable<Data> getData(){ return Observable.fromCallable(() -> { Data result = null; //do something, get your Data object return result; }); } then use your data getData().subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribe(data -> { //do something with your data }, error -> { //do something on error … Read more
Easy as pie, just use filter An example in scala import rx.lang.scala.Observable val o: Observable[String] = Observable.just(“a”, “b”, “c”, “a”, “b”, “b”, “b”, “a”) val hotO: Observable[String] = o.share val aSource: Observable[String] = hotO.filter(x ⇒ x == “a”) val bSource: Observable[String] = hotO.filter(x ⇒ x == “b”) val cSource: Observable[String] = hotO.filter(x ⇒ x == … Read more
Be sure to add implementation ‘com.squareup.retrofit2:adapter-rxjava2:2.4.0’ or whatever version you are using to your dependencies, and then configure retrofit with that converter: Retrofit retrofit = new Retrofit.Builder() .baseUrl(endpoint) .addConverterFactory(GsonConverterFactory.create()) .addCallAdapterFactory(RxJava2CallAdapterFactory.create()) .build(); Updated RxJavaCallAdapterFactory was renamed to RxJava2CallAdapterFactory. Changed the snipped above.
According to Mathias Doenitz at this point in time RxJava doesn’t have back pressure unlike Akkas Reactive Streams implementation. But RxJava seems to be working on adding back pressure. Both frameworks will be able to interact through the reactive streaming spi. So you will be able to do very very similar things. According to Mathias … Read more
The following is what I see as benefits of using reactive event streams in a single-threaded synchronous application. 1. More declarative, less side-effects and less mutable state. Event streams are capable of encapsulating logic and state, potentially leaving your code without side-effects and mutable variables. Consider an application that counts button clicks and displays the … Read more
Flow is a cold asynchronous stream, just like an Observable. All transformations on the flow, such as map and filter do not trigger flow collection or execution, only terminal operators (e.g. single) do trigger it. The onEach method is just a transformation. Therefore you should replace it with the terminal flow operator collect. Also you … Read more
I think another answer is outdated. You should probably check the following methods. singleOrError: Emits the one and only element, IndexOutOfBoundsException if the source is longer than 1 item or a NoSuchElementException if the source is empty. firstOrError: Emits the first element or a NoSuchElementException if the source is empty. lastOrError: Emits the lastelement or … Read more
A Subject is an Observable and an Observer at the same time, it can be unsubscribed from just like normal observables. What makes subject special is that it is sort of bridge between observables and observers. It can pass through the items it observes by reemitting them, and it can also emit new items. Subjects … Read more
You would do something like this: Observable<View> clickEventObservable = Observable.create(new Observable.OnSubscribe<View>() { @Override public void call(final Subscriber<? super View> subscriber) { viewIWantToMonitorForClickEvents.setOnClickListener(new View.OnClickListener() { @Override public void onClick(View v) { if (subscriber.isUnsubscribed()) return; subscriber.onNext(v); } }); } }); // You can then apply all sorts of operation here Subscription subscription = clickEventObservable.flatMap(/* */); // Unsubscribe … Read more
.doOnError() is an operator, and is not as such a part of the Subscriber. Therefore, having a .doOnError() does not count as an implemented onError(). About the question in one of the comments, of course it is possible to use lambdas. In this case simply replace .doOnError(throwable -> L.e(TAG, “Throwable ” + throwable.getMessage())) .subscribe(s -> … Read more