Невозможно загрузить несколько видеофайлов из 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;
    .....
}

Передача списка в setLocalPath из apiService.

 @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

Проблема в том порядке, в котором вы выполняете свои операции, в частности, здесь

....
.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()).

Answer 2

Если RxUtils.applySchedulers применяет следующее, то в момент, когда вы выполняете операцию отображения и затем нажимаете service.downloadFileByUrl, это будет выполняться в главном потоке.

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

Если вы перемещаете вызов наблюдаем после операции сопоставления, то 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...."));
}