Download file with OkHttp3+RxJava2 with progress notification

OkHttpDownloadProgress.java

@RunWith(AndroidJUnit4.class)
public class DownloadSample {

    private Service mService;

    public DownloadSample() {
        Retrofit retrofit = new Retrofit.Builder()
                .client(new OkHttpClient())
                .baseUrl("https://github.com")
                .addConverterFactory(GsonConverterFactory.create())
                .addCallAdapterFactory(RxJava2CallAdapterFactory.create())
                .build();
        mService = retrofit.create(Service.class);
    }

    @Test
    public void testDownload() {
        Context appContext = InstrumentationRegistry.getTargetContext();
        File dir = appContext.getFilesDir();
        download("http://7u2jir.com1.z0.glb.clouddn.com/img/okhttp.png", dir).subscribe(new Observer<ProgressEvent>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(ProgressEvent progressEvent) {
                Log.i("Download: ", progressEvent.progress + "%" + " " + progressEvent.downloadIdentifier);
            }

            @Override
            public void onError(Throwable e) {
                Log.e("Download", "Failed", e);
            }

            @Override
            public void onComplete() {
                Log.i("Download", "Complete");
            }
        });
        try {
            Thread.sleep(10*1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    /**
     * extract file extension, return .download for undefined extension
     *
     * @param path url or file path
     * @return extension
     */
    public static String extension(@NonNull String path) {
        String subPath = path;
        if (URLUtil.isValidUrl(path)) {
            try {
                subPath = new URL(path).getPath();
            } catch (MalformedURLException e) {
                e.printStackTrace();
            }
        }
        int dotIndex = subPath.lastIndexOf('.');
        String extension = null;
        if (dotIndex >= 0) {
            extension = subPath.substring(dotIndex);
        }
        return extension != null ? extension : ".download";
    }

    public Observable<ProgressEvent> download(final String url, final File dir) {
        return mService.download(url).flatMap(new Function<Response<ResponseBody>, ObservableSource<ProgressEvent>>() {
            @Override
            public ObservableSource<ProgressEvent> apply(Response<ResponseBody> response) throws Exception {
                String fileName = md5(url) + extension(url);
                final File outputFile = new File(dir, fileName);
                final ResponseBody responseBody = response.body();
                final String name = outputFile.getAbsolutePath();
                return Observable.create(new ObservableOnSubscribe<ProgressEvent>() {

                    @Override
                    public void subscribe(ObservableEmitter<ProgressEvent> emitter) throws Exception {
                        try {
                            int count;
                            byte data[] = new byte[1024 * 4];
                            long fileSize = responseBody.contentLength();
                            InputStream bis = new BufferedInputStream(responseBody.byteStream(), 1024 * 8);
                            File tmp = File.createTempFile("download", null);
                            OutputStream output = new FileOutputStream(tmp);
                            long readTotal = 0;
                            while ((count = bis.read(data)) != -1) {
                                readTotal += count;
                                ProgressEvent download = new ProgressEvent(name, fileSize, readTotal);
                                emitter.onNext(download);
                                output.write(data, 0, count);
                            }
                            output.flush();
                            output.close();
                            bis.close();
                            Utils.copy(tmp, outputFile);
                            emitter.onComplete();
                        } catch (IOException e) {
                            e.printStackTrace();
                            emitter.onError(e);
                        }
                    }
                });
            }
        }).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread());
    }

    public interface Service {
        @GET
        @Streaming
        Observable<Response<ResponseBody>> download(@Url String url);
    }

    public class ProgressEvent {
        final int progress;
        final long contentLength;
        final String downloadIdentifier;
        final long bytesRead;

        public ProgressEvent(String identifier, long contentLength, long bytesRead) {
            this.contentLength = contentLength;
            this.progress = (int) (bytesRead / (contentLength / 100f));
            downloadIdentifier = identifier;
            this.bytesRead = bytesRead;
        }

        public int getProgress() {
            return progress;
        }

        public String getDownloadIdentifier() {
            return downloadIdentifier;
        }

        public long getBytesRead() {
            return bytesRead;
        }

        public boolean percentIsAvailable() {
            return contentLength > 0;
        }
    }

}