RxDownload interval

gistfile1.txt

Observable<DownloadState> observable =
                Observable.interval(PROGRESS_INTERVAL_MILLIS, TimeUnit.MILLISECONDS)
                        .flatMap(new Function<Long, ObservableSource<DownloadState>>() {
                            @Override
                            public ObservableSource<DownloadState> apply(Long aLong) throws Exception {
                                DownloadManager.Query query = new DownloadManager.Query();
                                query.setFilterById(downloadId);
                                Cursor cursor = getDownloadManager().query(query);
                                if (!cursor.moveToFirst()) {
                                    cursor.close();
                                    downloadManager.remove(downloadId);
                                    progressSubjectMap.remove(downloadId);
                                    return Observable.error(new IllegalStateException("Cursor empty, this shouldn't happen"));
                                }
                                int bytesDownloaded = cursor.getInt(cursor
                                        .getColumnIndex(DownloadManager.COLUMN_BYTES_DOWNLOADED_SO_FAR));
                                int bytesTotal = cursor.getInt(cursor.getColumnIndex(DownloadManager.COLUMN_TOTAL_SIZE_BYTES));

                                int status = cursor.getInt(cursor.getColumnIndex(DownloadManager.COLUMN_STATUS));
                                int uriIndex = cursor.getColumnIndex(DownloadManager.COLUMN_LOCAL_URI);
                                String downloadedPackageUriString = cursor.getString(uriIndex);
                                cursor.close();
                                if (status == DownloadManager.STATUS_SUCCESSFUL) {
                                    progressSubjectMap.remove(downloadId);
                                    return Observable.just(DownloadState.create(100, downloadedPackageUriString));
                                } else if (status == DownloadManager.STATUS_FAILED) {
                                    progressSubjectMap.remove(downloadId);
                                    return Observable.error(new RuntimeException("Download not complete"));
                                }
                                final int progress = (int) ((bytesDownloaded * 1f) / bytesTotal * 100);
                                return Observable.just(DownloadState.create(progress, null));
                            }
                        })
                        .takeUntil(new Predicate<DownloadState>() {
                            @Override
                            public boolean test(DownloadState downloadState) throws Exception {
                                return !TextUtils.isEmpty(downloadState.path);
                            }
                        });