无法从 json 下载多个视频文件并将下载路径设置为同一个列表

Question

我正在使用 RxAndroid、Retrofit 和 SqlBrite。

POJO 类：
例如：file_path = "....../videos/.mp4"

public class VideoResponse {
    @SerializedName("files")
    @Expose
    private List<VideoFiles> files = null;
    .....
}

public class VideoFiles {
    @SerializedName("file_path")
    @Expose
    private String remotePath;

    private String localPath;
    .....
}

将列表从 apiService 传递给 setLocalPath。

 @Inject
    public RemoteDataSource(ApiService service,DownloadUtils downloadUtils) {
        this.service = service;
        this.downloadUtils = downloadUtils;
    }
    @Override
        public Observable<List<VideoResponse>> getVideoResponse() {
            return service.getVideoResponseFromServer()
                    .compose(RxUtils.applySchedulers())
       ==>             .map(this::setVideoLocalPath)
                    .doOnSubscribe(disposable -> Timber.d("*** Video Sync Started....."))
                    .doOnError(throwable -> Timber.d("*** Video Sync Failed ...."))
                    .doOnComplete(() -> Timber.d(" *** Video Sync Complete...."));
        }

将每个远程路径传递给 DownloadUtils 并取回更改后的 VideoResponse 列表。

   private List<VideoResponse> setVideoLocalPath(List<VideoResponse> videoResponses) {
        for (VideoResponse r : videoResponses) {
            for (VideoFiles file : r.getFiles()) {
                downloadUtils.downloadVideoFromInternet(file, service);
            }
        }
        return videoResponses;
    }

下载并设置本地路径；

public class DownloadUtils {

        public void downloadVideoFromInternet(VideoFiles video, ApiService service) {
        service.downloadFileByUrl(video.getRemotePath())
                .flatMap(processResponse("video", video.getFileTitle()))
                .subscribe(handleVideoResult(video));
    }

    private Observer<? super File> handleVideoResult(VideoFiles video) {
        return new Observer<File>() {
            @Override
            public void onSubscribe(Disposable d) {
                Timber.i("*** Download File OnSubscribe ***");
            }

            @Override
            public void onNext(File file) {
                Timber.d(" $$$$ Video File Path $$$ -> %s", file.getAbsolutePath());
                video.setLocalPath(file.getAbsolutePath());
            }

            @Override
            public void onError(Throwable e) {
                Timber.e(e);
            }

            @Override
            public void onComplete() {
                Timber.i("*** Download File Completed ****");
            }
        };
    }

    private Function<Response<ResponseBody>, Observable<File>> processResponse(String folderName, String fileTitle) {
        return response -> saveToDisk(response, folderName, fileTitle);
    }

    private Observable<File> saveToDisk(Response<ResponseBody> response, String fileTitle, String folderName) {
        return Observable.create(subscriber -> {
            try {
                File file = new File("/data/aster/" + folderName + fileTitle);
                if (!file.exists()) {
                    file.mkdirs();
                }
                BufferedSink bufferedSink = Okio.buffer(Okio.sink(file));
                bufferedSink.writeAll(response.body().source());
                bufferedSink.close();
                subscriber.onNext(file);
                subscriber.onComplete();
            } catch (IOException e) {
                e.printStackTrace();
                subscriber.onError(e);
            }
        });
    }
}

问题是视频文件没有被下载，并且每个都在订阅时停止。 将值传递给 setLocalVideoPath 后，下载没有完成，我得到 NetworkOnMainThreadException 并且应用程序崩溃..有没有更好的方法来实现这个逻辑..!!请帮忙。

Answer 1

如果 RxUtils.applySchedulers 正在应用以下内容，那么在您完成映射操作并随后点击 service.downloadFileByUrl 这将在主线程上执行。

.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread());

如果您在地图操作之后移动 observeOn 调用，那么 service.downloadFileByUrl 应该在主线程之外执行，即

@Override
public Observable<List<VideoResponse>> getVideoResponse() {
    return service.getVideoResponseFromServer()
            .subscribeOn(Schedulers.io())
            .map(this::setVideoLocalPath)
            .observeOn(AndroidSchedulers.mainThread());
            .doOnSubscribe(disposable -> Timber.d("*** Video Sync Started....."))
            .doOnError(throwable -> Timber.d("*** Video Sync Failed ...."))
            .doOnComplete(() -> Timber.d(" *** Video Sync Complete...."));
}

Answer 2

问题在于您执行操作的顺序，特别是这里

....
.compose(RxUtils.applySchedulers())
.map(this::setVideoLocalPath)
...

如果您查看RxUtils.applySchedulers() 的源代码，您会发现转换器如下所示：

static Observable.Transformer schedulersTransformer = new Observable.Transformer<Object, Object>() {
    @Override public Observable<Object> call(Observable<Object> observable) {
        return observable
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread());
    }
};

所以，observeOn(AndroidSchedulers.mainThread()) 将执行工作的线程切换到主线程，当您的map 下一步执行时，它会在主线程上执行工作。我看不出你的map 运营商应该在之后去的任何理由。只需将顺序更改如下：

....
.map(this::setVideoLocalPath)
.compose(RxUtils.applySchedulers())
...

在这里，执行链是固定的，您的map 运算符将在io 线程上工作。另外，请注意您的doOnSubscribe 在在 compose(RxUtils.applySchedulers()) 之后执行主线程上的工作。