RxJava & OkHttp: file download with progress updates

[David Mihola]

Hello,

I am currently trying to wrap an OkHttp file download in an Observable that provides periodic updates about the download progress. Here is what I have so far:

public Observable<DownloadInfo> downloadFile(final String source, final File targetFile) {
    final Request request = new Request.Builder()
            .url(source)
            .build();

    return executeRequest(request)
            .flatMap(new Func1<Response, Observable<DownloadInfo>>() {
                @Override
                public Observable<DownloadInfo> call(Response response) {
                    return copyResponseToFile(response, targetFile);
                }
            })
            .sample(500, TimeUnit.MILLISECONDS);
}

public Observable<Response> executeRequest(final Request request) {
    return Observable.fromCallable(new Callable<Response>() {
        @Override
        public Response call() throws Exception {
            return okHttpClient.newCall(request).execute();
        }
    });
}

private final Observable<DownloadInfo> copyResponseToFile(final Response response, final File file) {
    return Observable.create(new OkIoDownloadProgressOnSubscribe(response, file));
}

final class OkIoDownloadProgressOnSubscribe implements Observable.OnSubscribe<DownloadInfo> {
    private final Response response;
    private final File targetFile;

    public OkIoDownloadProgressOnSubscribe(Response response, File targetFile) {
        this.response = response;
        this.targetFile = targetFile;
    }

    @Override
    public void call(Subscriber<? super DownloadInfo> subscriber) {
        if (response.isSuccessful()) {
            subscriber.onNext(new RequestSuccessful());

            Timber.d("request was successful");
            Timber.d("downloading to: %s", targetFile.toString());

            final long length = response.body().contentLength();
            Timber.d("total size: %d", length);

            BufferedSource source = response.body().source();
            try {
                BufferedSink sink = Okio.buffer(Okio.sink(targetFile));

                long downloadedTotal = 0;
                long downloadedSinceLastOnNext = 0;

                // make sure that we are cancelable at "any" time
                while (!subscriber.isUnsubscribed()) {

                    long bytesRead = source.read(sink.buffer(), 4096);

                    if (bytesRead == -1) {
                        break;
                    }

                    downloadedTotal += bytesRead;
                    downloadedSinceLastOnNext += bytesRead;

                    if (downloadedSinceLastOnNext > 300000) {
                        downloadedSinceLastOnNext = 0;
                        subscriber.onNext(new Downloading(downloadedTotal, length));
                        Timber.d("downloaded so far: %d", downloadedTotal);
                    }
                }

                source.close();
                sink.close();

                if (!subscriber.isUnsubscribed()) {
                    subscriber.onNext(new Downloading(downloadedTotal, length));
                    subscriber.onCompleted();
                }

            } catch (IOException e) {
                if (!subscriber.isUnsubscribed()) {
                    subscriber.onError(e);
                }
            }
        } else {
            subscriber.onNext(new DownloadError(response));
        }
    }
}

And my questions are:

1.) Is the implementation basically OK? I know that using create() with Observable.OnSubscribe should be seen as a last resort but I could not come up with a "safer" version. In particular, SyncOnSubscribe does not seem to work, because I actually do not want backpressure - the download should proceed as fast as it can and I only want just enough updates to provide some feedback.
2.) Instead of the sample(500, TimeUnit.MILLISECONDS) I also tried onBackPressureDrop. This works for some time, but then the onNexts stop completely (while the download continues), but the onCompleted still arrives. Can updates be too fast for even onBackPressureDrop or what was happening there?

Thanks for any feedback!

David