While using RxJava and Retrofit 2 I am trying to create Unit Tests to cover when my app receives specific responses.

The issue I have is that with Retrofit 2 I cannot see a nice way of creating a retrofit.Response object without the use of reflection.

    @Test
    public void testLogin_throwsLoginBadRequestExceptionWhen403Error() {


        Request.Builder requestBuilder = new Request.Builder();
        requestBuilder.get();
        requestBuilder.url(&quot;http://localhost&quot;);

        Response.Builder responseBuilder = new Response.Builder();
        responseBuilder.code(403);
        responseBuilder.protocol(Protocol.HTTP_1_1);
        responseBuilder.body(ResponseBody.create(MediaType.parse(&quot;application/json&quot;), &quot;{\&quot;key\&quot;:[\&quot;somestuff\&quot;]}&quot;));
        responseBuilder.request(requestBuilder.build());

        retrofit.Response&lt;LoginResponse&gt; aResponse = null;

        try {
            Constructor&lt;retrofit.Response&gt; constructor= (Constructor&lt;retrofit.Response&gt;) retrofit.Response.class.getDeclaredConstructors()[0];
            constructor.setAccessible(true);
            aResponse = constructor.newInstance(responseBuilder.build(), null, null);
        } catch (Exception ex) {
            //reflection error
        }

        doReturn(Observable.just(aResponse)).when(mockLoginAPI).login(anyObject());

        TestSubscriber testSubscriber = new TestSubscriber();
        loginAPIService.login(loginRequest).subscribe(testSubscriber);

        Throwable resultError = (Throwable) testSubscriber.getOnErrorEvents().get(0);
        assertTrue(resultError instanceof LoginBadRequestException);
    }

I have tried using the following but this creates an OkHttp Response vs a Retrofit Response.

        Request.Builder requestBuilder = new Request.Builder();
        requestBuilder.get();
        requestBuilder.url(&quot;http://localhost&quot;);

        Response.Builder responseBuilder = new Response.Builder();
        responseBuilder.code(403);
        responseBuilder.protocol(Protocol.HTTP_1_1);

How to create a retrofit.Response object during Unit Testing with Retrofit 2

I am confused about the difference between EventBus and RxJava in android. I need to implement one of them for my issue about notifying some components when some changes have been done, so that they can update their state.  
Also, I read that EventsBus has became deprecated over RxJava and I don&#39;t know if this information is true or not.

What is the difference between EventBus and RxJava?

I was checking the documentation of RXJava and I notice that concat and merge operators seems do the same.
I wrote a couple test to be sure.


    @Test
    public void testContact() {

        Observable.concat(Observable.just(&quot;Hello&quot;),
                          Observable.just(&quot;reactive&quot;),
                          Observable.just(&quot;world&quot;))
                  .subscribe(System.out::println);
    }

    @Test
    public void testMerge() {

        Observable.merge(Observable.just(&quot;Hello&quot;),
                          Observable.just(&quot;reactive&quot;),
                          Observable.just(&quot;world&quot;))
                .subscribe(System.out::println);
    }


The documentation says 

&gt; The Merge operator is also similar. It combines the emissions of two or more Observables, but may interleave them, whereas Concat never interleaves the emissions from multiple Observables.

But still I don&#39;t full understand, running this test thousand of times the merge result always is the same. Since the order is not granted I was expecting sometimes &quot;reactive&quot; &quot;world&quot; &quot;hello&quot; for example.

The code is here https://github.com/politrons/reactive

Concat VS Merge operator

I need some help in implementing parallel asynchronous calls in RxJava. I have picked up a simple use case wherein the FIRST call fetches (rather searches) a list of products (Tile) to be displayed. The subsequent calls go out and fetch (A) REVIEWS and (B) PRODUCT IMAGES

After several attempts I got to this place. 

     1    Observable&lt;Tile&gt; searchTile = searchServiceClient.getSearchResults(searchTerm);
     2    List&lt;Tile&gt; allTiles = new ArrayList&lt;Tile&gt;();
     3    ClientResponse response = new ClientResponse();
    
     4    searchTile.parallel(oTile -&gt; {
     5      return oTile.flatMap(t -&gt; {
     6        Observable&lt;Reviews&gt; reviews = reviewsServiceClient.getSellerReviews(t.getSellerId());
     7        Observable&lt;String&gt; imageUrl = reviewsServiceClient.getProductImage(t.getProductId());
    
     8        return Observable.zip(reviews, imageUrl, (r, u) -&gt; {
     9          t.setReviews(r);
    10          t.setImageUrl(u);
    
    11          return t;
    12        });
    
    13      });
    14    }).subscribe(e -&gt; {
    15      allTiles.add((Tile) e);
    16    });

Line  1: goes out and fetches the product (Tile) to be displayed

Line  4: We take the list of the Observable&lt;Tile&gt; and SHARD it to fetch reviews and imageUrls

Lie 6,7: Fetch the Observable&lt;Review&gt; review and Observable&lt;String&gt; url

Line  8: Finally the 2 observables are zipped up to return an updated Observable&lt;Tile&gt;

Line 15: finally line 15 collates all the individual products to be displayed in a collection which can be returned back to the calling layer

While the Observable&lt;Tile&gt; has been sharded and in our tests run over 4 different threads; fetching of reviews and images seems to be one after another. I suspect that the zip step on line 8 is basically causing the sequential invocation of the the 2 observables (reviews and url).

![enter image description here][1]

Does this group have any suggestion to parallely fetch reiews and image urls. In essence the waterfall chart attached above should look more vertically stacked. The calls to reviews and images should be in parallel

thanks
anand raman


  [1]: http://i.stack.imgur.com/YVERX.png

RxJava Fetching Observables In Parallel

I would like to know how to ignore exceptions and continue infinite stream (in my case stream of locations)?

I&#39;m fetching current user position (using [Android-ReactiveLocation](https://github.com/mcharmas/Android-ReactiveLocation)) and then sending them to my API (using [Retrofit](http://square.github.io/retrofit/)). 

In my case, when exception occurs during network call (e.g. timeout) `onError` method is invoked and stream stops itself. How to avoid it?

Activity:

    private RestService mRestService;
    private Subscription mSubscription;
    private LocationRequest mLocationRequest = LocationRequest.create()
                .setPriority(LocationRequest.PRIORITY_HIGH_ACCURACY)
                .setInterval(100);
    ...
    private void start() {
        mRestService = ...;
        ReactiveLocationProvider reactiveLocationProvider = new ReactiveLocationProvider(this);
        mSubscription = reactiveLocationProvider.getUpdatedLocation(mLocationRequest)
                .buffer(50)
                .flatMap(locations -&gt; mRestService.postLocations(locations)) // can throw exception
                .subscribeOn(Schedulers.newThread())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe();
    }

RestService:

    public interface RestService {
        @POST(&quot;/.../&quot;)
        Observable&lt;Response&gt; postLocations(@Body List&lt;Location&gt; locations);
    }

   


How to ignore error and continue infinite stream?

In which cases I should use doOnNext, and in which cases
doOnEach?

     .doOnEach(new Action1&lt;Notification&lt;? super MessageProfile&gt;&gt;() {
                    @Override
                    public void call(Notification&lt;? super MessageProfile&gt; notification) {

                    }
                })
     .doOnNext(new Action1&lt;MessageProfile&gt;() {
                    @Override
                    public void call(MessageProfile profile) {
                        messageProfileDao.save(profile);
                    }
                })

This looks like the both operators have the same effect.

RxJava: difference between doOnNext and doOnEach

When to call the subscribeWith method rather than plain subscribe? And what is the use case?

    compositeDisposable.add(get()
        .observeOn(AndroidSchedulers.mainThread())
        .subscribeOn(Schedulers.io())
        .subscribe(this::handleResponse, this::handleError));

VS

       compositeDisposable.add(get()
                    .observeOn(AndroidSchedulers.mainThread())
                    .subscribeOn(Schedulers.io())
                  //  .subscribe(this::handleResponse, this::handleError);
                    .subscribeWith(new DisposableObserver&lt;News&gt;() {
                        @Override public void onNext(News value) {
                            handleResponse(value);
                        }
    
                        @Override public void onError(Throwable e) {
                            handleError(e);
                        }
    
                        @Override public void onComplete() {
                           // dispose here ? why? when the whole thing will get disposed later
                           //via  compositeDisposable.dispose();  in onDestroy();
                        }
                    }));


Thank you  


----------
**Added Later**
----------
According to the documentation, both return disposable SingleObserver instances:

    @CheckReturnValue
    @SchedulerSupport(SchedulerSupport.NONE)
    public final &lt;E extends SingleObserver&lt;? super T&gt;&gt; E subscribeWith(E observer) {
        subscribe(observer);
        return observer;
    }

    @SchedulerSupport(SchedulerSupport.NONE)
    public final Disposable subscribe(final Consumer&lt;? super T&gt; onSuccess, final Consumer&lt;? super Throwable&gt; onError) {
        ObjectHelper.requireNonNull(onSuccess, &quot;onSuccess is null&quot;);
        ObjectHelper.requireNonNull(onError, &quot;onError is null&quot;);
        ConsumerSingleObserver&lt;T&gt; s = new ConsumerSingleObserver&lt;T&gt;(onSuccess, onError);
        subscribe(s);
        return s;
    }

 Where ConsumerSingleObserver class implements SingleObserver and Disposable.



Subscribewith Vs subscribe in RxJava2(Android)?

What is the benefit of using Retrofit in combination with Rxjava?

Why is RxJava often used with Retrofit?

I am an Android Student. I want to learn RxJava. My Question is &quot;What is  CompositeDisposable in RxJava?&quot;. Please describe in detail.

What is CompositeDisposable in RxJava

I&#39;m working on networking for my app. So I decided to try out Square&#39;s [Retrofit][1]. I see that they support simple `Callback`

    @GET(&quot;/user/{id}/photo&quot;)
    void getUserPhoto(@Path(&quot;id&quot;) int id, Callback&lt;Photo&gt; cb);

and RxJava&#39;s `Observable`

    @GET(&quot;/user/{id}/photo&quot;)
    Observable&lt;Photo&gt; getUserPhoto(@Path(&quot;id&quot;) int id);

Both look pretty similar at first glance, but when it gets to implementation it gets interesting...

While with simple callback implementation would look similar to this:


    api.getUserPhoto(photoId, new Callback&lt;Photo&gt;() {
        @Override
        public void onSuccess() {
        }
    });

which is quite simple and straightforward. And with `Observable` it quickly gets verbose and quite complicated.

    public Observable&lt;Photo&gt; getUserPhoto(final int photoId) {
        return Observable.create(new Observable.OnSubscribeFunc&lt;Photo&gt;() {
            @Override
            public Subscription onSubscribe(Observer&lt;? super Photo&gt; observer) {
                try {
                    observer.onNext(api.getUserPhoto(photoId));
                    observer.onCompleted();
                } catch (Exception e) {
                    observer.onError(e);
                }

                return Subscriptions.empty();
            }
        }).subscribeOn(Schedulers.threadPoolForIO());
    }

And that is not it. You still have to do something like this:

    Observable.from(photoIdArray)
            .mapMany(new Func1&lt;String, Observable&lt;Photo&gt;&gt;() {
                @Override
                public Observable&lt;Photo&gt; call(Integer s) {
                    return getUserPhoto(s);
                }
            })
            .subscribeOn(Schedulers.threadPoolForIO())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Action1&lt;Photo&gt;() {
                @Override
                public void call(Photo photo) {
                    //save photo?
                }
            });

Am I missing something here? Or is this a wrong case to use `Observable`s?
When would/should one prefer `Observable` over simple Callback?

## Update

Using retrofit is much simpler than example above as @Niels showed in his answer or in Jake Wharton&#39;s example project [U2020][2]. But essentially the question stays the same - when should one use one way or the other?


  [1]: https://github.com/square/retrofit
  [2]: https://github.com/JakeWharton/u2020

When should one use RxJava Observable and when simple Callback on Android?

In RxJava there are [5 different schedulers][1] to choose from:

&gt; 1. **immediate()**: Creates and returns a Scheduler that executes work immediately on the current thread.
&gt;
&gt; 2. **trampoline()**: Creates and returns a Scheduler that queues work on the current thread to be executed after the current work completes.
&gt;
&gt; 3. **newThread()**: Creates and returns a Scheduler that creates a new Thread for each unit of work.
&gt;
&gt; 4. **computation()**: Creates and returns a Scheduler intended for computational work. This can be used for event-loops, processing callbacks and other computational work. Do not perform IO-bound work on this scheduler. Use Schedulers.**io()** instead.
&gt;
&gt; 5. **io()**: Creates and returns a Scheduler intended for IO-bound work.
The implementation is backed by an Executor thread-pool that will grow as needed. This can be used for asynchronously performing blocking IO. Do not perform computational work on this scheduler. Use Schedulers.**computation()** instead.

## Questions: ##
The first 3 schedulers are pretty self explanatory; however, I&#39;m a little confused about **computation** and **io**.

 1. What exactly is &quot;IO-bound work&quot;? Is it used for dealing with streams (`java.io`) and files (`java.nio.files`)? Is it used for database queries? Is it used for downloading files or accessing REST APIs?
 2. How is **computation()** different from **newThread()**? Is it that all **computation()** calls are on a single (background) thread instead of a new (background) thread each time?
 3. Why is it bad to call **computation()** when doing IO work?
 4. Why is it bad to call **io()** when doing computational work?


  [1]: http://reactivex.io/RxJava/javadoc/rx/schedulers/Schedulers.html

Use cases for RxJava schedulers

I would like to know the difference between 
`CompletableFuture`,`Future` and `Observable` `RxJava`.

What I know is all are asynchronous but 

`Future.get()` blocks the thread 

`CompletableFuture` gives the callback methods

`RxJava Observable` --- similar to `CompletableFuture` with other benefits(not sure)

For example: if client needs to make multiple service calls and when we use `Futures` (Java) `Future.get()` will be executed sequentially...would like to know how its better in RxJava..

And the documentation http://reactivex.io/intro.html says 

**It is difficult to use Futures to optimally compose conditional asynchronous execution flows (or impossible, since latencies of each request vary at runtime). This can be done, of course, but it quickly becomes complicated (and thus error-prone) or it prematurely blocks on Future.get(), which eliminates the benefit of asynchronous execution.**

Really interested to know how `RxJava` solves this problem. I found it difficult to understand from the documentation.

Difference between CompletableFuture, Future and RxJava&#39;s Observable

When do you use `map` vs `flatMap` in **RxJava**?

Say, for example, we want to map Files containing JSON into Strings that contain the JSON--

Using `map`, we have to deal with the `Exception` somehow. But how?:
    
    Observable.from(jsonFile).map(new Func1&lt;File, String&gt;() {
        @Override public String call(File file) {
            try {
                return new Gson().toJson(new FileReader(file), Object.class);
            } catch (FileNotFoundException e) {
                // So Exception. What to do ?
            }
            return null; // Not good :(
        }
    });

Using `flatMap`, it&#39;s much more verbose, but we can forward the problem down the chain of `Observables` and handle the error if we choose somewhere else and even retry:
    
    Observable.from(jsonFile).flatMap(new Func1&lt;File, Observable&lt;String&gt;&gt;() {
        @Override public Observable&lt;String&gt; call(final File file) {
            return Observable.create(new Observable.OnSubscribe&lt;String&gt;() {
                @Override public void call(Subscriber&lt;? super String&gt; subscriber) {
                    try {
                        String json = new Gson().toJson(new FileReader(file), Object.class);
    
                        subscriber.onNext(json);
                        subscriber.onCompleted();
                    } catch (FileNotFoundException e) {
                        subscriber.onError(e);
                    }
                }
            });
        }
    });

I like the simplicity of the `map`, but the error handling of `flatmap` (not the verbosity). I haven&#39;t seen any best practices on this floating around and I&#39;m curious how this is being used in practice.

When do you use map vs flatMap in RxJava?

The [rxjava doc][1] definition of switchmap is rather vague and it links to the [same page][2] as flatmap. What is the difference between the two operators?


  [1]: https://github.com/ReactiveX/RxJava/wiki/Transforming-Observables
  [2]: http://reactivex.io/documentation/operators/flatmap.html

What is the difference between flatmap and switchmap in RxJava?

Are Java 8 streams similar to RxJava observables?

Java 8 stream definition:

&gt; Classes in the new `java.util.stream` package provide a Stream API to
&gt; support functional-style operations on streams of elements.




Difference between Java 8 streams and RxJava observables

I have been looking at new rx java 2 and I&#39;m not quite sure I understand the idea of `backpressure` anymore... 

I&#39;m aware that we have `Observable` that does not have `backpressure` support and `Flowable` that has it.

So based on example, lets say I have `flowable` with `interval`:

            Flowable.interval(1, TimeUnit.MILLISECONDS, Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer&lt;Long&gt;() {
                    @Override
                    public void accept(Long aLong) throws Exception {
                        // do smth
                    }
                });

This is going to crash after around 128 values, and thats pretty obvious I am consuming slower than getting items. 

But then we have the same with `Observable` 

         Observable.interval(1, TimeUnit.MILLISECONDS, Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer&lt;Long&gt;() {
                    @Override
                    public void accept(Long aLong) throws Exception {
                        // do smth
                    }
                });

This will not crash at all, even when I put some delay on consuming it still works. To make `Flowable` work lets say I put `onBackpressureDrop` operator, crash is gone but not all values are emitted either.

So the base question I can not find answer currently in my head is why should I care about `backpressure` when I can use plain `Observable` still receive all values without managing the `buffer`? Or maybe from the other side, what advantages do `backpressure` give me in favour of managing and handling the consuming?

Observable vs Flowable rxJava2

Can anyone please explain the difference between Observable, Completable and Single in RxJava with clear examples?

In which scenario we use one over the others?

What is the difference between Observable, Completable and Single in RxJava

I&#39;m trying to upgrade to Retrofit 2.0 and add RxJava in my android project. I&#39;m making an api call and want to retrieve the error code in case of an error response from the server. 

    Observable&lt;MyResponseObject&gt; apiCall(@Body body);

And in the RxJava call:

    myRetrofitObject.apiCall(body).subscribe(new Subscriber&lt;MyResponseObject&gt;() {
            @Override
            public void onCompleted() {

            }

            @Override
            public void onError(Throwable e) {
              
            }

            @Override
            public void onNext(MyResponseObject myResponseObject) {
               //On response from server
            }
        });
    

In Retrofit 1.9, the RetrofitError still existed and we could get the status by doing:

    error.getResponse().getStatus()

How do you do this with Retrofit 2.0 using RxJava?


Get response status code using Retrofit 2.0 and RxJava

I&#39;m going to send a simple get method to my server(it is Rails app) and get the result using RxJava and Retrofit. The thing that I did is:

My interface:

&lt;!-- language: java --&gt;

    public interface ApiCall {
        String SERVICE_ENDPOINT = &quot;https://198.50.214.15&quot;;
        @GET(&quot;/api/post&quot;)
        io.reactivex.Observable&lt;Post&gt; getPost();
    }

My model is this:

&lt;!-- language: java --&gt;

    public class Post
    {
        @SerializedName(&quot;id&quot;)
        private String id;
        @SerializedName(&quot;body&quot;)
        private String body;
        @SerializedName(&quot;title&quot;)
        private String title;
    
        public String getId ()
        {
            return id;
        }
    
    
        public String getBody ()
        {
            return body;
        }
    
    
        public String getTitle ()
        {
            return title;
        }
    
    }

And this is what I did in my activity:

&lt;!-- language: java --&gt;

    public class Javax extends AppCompatActivity {
        RecyclerView rvListContainer;
        postAdapter postAdapter;
        List&lt;String&gt; messageList=new ArrayList&lt;&gt;();
        @Override
        protected void onCreate(Bundle savedInstanceState) {
            super.onCreate(savedInstanceState);
            setContentView(R.layout.activity_javax);
    
            rvListContainer=(RecyclerView)findViewById(R.id.recyclerView);
            postAdapter=new postAdapter(messageList);
    
            rvListContainer.setAdapter(postAdapter);
        }
        @Override
        protected void onResume() {
            super.onResume();
            Retrofit retrofit=new Retrofit.Builder()
                    .addCallAdapterFactory(RxJavaCallAdapterFactory.create())
                    .addConverterFactory(GsonConverterFactory.create())
                    .baseUrl(&quot;https://198.50.214.15&quot;)
                    .build();
            ApiCall apiService=retrofit.create(ApiCall.class);
    
            Observable&lt;Post&gt; observable=apiService.getPost();
    
            observable.subscribeOn(Schedulers.newThread())
                    .observeOn(AndroidSchedulers.mainThread())
                    .subscribe(responseData -&gt; {
                        messageList.add(responseData.getTitle());
                        postAdapter.notifyDataSetChanged();
                    });
    
        }
    }

I don&#39;t know why I get this error which says I have problem with adapter. I also included the adapter into the gradle:

    E/AndroidRuntime: FATAL EXCEPTION: main
	Process: com.example.hussein.sqlitedatabase, PID: 19445
	java.lang.RuntimeException: Unable to resume activity {com.example.hussein.sqlitedatabase/com.example.hussein.sqlitedatabase.Javax}: java.lang.IllegalArgumentException: Unable to create call adapter for io.reactivex.Observable&lt;com.example.hussein.sqlitedatabase.Post&gt;
		for method ApiCall.getPost
		at android.app.ActivityThread.performResumeActivity(ActivityThread.java:2964)
		at android.app.ActivityThread.handleResumeActivity(ActivityThread.java:2993)
		at android.app.ActivityThread.handleLaunchActivity(ActivityThread.java:2395)
		at android.app.ActivityThread.access$800(ActivityThread.java:151)
		at android.app.ActivityThread$H.handleMessage(ActivityThread.java:1321)
		at android.os.Handler.dispatchMessage(Handler.java:110)
		at android.os.Looper.loop(Looper.java:193)
		at android.app.ActivityThread.main(ActivityThread.java:5292)
		at java.lang.reflect.Method.invokeNative(Native Method)
		at java.lang.reflect.Method.invoke(Method.java:515)
		at com.android.internal.os.ZygoteInit$MethodAndArgsCaller.run(ZygoteInit.java:824)
		at com.android.internal.os.ZygoteInit.main(ZygoteInit.java:640)
		at dalvik.system.NativeStart.main(Native Method)
	 Caused by: java.lang.IllegalArgumentException: Unable to create call adapter for io.reactivex.Observable&lt;com.example.hussein.sqlitedatabase.Post&gt;
		for method ApiCall.getPost
		at retrofit2.ServiceMethod$Builder.methodError(ServiceMethod.java:751)
		at retrofit2.ServiceMethod$Builder.createCallAdapter(ServiceMethod.java:236)
		at retrofit2.ServiceMethod$Builder.build(ServiceMethod.java:161)
		at retrofit2.Retrofit.loadServiceMethod(Retrofit.java:169)
		at retrofit2.Retrofit$1.invoke(Retrofit.java:146)
		at $Proxy0.getPost(Native Method)
		at com.example.hussein.sqlitedatabase.Javax.onResume(Javax.java:42)
		at android.app.Instrumentation.callActivityOnResume(Instrumentation.java:1197)
		at android.app.Activity.performResume(Activity.java:5343)
		at android.app.ActivityThread.performResumeActivity(ActivityThread.java:2950)
			... 12 more
	 Caused by: java.lang.IllegalArgumentException: Could not locate call adapter for io.reactivex.Observable&lt;com.example.hussein.sqlitedatabase.Post&gt;.
	  Tried:
	   * retrofit2.adapter.rxjava.RxJavaCallAdapterFactory
	   * retrofit2.ExecutorCallAdapterFactory
		at retrofit2.Retrofit.nextCallAdapter(Retrofit.java:240)
		at retrofit2.Retrofit.callAdapter(Retrofit.java:204)
		at retrofit2.ServiceMethod$Builder.createCallAdapter(ServiceMethod.java:234)
			... 20 more

This is my Gradle dependency:

    dependencies {
        compile fileTree(dir: &#39;libs&#39;, include: [&#39;*.jar&#39;])
        androidTestCompile(&#39;com.android.support.test.espresso:espresso-core:2.2.2&#39;, {
            exclude group: &#39;com.android.support&#39;, module: &#39;support-annotations&#39;
        })
        compile &#39;com.android.support.constraint:constraint-layout:1.0.2&#39;
        compile &#39;com.android.support:appcompat-v7:25.3.1&#39;
        testCompile &#39;junit:junit:4.12&#39;
    
        compile &#39;com.squareup.retrofit2:retrofit:2.2.0&#39;
    
        compile &#39;com.android.support:recyclerview-v7:25.3.1&#39;
        compile &#39;com.android.support:design:25.3.1&#39;
    
        compile &#39;io.reactivex.rxjava2:rxandroid:2.0.1&#39;
        compile &#39;io.reactivex.rxjava2:rxjava:2.0.1&#39;
    
        compile &#39;com.jakewharton.retrofit:retrofit2-rxjava2-adapter:1.0.0&#39;
        compile &#39;com.squareup.retrofit2:adapter-rxjava:2.0.2&#39;
    
        compile &#39;com.squareup.retrofit2:converter-gson:2.0.0&#39;
    
        compile &#39;com.google.code.gson:gson:2.4&#39;
    
    }



Unable to create call adapter for io.reactivex.Observable

**TL;DR**
How to convert `Task.whenAll(List&lt;Task&gt;)` into `RxJava`?

My existing code uses Bolts to build up a list of asynchronous tasks and waits until all of those tasks finish before performing other steps. Essentially, it builds up a `List&lt;Task&gt;` and returns a single `Task` which is marked as completed when *all* tasks in the list complete, as per the [example on the Bolts site][1].

I&#39;m looking to replace `Bolts` with `RxJava` and I&#39;m assuming this method of building up a list of async tasks (size not known in advance) and wrapping them all into a single `Observable` is possible, but I don&#39;t know how. 

I&#39;ve tried looking at `merge`, `zip`, `concat` etc... but can&#39;t get to work on the `List&lt;Observable&gt;` that I&#39;d be building up as they all seem geared to working on just two `Observables` at a time if I understand the docs correctly.

I&#39;m trying to learn `RxJava` and am still very new to it so forgive me if this is an obvious question or explained in the docs somewhere; I have tried searching. Any help would be much appreciated.


  [1]: https://github.com/BoltsFramework/Bolts-Android#tasks-in-parallel

Combine a list of Observables and wait until all completed

I am using rxjava in my Android app to handle network requests asynchronously. Now I would like to retry a failed network request only after a certain time has passed.

Is there any way to use retry() on an Observable but to retry only after a certain delay?

Is there a way to let the Observable know that is is currently being retried (as opposed to tried for the first time)?

I had a look at debounce()/throttleWithTimeout() but they seem to be doing something different.

**Edit:**

I think I found one way to do it, but I&#39;d be interested in either confirmation that this is the correct way to do it or for other, better ways.

What I am doing is this: In the call() method of my Observable.OnSubscribe, before I call the Subscribers onError() method, I simply let the Thread sleep for the desired amount of time. So, to retry every 1000 milliseconds, I do something like this:

    @Override
    public void call(Subscriber&lt;? super List&lt;ProductNode&gt;&gt; subscriber) {
	    try {
		    Log.d(TAG, &quot;trying to load all products with pid: &quot; + pid);
		    subscriber.onNext(productClient.getProductNodesForParentId(pid));
		    subscriber.onCompleted();
	    } catch (Exception e) {
		    try {
			    Thread.sleep(1000);
		    } catch (InterruptedException e1) {
			    e.printStackTrace();
		    }
		    subscriber.onError(e);
	    }
    }

Since this method is running on an IO thread anyway it does not block the UI. The only problem I can see is that even the first error is reported with delay so the delay is there even if there&#39;s no retry(). I&#39;d like it better if the delay wasn&#39;t applied *after* an error but instead *before* a retry (but not before the first try, obviously).

rxjava: Can I use retry() but with delay?

I&#39;m using RxJava in one of my projects, I converted one of my classes to Kotlin using the Android Studio plugin and in one of map `flatMap` lambda (Func1 in java), intermediates returns looks like the following `@Func1`.

I have no idea what this means. 


    something.flatMap(Func1&lt;ArticleCriteria, Observable&lt;Pair&lt;String, String&gt;&gt;&gt; {
        val isTemporaryClone = it.isATemporaryClone
        val isTheOriginalToken = it.tokenIsOriginalHere

        if (isTemporaryClone) {
            if (!isTheOriginalToken) {
                return@Func1 paramsError(&quot;Token is always original for temp articles&quot;)
            }

            return@Func1 mJobRunner.doNotRun(DeleteArticleJob.TAG)
                                .doOnNext(deletePersonalActionById(articleId))
        }

        runArticleJobAsync(DeleteArticleJob.TAG, it)
    })

Kotlin: Whats does &quot;return@&quot; mean?

I am trying to decipher the following function:

    Subscription getCar(id, Observer&lt;Car&gt; observer) {
        return getCarDetails(id, new Observer&lt;CarDetails&gt; {
                                 @Override
                                 onNext(CarDetails details) {           
                                     observer.onNext(details.getCar());
                                 } });
    }

I got a good intro to rxjava from http://blog.danlew.net/2014/09/15/grokking-rxjava-part-1/ but it only mentioned Observer in passing, saying you&#39;ll be using Subscriber most of the time to consumer items emitted from an Observable. 

Can someone explain to me 


 1. What is an observer?
 2. How is an observer different than a subscriber?
 3.  What does the above code snippet do?

[Javadoc](http://reactivex.io/RxJava/javadoc/) made it seem just like a subscriber.  The javadoc for subscriber says it implements observer and subscription.  I am very confused.

What is the difference between an Observer and a Subscriber?

I am just learning Rx-java and Rxandroid2 and I am just confused what is the major difference between in SubscribeOn and ObserveOn.

Rxandroid What&#39;s the difference between SubscribeOn and ObserveOn

I am chaining async operations using RxJava, and I&#39;d like to pass some variable downstream:

    Observable
       .from(modifications)
       .flatmap( (data1) -&gt; { return op1(data1); })
       ...
       .flatmap( (data2) -&gt; { 
           // How to access data1 here ?
           return op2(data2);
       })

It seems like a common pattern but I couldn&#39;t find information about it.


In RxJava, how to pass a variable along when chaining observables?

What are the benefits to use `Schedulers.newThread()` vs `Schedulers.io()` in `Retrofit` network request. I have seen many examples that use `io()`, but I want to understand why.

Example situation:

    observable.onErrorResumeNext(refreshTokenAndRetry(observable))
        .subscribeOn(Schedulers.newThread())
        .observeOn(AndroidSchedulers.mainThread())...

vs

    observable.onErrorResumeNext(refreshTokenAndRetry(observable))
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())...

One of the reasons i have seen is - 

`newThread()` creates a new thread for each unit of work. `io()` will use a thread pool

But what is the influence of that argument on the app? And what other aspects there are?

Retrofit with Rxjava Schedulers.newThread() vs Schedulers.io()

    @DELETE(&quot;/job/deletejob&quot;)
     Observable&lt;JobDeleteResponseModel&gt; jobDelete( @Body JobDeleteRequestModel model);

am getting this error:

&gt; Non-body HTTP method cannot contain @Body or @TypedOutput

@DELETE method is not supporting(Non-body HTTP method cannot contain @Body or @TypedOutput.)

In RxJava 1, there was CompositeSubscription, but that is not present in RxJava2, There is something CompositeDisposable in rxJava2. How do I use CompositeDisposable or Disposable in RxJava2?

How to use CompositeDisposable of RxJava 2?

I have two Completable. I would like to do following scenario:
If first Completable gets to onComplete , continue with second Completable. The final results would be onComplete of second Completable.

This is how I do it when I have Single **getUserIdAlreadySavedInDevice()** and Completable **login()**:

    @Override
    public Completable loginUserThatIsAlreadySavedInDevice(String password) {
        return getUserIdAlreadySavedInDevice()
                .flatMapCompletable(s -&gt; login(password, s))
                
    }



How to chain two Completable in RxJava2

It seems on every iteration of Java for the last few major releases, there are consistently new ways to manage concurrent tasks.

In Java 9, we have the [Flow API][1] which resembles the [Flowable API][2] of RxJava but with Java 9 has a much simpler set of classes and interfaces.

**Java 9**

Has a `Flow.Publisher`, `Flow.Subscriber`, `Flow.Processor`, `Flow.Subscription`, and `SubmissionPublisher`, and that&#39;s about it.

**RxJava**

Has whole *packages* of [Flow API][1]-like classes, i.e. `io.reactivex.flowables`, `io.reactivex.subscribers`, `io.reactivex.processors`, `io.reactivex.observers`, and `io.reactivex.observables` which seem to do something similar.

What are the main differences between these two libraries? Why would someone use the Java 9 Flow library over the much more diverse RxJava library or vice versa?

  [1]: https://docs.oracle.com/javase/9/docs/api/index.html?java/util/concurrent/Flow.html
  [2]: http://reactivex.io/RxJava/javadoc/index.html?io/reactivex/Flowable.html

Difference between RxJava API and the Java 9 Flow API

I&#39;m struggling to implement something I assumed would be fairly simple in Rx.

I have a list of items, and I want to have each item emitted with a delay.

It seems the Rx delay() operator just shifts the emission of all items by the specified delay, not each individual item.

Here&#39;s some testing code. It groups items in a list. Each group should then have a delay applied before being emitted.

    Observable.range(1, 5)
        .groupBy(n -&gt; n % 5)
        .flatMap(g -&gt; g.toList())
        .delay(50, TimeUnit.MILLISECONDS)
        .doOnNext(item -&gt; {
            System.out.println(System.currentTimeMillis() - timeNow);
            System.out.println(item);
            System.out.println(&quot; &quot;);
        }).toList().toBlocking().first();

The result is:

    154ms
    [5]
 
    155ms
    [2]
 
    155ms
    [1]
 
    155ms
    [3]

    155ms
    [4]
 
But what I would expect to see is something like this:

    174ms
    [5]
     
    230ms
    [2]
     
    285ms
    [1]
     
    345ms
    [3]
     
    399ms
    [4]
     

What am I doing wrong?

RxJava delay for each item of list emitted

I am encountering a RuntimeException when attempting to run JUnit tests for a presenter that is using `observeOn(AndroidSchedulers.mainThread())`. 

Since they are pure JUnit tests and not Android instrumentation tests, they don&#39;t have access to Android dependencies, causing me to encounter the following error when executing the tests:

    java.lang.ExceptionInInitializerError
    	at io.reactivex.android.schedulers.AndroidSchedulers$1.call(AndroidSchedulers.java:35)
    	at io.reactivex.android.schedulers.AndroidSchedulers$1.call(AndroidSchedulers.java:33)
    	at io.reactivex.android.plugins.RxAndroidPlugins.callRequireNonNull(RxAndroidPlugins.java:70)
    	at io.reactivex.android.plugins.RxAndroidPlugins.initMainThreadScheduler(RxAndroidPlugins.java:40)
    	at io.reactivex.android.schedulers.AndroidSchedulers.&lt;clinit&gt;(AndroidSchedulers.java:32)
    	…
    Caused by: java.lang.RuntimeException: Method getMainLooper in android.os.Looper not mocked. See http://g.co/androidstudio/not-mocked for details.
    	at android.os.Looper.getMainLooper(Looper.java)
    	at io.reactivex.android.schedulers.AndroidSchedulers$MainHolder.&lt;clinit&gt;(AndroidSchedulers.java:29)
    	...


    java.lang.NoClassDefFoundError: Could not initialize class io.reactivex.android.schedulers.AndroidSchedulers
    	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    	at java.lang.reflect.Method.invoke(Method.java:498)
    	…

Android RxJava 2 JUnit test - getMainLooper in android.os.Looper not mocked RuntimeException

When should we use [doOnNext()][1] from Observable instead of just onNext()?


  [1]: http://reactivex.io/RxJava/javadoc/rx/Observable.html#doOnNext-rx.functions.Action1-

What is the purpose of doOnNext(...) in RxJava

I have the List of SourceObjects and I need to convert it to the List of ResultObjects.

I can fetch one object to another using method of ResultObject:

    convertFromSource(srcObj);

of course I can do it like this:

    public void onNext(List&lt;SourceObject&gt; srcObjects) {
       List&lt;ResultsObject&gt; resObjects = new ArrayList&lt;&gt;();
       for (SourceObject srcObj : srcObjects) {
           resObjects.add(new ResultsObject().convertFromSource(srcObj));
       }
    }

but I will be very appreciate to someone who can show how to do the same using **rxJava**.

RxJava: How to convert List of objects to List of another objects

Given a list of cars (`List&lt;Car&gt; cars`), I can do:

    Observable.just(cars); //returns an Observable that emits one List&lt;Car&gt;
    Observable.from(cars); //returns an Observable that emits a squence of Car

Is there a way I can go from an `Observable` of a `List&lt;Car&gt;` to a sequence of `Observable&lt;Car&gt;`?

Something like a `from` without parameters

    Obserable.just(cars).from()

Convert Observable&lt;List&lt;Car&gt;&gt; to a sequence of Observable&lt;Car&gt; in RxJava

It seems that these 2 functions are pretty similar. They have same signature (accepting `rx.functions.Func1&lt;? super T, ? extends Observable&lt;? extends R&gt;&gt; func`), and their marble diagrams look exactly same. Can&#39;t paste the pics here, but here&#39;s one for [concatMap][1], and here&#39;s one for [flatMap][2]. There seems to be some subtle difference in the description of resulting `Observable`, where one produced by `concatMap` contains items that result from concatinating resulting Observables, and the one produced by `flatMap` contains items that result from first merging the resulting Observables, and emitting the result of that merger.

However, this subtlety is totally unclear to me. Can anyone give a better explanation of this difference, and ideally give some examples illustrating this difference.

  [1]: http://netflix.github.io/RxJava/javadoc/rx/Observable.html#concatMap%28rx.functions.Func1%29 &quot;concatMap&quot;
  [2]: http://netflix.github.io/RxJava/javadoc/rx/Observable.html#flatMap%28rx.functions.Func1%29 &quot;flatMap&quot;

What is the difference between concatMap and flatMap in RxJava

I am having a lot of trouble understanding the zip operator in RxJava for my android project. 
Problem
I need to be able to send a network request to upload a video
Then i need to send a network request to upload a picture to go with it
finally i need to add a description and use the responses from the previous two requests to upload the location urls of the video and picture along with the description to my server.

I assumed that the zip operator would be perfect for this task as I understood we could take the response of two observables (video and picture requests) and use them for my final task.
But I cant seem to get this to occur how I envision it.

I am looking for someone to answer how this can be done conceptually with a bit of psuedo code.
Thank you

All Rx Java Solutions on Gang of Coders

Total of 37 Rx Java Solutions