Downloading files using OkHttp, Okio and RxJava

Question

I'm trying to download files using OkHttp and writing to disk with Okio. Also I've created an rx observable for this process. It is working, however it is noticeably slower than what I had previously used (Koush's Ion library).

Here's how I create the observable:

public Observable<FilesWrapper> download(List<Thing> things) {
    return Observable.from(things)
        .map(thing -> {
            File file = new File(getExternalCacheDir() + File.separator + thing.getName());

            if (!file.exists()) {
                Request request = new Request.Builder().url(thing.getUrl()).build();
                Response response;
                try {
                    response = client.newCall(request).execute();
                    if (!response.isSuccessful()) new IOException();
                    else {
                        BufferedSink sink = Okio.buffer(Okio.sink(file));
                        sink.writeAll(response.body().source());
                        sink.close();
                    }
                } catch (IOException e) {
                    new IOException();
                }
            }

            return file;
        })
        .toList()
        .map(files -> new FilesWrapper(files);
}

Does anyone know what could be causing the slow speed, or if I am using the operators incorrectly?

Answer 1

Using flatMap instead of map will allow you to execute the downloads in parallel:

public Observable<FilesWrapper> download(List<Thing> things) {
    return Observable.from(things)
            .flatMap(thing -> {
                File file = new File(getExternalCacheDir() + File.separator + thing.getName());
                if (file.exists()) {
                    return Observable.just(file);
                }

                final Observable<File> fileObservable = Observable.create(sub -> {
                    if (sub.isUnsubscribed()) {
                        return;
                    }

                    Request request = new Request.Builder().url(thing.getUrl()).build();

                    Response response;
                    try {
                        response = client.newCall(request).execute();
                        if (!response.isSuccessful()) { throw new IOException(); }
                    } catch (IOException io) {
                        throw OnErrorThrowable.from(OnErrorThrowable.addValueAsLastCause(io, thing));
                    }

                    if (!sub.isUnsubscribed()) {
                        try (BufferedSink sink = Okio.buffer(Okio.sink(file))) {
                            sink.writeAll(response.body().source());
                        } catch (IOException io) {
                            throw OnErrorThrowable.from(OnErrorThrowable.addValueAsLastCause(io, thing));
                        }
                        sub.onNext(file);
                        sub.onCompleted();
                    }

                });
                return fileObservable.subscribeOn(Schedulers.io());
            }, 5)
            .toList()
            .map(files -> new FilesWrapper(files));
}

We limit the number of simultaneous requests per subscriber using the maxConcurrent on flatMap.