CNLive Android Team / DownloadFileDemo · Files

package com.cnlive.downloadfile.download.limit;

import com.cnlive.downloadfile.Constant;
import com.cnlive.downloadfile.MyApplication;

import org.greenrobot.eventbus.EventBus;

import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.concurrent.atomic.AtomicReference;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.ObservableSource;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.functions.Function;
import io.reactivex.functions.Predicate;
import io.reactivex.schedulers.Schedulers;
import okhttp3.Call;
import okhttp3.OkHttpClient;
import okhttp3.Request;
import okhttp3.Response;

/**
 * About: 下载管理（限制最大个数）
 */
public class DownloadLimitManager {

    private static final AtomicReference<DownloadLimitManager> INSTANCE = new AtomicReference<>();
    private OkHttpClient mClient;
    private HashMap<String, Call> downCalls; // 用来存放各个下载的请求
    private List<DownloadInfo> downWait;           // 用来存放等待下载的请求
    private DownloadInfo currentDownInfo;

    public static DownloadLimitManager getInstance() {
        for (; ; ) {
            DownloadLimitManager current = INSTANCE.get();
            if (current != null) {
                return current;
            }
            current = new DownloadLimitManager();
            if (INSTANCE.compareAndSet(null, current)) {
                return current;
            }
        }
    }

    private DownloadLimitManager() {
        downCalls = new HashMap<>();
        mClient = new OkHttpClient.Builder().build();
        downWait = new ArrayList<>();
    }

    /**
     * 查看是否在下载任务中
     *
     * @param url
     * @return
     */
    public boolean getDownloadUrl(String url) {
        return downCalls.containsKey(url);
    }

    /**
     * 查看是否在等待任务中
     *
     * @param url
     * @return
     */
    public boolean getWaitUrl(String url) {
        for (DownloadInfo item : downWait) {
            if (item.getUrl().equals(url)) {
                return true;
            }
        }
        return false;
    }

    /**
     * 开始下载
     *
     * @param downloadInfo 下载请求的网址
     */
    public void download(final DownloadInfo downloadInfo) {
        List<DownloadInfo> list = MyApplication.getInstances().getDaoSession().loadAll(DownloadInfo.class);
        downWait.clear();
        downWait = list;
        if (currentDownInfo != null) pauseDownload(currentDownInfo, true);
        if (downCalls.containsKey(downloadInfo.getUrl())) downCalls.remove(downloadInfo.getUrl());
        Observable.just(downloadInfo.getUrl())
                .filter(new Predicate<String>() { // 过滤 call的map中已经有了,就证明正在下载,则这次不下载
                    @Override
                    public boolean test(String s) {
                        boolean flag = downCalls.containsKey(s);
                        if (flag) {
                            // 如果已经在下载，并且不是暂停的数据，查找下一个文件进行下载
                            downNext();
                            return false;
                        } else {
                            // 判断如果正在下载的个数达到最大限制，存到等下下载列表中
                            return true;
                        }
                    }
                })
                .flatMap(new Function<String, ObservableSource<?>>() { // 生成 DownloadInfo
                    @Override
                    public ObservableSource<?> apply(String s) {
                        return Observable.just(createDownInfo(downloadInfo));
                    }
                })
                .map(new Function<Object, DownloadInfo>() { // 如果已经下载，重新命名
                    @Override
                    public DownloadInfo apply(Object o) {
                        return getRealFileName((DownloadInfo) o);
                    }
                })
                .flatMap(new Function<DownloadInfo, ObservableSource<DownloadInfo>>() { // 下载
                    @Override
                    public ObservableSource<DownloadInfo> apply(DownloadInfo downloadInfo) {
                        currentDownInfo = downloadInfo;
                        return Observable.create(new DownloadSubscribe(downloadInfo));
                    }
                })
                .observeOn(AndroidSchedulers.mainThread()) // 在主线程中回调
                .subscribeOn(Schedulers.io()) //  在子线程中执行
                .subscribe(new DownloadLimitObserver()); //  添加观察者，监听下载进度
    }

    /**
     * 下载等待下载中的第一条
     */
    public void downNext() {
        for (int i = 0; i < downWait.size(); i++) {
            if (!downCalls.containsKey(downWait.get(i).getUrl()) && !downWait.get(i).getUrl().equals(currentDownInfo)) {
                download(downWait.get(i));
                break;
            }
        }
    }

    /**
     * 下载取消或者暂停
     *
     * @param downloadInfo
     * @param isPauseCurrent 点击开始下载，那正在下载的文件要暂停，并且执行点击的item，不进行next
     */
    public void pauseDownload(DownloadInfo downloadInfo, boolean isPauseCurrent) {
        Call call = downCalls.get(downloadInfo.getUrl());
        if (call != null) {
            call.cancel();//取消
        }
        downloadInfo.setDownloadStatus(DownloadStatus.DOWNLOAD_PAUSE);
        MyApplication.getInstances().getDaoSession().update(downloadInfo);
        EventBus.getDefault().post(downloadInfo);//通知页面
        for(int i = 0;i < downWait.size();i++){
            if(downWait.get(i).getUrl().equals(downloadInfo.getUrl())){
                downWait.set(i,downloadInfo);
                break;
            }
        }
        //如果暂停的当前文件，next
        if (currentDownInfo != null && downloadInfo.getUrl().equals(currentDownInfo.getUrl()) && !isPauseCurrent) {
            downNext();
        } else {
            downCalls.put(downloadInfo.getUrl(), null);
        }
    }

    /**
     * 出错暂停,突然断网
     */
    public void pauseErrorDownload() {
        if (currentDownInfo != null) {
            Call call = downCalls.get(currentDownInfo);
            if (call != null) {
                call.cancel();//取消
            }
            downCalls.remove(currentDownInfo.getUrl());
        }
        for(int i = 0;i < downWait.size();i++){
            if(downWait.get(i).getDownloadStatus().equals(DownloadStatus.DOWNLOAD_WAIT)){
                downWait.get(i).setDownloadStatus(DownloadStatus.DOWNLOAD_NET_PAUSE);
            }

            if(currentDownInfo != null && downWait.get(i).getUrl().equals(currentDownInfo.getUrl())){
                currentDownInfo.setDownloadStatus(DownloadStatus.DOWNLOAD_NET_PAUSE);
                downWait.set(i,currentDownInfo);
            }
            EventBus.getDefault().post(downWait.get(i));//通知页面
            MyApplication.getInstances().getDaoSession().update(downWait.get(i));
        }

    }


    /**
     * 暂停状态转变为等待状态
     * @param downloadInfo
     */
    public void pauseToWaitDownload(DownloadInfo downloadInfo){
        boolean hasWaiting = false;
        int downIndex = 0;
        for(int i = 0;i < downWait.size();i++){
            if(downWait.get(i).getUrl().equals(downloadInfo.getUrl())){
                downIndex = i;
                if(hasWaiting){
                    break;
                }
            }
            if(downWait.get(i).getDownloadStatus().equals(DownloadStatus.DOWNLOAD_WAIT) || downWait.get(i).getDownloadStatus().equals(DownloadStatus.DOWNLOAD)){
                hasWaiting = true;
                if(downIndex != 0) {
                    break;
                }
            }
        }
        if(hasWaiting){
            downloadInfo.setDownloadStatus(DownloadStatus.DOWNLOAD_WAIT);
            downCalls.remove(downloadInfo.getUrl());
            EventBus.getDefault().post(downloadInfo);//通知页面
        }else {
            downloadInfo.setDownloadStatus(DownloadStatus.DOWNLOAD);
            download(downloadInfo);
        }
        downWait.set(downIndex,downloadInfo);
        MyApplication.getInstances().getDaoSession().update(downloadInfo);//更新数据库状态
    }

    /**
     * 暂停所有的下载
     */
    public void pauseAllDownload() {
        List<DownloadInfo> list = MyApplication.getInstances().getDaoSession().loadAll(DownloadInfo.class);
        downWait.clear();
        downWait = list;
        for (int i = 0; i < downWait.size(); i++) {
            Call call = downCalls.get(downWait.get(i).getUrl());
            if (call != null) {
                call.cancel();
                if(currentDownInfo != null && downWait.get(i).getUrl() .equals(currentDownInfo.getUrl())){
                    downWait.set(i,currentDownInfo);
                }
            }else {
                downCalls.put(downWait.get(i).getUrl(),null);
            }
            downWait.get(i).setDownloadStatus(DownloadStatus.DOWNLOAD_PAUSE);//改变状态
            EventBus.getDefault().post(downWait.get(i));//通知页面
            MyApplication.getInstances().getDaoSession().update(downWait.get(i));//更新数据库状态
        }
        if(currentDownInfo != null){
            currentDownInfo = null;
        }
    }

    /**
     * 全部开始
     */
    public void startAllDownload() {
        downCalls.clear();
        downWait.clear();
        List<DownloadInfo> list = MyApplication.getInstances().getDaoSession().loadAll(DownloadInfo.class);
        downWait = list;
        if (downWait.size() > 0) {
            for (int i = 0; i < downWait.size(); i++) {
                downWait.get(i).setDownloadStatus(DownloadStatus.DOWNLOAD_WAIT);//改变状态
                MyApplication.getInstances().getDaoSession().update(downWait.get(i));//更新数据库状态
                EventBus.getDefault().post(downWait.get(i));//通知页面
            }
            download(downWait.get(0));
        }
    }


    /**
     * 全部删除清空
     */
    public void clearDownload() {
        for (int i = 0; i < downWait.size(); i++) {
            Call call = downCalls.get(currentDownInfo);
            if (call != null) {
                call.cancel();//取消
            }
            Constant.deleteFile(downWait.get(i).getUrl().substring(downWait.get(i).getUrl().lastIndexOf("/")));
        }
        currentDownInfo = null;
        downWait.clear();
        downCalls.clear();
        MyApplication.getInstances().getDaoSession().deleteAll(DownloadInfo.class);
    }

    /**
     * 取消下载 删除本地文件
     *
     * @param info
     */
    public void cancelDownload(DownloadInfo info, boolean fileDelect) {
        //本地数据操作
        for (int i = 0; i < downWait.size(); i++) {
            if (downWait.get(i).getUrl().equals(info.getUrl())) {
                downWait.remove(i);
                MyApplication.getInstances().getDaoSession().delete(info);
                break;
            }
        }
        downCalls.remove(info);
        info.setDownloadStatus(DownloadStatus.DOWNLOAD_CANCEL);
        EventBus.getDefault().post(info);
        if (fileDelect) Constant.deleteFile(info.getFileName());

        Call call = downCalls.get(info.getUrl());
        if (call != null) {
            call.cancel();//取消
        }
        //如果暂停的当前文件，next
        if (currentDownInfo != null && info.getUrl().equals(currentDownInfo.getUrl())) {
            downNext();
        }
    }

    /**
     * 得到当前正在下载的任务
     *
     * @return
     */
    public DownloadInfo getCurremtDownload() {
        return currentDownInfo;
    }

    /**
     * 创建DownInfo
     *
     * @param downloadInfo 请求网址
     * @return DownInfo
     */
    private DownloadInfo createDownInfo(DownloadInfo downloadInfo) {
        long contentLength = getContentLength(downloadInfo.getUrl());//获得文件大小
        downloadInfo.setTotal(contentLength);
        String fileName = downloadInfo.getUrl().substring(downloadInfo.getUrl().lastIndexOf("/"));
        downloadInfo.setFileName(fileName);
        return downloadInfo;
    }

    /**
     * 如果文件已下载重新命名新文件名
     *
     * @param downloadInfo
     * @return
     */
    private DownloadInfo getRealFileName(DownloadInfo downloadInfo) {
        String fileName = downloadInfo.getFileName();
        long downloadLength = 0, contentLength = downloadInfo.getTotal();
        File path = new File(Constant.FILE_PATH);
        if (!path.exists()) {
            path.mkdir();
        }
        File file = new File(Constant.FILE_PATH, fileName);
        if (file.exists()) {
            //找到了文件,代表已经下载过,则获取其长度
            downloadLength = file.length();
        }
        //之前下载过,需要重新来一个文件
        int i = 1;
        while (downloadLength >= contentLength) {
            int dotIndex = fileName.lastIndexOf(".");
            String fileNameOther;
            if (dotIndex == -1) {
                fileNameOther = fileName + "(" + i + ")";
            } else {
                fileNameOther = fileName.substring(0, dotIndex)
                        + "(" + i + ")" + fileName.substring(dotIndex);
            }
            File newFile = new File(Constant.FILE_PATH, fileNameOther);
            file = newFile;
            downloadLength = newFile.length();
            i++;
        }
        //设置改变过的文件名/大小
        downloadInfo.setProgress(downloadLength);
        downloadInfo.setFileName(file.getName());
        return downloadInfo;
    }

    private class DownloadSubscribe implements ObservableOnSubscribe<DownloadInfo> {
        private DownloadInfo downloadInfo;

        public DownloadSubscribe(DownloadInfo downloadInfo) {
            this.downloadInfo = downloadInfo;
        }

        @Override
        public void subscribe(ObservableEmitter<DownloadInfo> e) throws Exception {
            String url = downloadInfo.getUrl();
            long downloadLength = downloadInfo.getProgress();//已经下载好的长度
            long contentLength = downloadInfo.getTotal();//文件的总长度
            //初始进度信息
            e.onNext(downloadInfo);
            Request request = new Request.Builder()
                    //确定下载的范围,添加此头,则服务器就可以跳过已经下载好的部分
                    .addHeader("RANGE", "bytes=" + downloadLength + "-" + contentLength)
                    .url(url)
                    .build();
            Call call = mClient.newCall(request);
            downCalls.put(url, call);//把这个添加到call里,方便取消

            Response response = call.execute();
            File file = new File(Constant.FILE_PATH, downloadInfo.getFileName());
            InputStream is = null;
            FileOutputStream fileOutputStream = null;
            try {
                is = response.body().byteStream();
                fileOutputStream = new FileOutputStream(file, true);
                byte[] buffer = new byte[2048];//缓冲数组2kB
                int len;
                while ((len = is.read(buffer)) != -1) {
                    fileOutputStream.write(buffer, 0, len);
                    downloadLength += len;
                    downloadInfo.setProgress(downloadLength);
                    currentDownInfo.setProgress(downloadLength);
                    e.onNext(downloadInfo);
                }
                fileOutputStream.flush();
                for (int i = 0; i < downWait.size(); i++) {
                    if (downWait.get(i).getUrl().equals(url)) {
                        MyApplication.getInstances().getDaoSession().delete(downWait.get(i));
                        downWait.remove(i);
                        downCalls.remove(url);
                        currentDownInfo = null;
                        downNext();
                        break;
                    }
                }
            } finally {
                //关闭IO流
                DownloadIO.closeAll(is, fileOutputStream);

            }
            e.onComplete();//完成
        }
    }


    /**
     * 获取下载长度
     *
     * @param downloadUrl
     * @return
     */
    private long getContentLength(String downloadUrl) {
        Request request = new Request.Builder()
                .url(downloadUrl)
                .build();
        try {
            Response response = mClient.newCall(request).execute();
            if (response != null && response.isSuccessful()) {
                long contentLength = response.body().contentLength();
                response.close();
                return contentLength == 0 ? DownloadStatus.TOTAL_ERROR : contentLength;
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        return DownloadStatus.TOTAL_ERROR;
    }

}