RxJava - Multiple tasks in a observable

Question

I have a list that contain many objects and i will use it as observable, Each object has two url, In each consuming, I should ensure both url downloaded successfully, If one of url did not downloaded, RxJava should be stop the job, I am new in RxJava and i can just do simple works.

This is my object

public class mediaChunk
{
    public String audioChunkUrl;
    public String videoChunkUrl;
}

List<mediaChunk> list = new ArrayList<>();

homerman · Accepted Answer · 2018-08-14 13:41:54Z

2

one approach would be:

use the Observable.fromIterable factory method to setup the stream from the initial List<MediaChunk>
flatMap each emission from that stream and use the Observable.just factory method to create a stream of audio + video URLs from each MediaChunk instance

the result is a flattened stream of URLs for which subscribers can plug in their own onNext, onError, and onComplete handlers.

the code would look something like this:

Observable.fromIterable(
        Arrays.asList(
            new MediaChunk("audio-1", "video-1"),
            new MediaChunk("audio-2", "video-2"),
            new MediaChunk("audio-3", "video-3")
        ))
        .flatMap(chunk -> Observable.just(chunk.audioChunkUrl, chunk.videoChunkUrl))
        .subscribe(
            value -> {
              System.out.println("## onNext(" + value + ")");
            },
            error -> {
              System.out.println("## onError(" + error.getMessage() + ")");
            },
            () -> {
              System.out.println("## onComplete()");
            }
        );

not sure if this fits the bill, but hopefully it's enough to at least inspire some thought.

Update - example mapping emissions to Completable

Observable.fromIterable(
    Arrays.asList(
        new MediaChunk("audio-1", "video-1"),
        new MediaChunk("audio-2", "video-2"),
        new MediaChunk("audio-3", "video-3")
    ))
    .flatMap(chunk -> Observable.just(chunk.audioChunkUrl, chunk.videoChunkUrl))
    .flatMapCompletable(url -> {
      return Completable.fromCallable(() -> {
        return "## ...handling network call for [" + url + "]";
      });
    })
    .subscribe(
        () -> {
          System.out.println("## onComplete()");
        },
        error -> {
          System.out.println("## onError(" + error.getMessage() + ")");
        }
    );

edited Aug 14, 2018 at 13:41

answered Aug 14, 2018 at 7:23

homerman

3,6291 gold badge19 silver badges35 bronze badges

Sign up to request clarification or add additional context in comments.

2 Comments

Ali Over a year ago

Thanks homerman.Should i implement downloading job in subscribe, or i should implement it in flatMap? like this:

.flatMap(chunk -> Observable.just(chunk.videoUrl, chunk.audioUrl).doOnNext(value -> {                     download(value);                 }))

homerman Over a year ago

there may be a compelling technical argument for one approach over the other, but i can't think of it. try it both ways and see which is the better fit for your design. i might offer some food for thought - if the download job is simply stashing the files somewhere and the subscriber doesn't really have any work to do in onNext, then you might consider converting each emission to a Completable to simply signal that all the downloads have completed, or something failed. i've updated my answer with an example.

Collectives™ on Stack Overflow

RxJava - Multiple tasks in a observable

1 Answer 1

2 Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

2 Comments

Your Answer

Sign up or log in

Post as a guest

Related