Limit throughput with RxJava

Question

The case I'm into right now is quite hard to explain so I will write a simpler version just to explain the issue.

I have an Observable.from() which emits a sequence of files defined by an ArrayList of files. All of these files should be uploaded to a server. For that I have an function that does the job and returns an Observable.

Observable<Response> uploadFile(File file);

When I run this code it gets crazy, the Observable.from() emits all of the files and they are uploaded all at ones, or at least for a max of threads it can handle.

I want to have a max of 2 file uploads in parallel. Is there any operator that can handle this for me?

I tried buffer, window and some others but they seems to only emit two items together instead of having two parallel file uploads constantly. I also tried to set a max threads pool on the uploading part, but this cannot be used in my case.

There should be a simple operator for this right? Am I missing something?

Answer 1

I think all files are uploaded in parallel because you're using flatMap(), which executes all transformations simultaneously. Instead you should use concatMap(), which runs one transformation after another. And to run two parallel uploads you need to call window(2) on you files observable and then invoke flatMap() as you did in your code.

Observable<Response> responses = 
    files
      .window(2)
      .concatMap(windowFiles ->
        windowFiles.flatMap(file -> uploadFile(file));
      );

UPDATE:

I found a better solution, which does exactly what you want. There's an overload of flatMap() that accepts the max number of concurrent threads.

Observable<Response> responses = 
    files
      .onBackpressureBuffer()
      .flatMap(index -> {
        return uploadFile(file).subscribeOn(Schedulers.io());
      }, 2);