时间戳

首页 > 解决方案 > RxJava - observable 中的多个任务

java - RxJava - observable 中的多个任务

问题描述

我有一个包含许多对象的列表，我将其用作可观察对象，每个对象都有两个 url，在每个消费中，我应该确保两个 url 都成功下载，如果一个 url 没有下载，RxJava 应该停止工作，我我是 RxJava 的新手，我可以做一些简单的工作。

这是我的对象

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

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

标签： javaandroidrx-javarx-android

解决方案

一种方法是：

  1. 使用Observable.fromIterable工厂方法从初始设置流List<MediaChunk>
  2. flatMap来自该流的每个发射，并使用Observable.just工厂方法从每个MediaChunk实例创建一个音频 + 视频 URL 流

结果是一个扁平化的 URL 流，订阅者可以为其插入自己的onNext、onError和onComplete处理程序。

代码看起来像这样：

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()");
            }
        );

不确定这是否符合要求，但希望它至少足以激发一些想法。

更新 - 示例将排放映射到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() + ")");
        }
    );

推荐阅读