diff --git a/Aria/src/main/java/com/arialyy/aria/core/command/group/GroupCancelCmd.java b/Aria/src/main/java/com/arialyy/aria/core/command/group/GroupCancelCmd.java deleted file mode 100644 index 8bedf403..00000000 --- a/Aria/src/main/java/com/arialyy/aria/core/command/group/GroupCancelCmd.java +++ /dev/null @@ -1,34 +0,0 @@ -/* - * Copyright (C) 2016 AriaLyy(https://github.com/AriaLyy/Aria) - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package com.arialyy.aria.core.command.group; - -import com.arialyy.aria.core.inf.AbsGroupTaskWrapper; - -/** - * Created by AriaL on 2017/6/29. - * 删除任务组子任务 - */ -class GroupCancelCmd extends AbsGroupCmd { - GroupCancelCmd(T entity) { - super(entity); - } - - @Override public void executeCmd() { - if (checkTask()) { - tempTask.cancelSubTask(childUrl); - } - } -} diff --git a/Aria/src/main/java/com/arialyy/aria/core/command/group/GroupCmdFactory.java b/Aria/src/main/java/com/arialyy/aria/core/command/group/GroupCmdFactory.java index c931eb5a..5fcb0826 100644 --- a/Aria/src/main/java/com/arialyy/aria/core/command/group/GroupCmdFactory.java +++ b/Aria/src/main/java/com/arialyy/aria/core/command/group/GroupCmdFactory.java @@ -31,10 +31,6 @@ public class GroupCmdFactory { * 停止子任务 */ public static final int SUB_TASK_STOP = 0xa2; - /** - * 取消子任务 - */ - public static final int SUB_TASK_CANCEL = 0xa3; private static volatile GroupCmdFactory INSTANCE = null; @@ -54,7 +50,7 @@ public class GroupCmdFactory { /** * @param target 创建任务的对象 * @param wrapper 参数信息 - * @param type 命令类型{@link #SUB_TASK_START}、{@link #SUB_TASK_STOP}、{@link #SUB_TASK_CANCEL} + * @param type 命令类型{@link #SUB_TASK_START}、{@link #SUB_TASK_STOP} * @param childUrl 需要控制的子任务url */ public AbsGroupCmd createCmd(String target, AbsGroupTaskWrapper wrapper, int type, @@ -67,8 +63,6 @@ public class GroupCmdFactory { case SUB_TASK_STOP: cmd = new GroupStopCmd<>(wrapper); break; - case SUB_TASK_CANCEL: - cmd = new GroupCancelCmd<>(wrapper); } if (cmd != null) { cmd.childUrl = childUrl; diff --git a/Aria/src/main/java/com/arialyy/aria/core/common/IUtil.java b/Aria/src/main/java/com/arialyy/aria/core/common/IUtil.java index f7354574..467c62ae 100644 --- a/Aria/src/main/java/com/arialyy/aria/core/common/IUtil.java +++ b/Aria/src/main/java/com/arialyy/aria/core/common/IUtil.java @@ -16,12 +16,24 @@ package com.arialyy.aria.core.common; +import com.arialyy.aria.core.download.DownloadEntity; +import com.arialyy.aria.core.download.DownloadGroupEntity; +import com.arialyy.aria.core.upload.UploadEntity; + /** * Created by lyy on 2016/10/31. * 任务功能接口 */ public interface IUtil { + /** + * 获取任务标志 + * + * @return {@link DownloadEntity#getKey()}、{@link DownloadGroupEntity#getKey()}、{@link + * UploadEntity#getKey()} + */ + String getKey(); + /** * 获取文件大小 */ diff --git a/Aria/src/main/java/com/arialyy/aria/core/download/DownloadGroupTarget.java b/Aria/src/main/java/com/arialyy/aria/core/download/DownloadGroupTarget.java index f28f595d..05d21afa 100644 --- a/Aria/src/main/java/com/arialyy/aria/core/download/DownloadGroupTarget.java +++ b/Aria/src/main/java/com/arialyy/aria/core/download/DownloadGroupTarget.java @@ -113,7 +113,7 @@ public class DownloadGroupTarget extends BaseGroupTarget im * 任务组总任务大小，任务组是一个抽象的概念，没有真实的数据实体，任务组的大小是Aria动态获取子任务大小相加而得到的， * 如果你知道当前任务组总大小，你也可以调用该方法给任务组设置大小 * - * 为了更好的用户体验，建议直接设置任务组文件大小 + * 为了更好的用户体验，组合任务必须设置文件大小 * * @param fileSize 任务组总大小 */ @@ -185,6 +185,11 @@ public class DownloadGroupTarget extends BaseGroupTarget im return false; } + if (mTaskWrapper.getEntity().getFileSize() == 0) { + ALog.e(TAG, "组合任务必须设置文件文件大小"); + return false; + } + if (mTaskWrapper.asHttp().getRequestEnum() == RequestEnum.POST) { for (DTaskWrapper subTask : mTaskWrapper.getSubTaskWrapper()) { subTask.asHttp().setRequestEnum(RequestEnum.POST); diff --git a/Aria/src/main/java/com/arialyy/aria/core/download/downloader/Downloader.java b/Aria/src/main/java/com/arialyy/aria/core/download/downloader/Downloader.java index 55ebfd95..51c3bd5b 100644 --- a/Aria/src/main/java/com/arialyy/aria/core/download/downloader/Downloader.java +++ b/Aria/src/main/java/com/arialyy/aria/core/download/downloader/Downloader.java @@ -15,7 +15,6 @@ */ package com.arialyy.aria.core.download.downloader; -import com.arialyy.aria.core.AriaManager; import com.arialyy.aria.core.common.AbsFileer; import com.arialyy.aria.core.common.AbsThreadTask; import com.arialyy.aria.core.common.SubThreadConfig; @@ -35,23 +34,22 @@ import java.io.IOException; public class Downloader extends AbsFileer { private String TAG = "Downloader"; - public Downloader(IDownloadListener listener, DTaskWrapper taskEntity) { - super(listener, taskEntity); + public Downloader(IDownloadListener listener, DTaskWrapper taskWrapper) { + super(listener, taskWrapper); mTempFile = new File(mEntity.getDownloadPath()); - AriaManager manager = AriaManager.getInstance(AriaManager.APP); - setUpdateInterval(manager.getDownloadConfig().getUpdateInterval()); + setUpdateInterval(taskWrapper.getConfig().getUpdateInterval()); } + /** + * 小于1m的文件或是任务组的子任务、线程数都是1 + */ @Override protected int setNewTaskThreadNum() { - int threadNum = AriaManager.getInstance(mContext).getDownloadConfig().getThreadNum(); - return - // 小于1m的文件或是任务组的子任务、使用虚拟文件，线程数都是1 - mEntity.getFileSize() <= SUB_LEN - || mTaskWrapper.getRequestType() == AbsTaskWrapper.D_FTP_DIR - || mTaskWrapper.getRequestType() == AbsTaskWrapper.DG_HTTP - || threadNum == 1 - ? 1 - : threadNum; + int threadNum = mTaskWrapper.getConfig().getThreadNum(); + return mEntity.getFileSize() <= SUB_LEN + || mTaskWrapper.isGroupTask() + || threadNum == 1 + ? 1 + : threadNum; } @Override protected boolean handleNewTask() { diff --git a/Aria/src/main/java/com/arialyy/aria/core/download/downloader/SimpleDownloadUtil.java b/Aria/src/main/java/com/arialyy/aria/core/download/downloader/SimpleDownloadUtil.java index ca031345..d5a35dd6 100644 --- a/Aria/src/main/java/com/arialyy/aria/core/download/downloader/SimpleDownloadUtil.java +++ b/Aria/src/main/java/com/arialyy/aria/core/download/downloader/SimpleDownloadUtil.java @@ -33,13 +33,17 @@ public class SimpleDownloadUtil implements IUtil, Runnable { private String TAG = "SimpleDownloadUtil"; private IDownloadListener mListener; private Downloader mDownloader; - private DTaskWrapper mTaskEntity; + private DTaskWrapper mTaskWrapper; private boolean isStop = false, isCancel = false; - public SimpleDownloadUtil(DTaskWrapper entity, IDownloadListener downloadListener) { - mTaskEntity = entity; + public SimpleDownloadUtil(DTaskWrapper wrapper, IDownloadListener downloadListener) { + mTaskWrapper = wrapper; mListener = downloadListener; - mDownloader = new Downloader(downloadListener, entity); + mDownloader = new Downloader(downloadListener, wrapper); + } + + @Override public String getKey() { + return mTaskWrapper.getKey(); } @Override public long getFileSize() { @@ -103,10 +107,10 @@ public class SimpleDownloadUtil implements IUtil, Runnable { if (isStop || isCancel) { return; } - if (mTaskEntity.getEntity().getFileSize() <= 1 - || mTaskEntity.isRefreshInfo() - || mTaskEntity.getRequestType() == AbsTaskWrapper.D_FTP - || mTaskEntity.getState() == IEntity.STATE_FAIL) { + if (mTaskWrapper.getEntity().getFileSize() <= 1 + || mTaskWrapper.isRefreshInfo() + || mTaskWrapper.getRequestType() == AbsTaskWrapper.D_FTP + || mTaskWrapper.getState() == IEntity.STATE_FAIL) { new Thread(createInfoThread()).start(); } else { mDownloader.start(); @@ -117,9 +121,9 @@ public class SimpleDownloadUtil implements IUtil, Runnable { * 通过链接类型创建不同的获取文件信息的线程 */ private Runnable createInfoThread() { - switch (mTaskEntity.getRequestType()) { + switch (mTaskWrapper.getRequestType()) { case AbsTaskWrapper.D_FTP: - return new FtpFileInfoThread(mTaskEntity, new OnFileInfoCallback() { + return new FtpFileInfoThread(mTaskWrapper, new OnFileInfoCallback() { @Override public void onComplete(String url, CompleteInfo info) { mDownloader.start(); } @@ -130,7 +134,7 @@ public class SimpleDownloadUtil implements IUtil, Runnable { } }); case AbsTaskWrapper.D_HTTP: - return new HttpFileInfoThread(mTaskEntity, new OnFileInfoCallback() { + return new HttpFileInfoThread(mTaskWrapper, new OnFileInfoCallback() { @Override public void onComplete(String url, CompleteInfo info) { mDownloader.start(); } diff --git a/Aria/src/main/java/com/arialyy/aria/core/download/group/AbsGroupUtil.java b/Aria/src/main/java/com/arialyy/aria/core/download/group/AbsGroupUtil.java index 43d3fb95..dc90ee69 100644 --- a/Aria/src/main/java/com/arialyy/aria/core/download/group/AbsGroupUtil.java +++ b/Aria/src/main/java/com/arialyy/aria/core/download/group/AbsGroupUtil.java @@ -15,23 +15,16 @@ */ package com.arialyy.aria.core.download.group; -import com.arialyy.aria.core.AriaManager; +import android.os.Handler; import com.arialyy.aria.core.common.IUtil; +import com.arialyy.aria.core.config.Configuration; import com.arialyy.aria.core.download.DTaskWrapper; -import com.arialyy.aria.core.download.DownloadEntity; import com.arialyy.aria.core.download.DGTaskWrapper; -import com.arialyy.aria.core.download.downloader.Downloader; -import com.arialyy.aria.core.inf.IDownloadListener; import com.arialyy.aria.core.inf.IEntity; -import com.arialyy.aria.exception.BaseException; -import com.arialyy.aria.exception.TaskException; import com.arialyy.aria.util.ALog; -import com.arialyy.aria.util.CommonUtil; -import com.arialyy.aria.util.NetUtils; -import java.io.File; import java.util.Map; import java.util.Set; -import java.util.concurrent.ConcurrentHashMap; +import java.util.WeakHashMap; import java.util.concurrent.ScheduledThreadPoolExecutor; import java.util.concurrent.TimeUnit; @@ -40,7 +33,6 @@ import java.util.concurrent.TimeUnit; * 任务组核心逻辑 */ public abstract class AbsGroupUtil implements IUtil, Runnable { - private static final Object LOCK = new Object(); private final String TAG = "AbsGroupUtil"; /** * FTP文件夹 @@ -51,53 +43,46 @@ public abstract class AbsGroupUtil implements IUtil, Runnable { */ int HTTP_GROUP = 0xa2; - /** - * 任务组所有任务总长度 - */ - long mTotalLen = 0; - long mCurrentLocation = 0; + private long mCurrentLocation = 0; protected IDownloadGroupListener mListener; - DGTaskWrapper mGTWrapper; - private boolean isRunning = false; private ScheduledThreadPoolExecutor mTimer; - /** - * 保存所有没有下载完成的任务，key为下载地址 - */ - Map mExeMap = new ConcurrentHashMap<>(); - - /** - * 下载失败的映射表，key为下载地址 - */ - Map mFailMap = new ConcurrentHashMap<>(); - - /** - * 该任务组对应的所有任务 - */ - private Map mTasksMap = new ConcurrentHashMap<>(); + private long mUpdateInterval; + private boolean isStop = false, isCancel = false; + private Handler mScheduler; + private SimpleSubQueue mSubQueue = SimpleSubQueue.newInstance(); + private Map mExeLoader = new WeakHashMap<>(); + private Map mCache = new WeakHashMap<>(); + DGTaskWrapper mGTWrapper; + GroupRunState mState; - /** - * 下载器映射表，key为下载地址 - */ - private Map mDownloaderMap = new ConcurrentHashMap<>(); + AbsGroupUtil(IDownloadGroupListener listener, DGTaskWrapper groupWrapper) { + mListener = listener; + mGTWrapper = groupWrapper; + mUpdateInterval = Configuration.getInstance().downloadCfg.getUpdateInterval(); + mState = new GroupRunState(groupWrapper.getKey(), mListener, + groupWrapper.getSubTaskWrapper().size(), mSubQueue); + mScheduler = new Handler(SimpleSchedulers.newInstance(mState)); + initState(); + } /** - * 是否需要读取文件长度，{@code true}需要 + * 初始化组合任务状态 */ - boolean isNeedLoadFileSize = true; - //已经完成的任务数 - int mCompleteNum = 0; - //停止的任务数 - private int mStopNum = 0; - //任务组大小 - int mGroupSize = 0; - private long mUpdateInterval; - private boolean isStop = false, isCancel = false; + private void initState() { + for (DTaskWrapper wrapper : mGTWrapper.getSubTaskWrapper()) { + if (wrapper.getEntity().getState() == IEntity.STATE_COMPLETE) { + mState.updateCompleteNum(); + mCurrentLocation += wrapper.getEntity().getFileSize(); + } else { + mCache.put(wrapper.getKey(), wrapper); + mCurrentLocation += wrapper.getEntity().getCurrentProgress(); + } + } + mState.updateProgress(mCurrentLocation); + } - AbsGroupUtil(IDownloadGroupListener listener, DGTaskWrapper groupEntity) { - mListener = listener; - mGTWrapper = groupEntity; - mUpdateInterval = - AriaManager.getInstance(AriaManager.APP).getDownloadConfig().getUpdateInterval(); + @Override public String getKey() { + return mGTWrapper.getKey(); } /** @@ -107,16 +92,6 @@ public abstract class AbsGroupUtil implements IUtil, Runnable { */ abstract int getTaskType(); - /** - * 更新任务组文件大小 - */ - void updateFileSize() { - if (isNeedLoadFileSize) { - mGTWrapper.getEntity().setFileSize(mTotalLen); - mGTWrapper.getEntity().update(); - } - } - /** * 启动子任务下载 * @@ -124,13 +99,12 @@ public abstract class AbsGroupUtil implements IUtil, Runnable { */ public void startSubTask(String url) { if (!checkSubTask(url, "开始")) return; - if (!isRunning) { + if (!mState.isRunning) { startTimer(); } - Downloader d = getDownloader(url, false); + SubDownloadLoader d = getDownloader(url); if (d != null && !d.isRunning()) { - d.setNewTask(false); - d.start(); + mSubQueue.startTask(d); } } @@ -141,36 +115,9 @@ public abstract class AbsGroupUtil implements IUtil, Runnable { */ public void stopSubTask(String url) { if (!checkSubTask(url, "停止")) return; - Downloader d = getDownloader(url, false); + SubDownloadLoader d = getDownloader(url); if (d != null && d.isRunning()) { - d.stop(); - } - } - - /** - * 删除子任务 - * - * @param url 子任务下载地址 - */ - public void cancelSubTask(String url) { - Set urls = mTasksMap.keySet(); - if (!urls.isEmpty() && urls.contains(url)) { - DTaskWrapper det = mTasksMap.get(url); - if (det != null) { - mTotalLen -= det.getEntity().getFileSize(); - mCurrentLocation -= det.getEntity().getCurrentProgress(); - mExeMap.remove(det.getKey()); - mFailMap.remove(det.getKey()); - mGroupSize--; - if (mGroupSize == 0) { - closeTimer(); - mListener.onCancel(); - } - } - } - Downloader d = getDownloader(url, false); - if (d != null) { - d.cancel(); + mSubQueue.stopTask(d); } } @@ -182,9 +129,9 @@ public abstract class AbsGroupUtil implements IUtil, Runnable { * @return {@code true} 任务可以下载 */ private boolean checkSubTask(String url, String type) { - DTaskWrapper entity = mTasksMap.get(url); - if (entity != null) { - if (entity.getState() == IEntity.STATE_COMPLETE) { + DTaskWrapper wrapper = mCache.get(url); + if (wrapper != null) { + if (wrapper.getState() == IEntity.STATE_COMPLETE) { ALog.w(TAG, "任务【" + url + "】已完成，" + type + "失败"); return false; } @@ -199,18 +146,17 @@ public abstract class AbsGroupUtil implements IUtil, Runnable { * 通过地址获取下载器 * * @param url 子任务下载地址 - * @param start 是否启动任务 */ - private Downloader getDownloader(String url, boolean start) { - Downloader d = mDownloaderMap.get(url); + private SubDownloadLoader getDownloader(String url) { + SubDownloadLoader d = mExeLoader.get(url); if (d == null) { - return createChildDownload(mTasksMap.get(url), start); + return createSubLoader(mCache.get(url)); } return d; } @Override public long getFileSize() { - return mTotalLen; + return mGTWrapper.getEntity().getFileSize(); } @Override public long getCurrentLocation() { @@ -218,21 +164,22 @@ public abstract class AbsGroupUtil implements IUtil, Runnable { } @Override public boolean isRunning() { - return isRunning; + return mState.isRunning; } @Override public void cancel() { isCancel = true; closeTimer(); onCancel(); - Set keys = mDownloaderMap.keySet(); + Set keys = mExeLoader.keySet(); + mSubQueue.clear(); + for (String key : keys) { - Downloader dt = mDownloaderMap.get(key); - if (dt != null) { - dt.cancel(); + SubDownloadLoader loader = mExeLoader.get(key); + if (loader != null && loader.isRunning()) { + loader.cancel(); } } - clearState(); mListener.onCancel(); } @@ -245,14 +192,15 @@ public abstract class AbsGroupUtil implements IUtil, Runnable { closeTimer(); onStop(); - Set keys = mDownloaderMap.keySet(); + Set keys = mExeLoader.keySet(); + mSubQueue.clear(); + for (String key : keys) { - Downloader dt = mDownloaderMap.get(key); - if (dt != null) { - dt.stop(); + SubDownloadLoader loader = mExeLoader.get(key); + if (loader != null && loader.isRunning()) { + mSubQueue.stopTask(loader); } } - clearState(); mListener.onStop(mCurrentLocation); } @@ -260,32 +208,6 @@ public abstract class AbsGroupUtil implements IUtil, Runnable { } - /** - * 预处理操作 - * 而FTP文件夹的，需要获取完成所有子任务信息才算预处理完成 - */ - protected void onPre() { - mListener.onPre(); - isRunning = true; - mGroupSize = mGTWrapper.getSubTaskWrapper().size(); - mTotalLen = mGTWrapper.getEntity().getFileSize(); - isNeedLoadFileSize = mTotalLen <= 10; - for (DTaskWrapper te : mGTWrapper.getSubTaskWrapper()) { - File file = new File(te.getKey()); - if (te.getState() == IEntity.STATE_COMPLETE && file.exists()) { - mCompleteNum++; - mCurrentLocation += te.getEntity().getFileSize(); - } else { - mExeMap.put(te.getKey(), te); - mCurrentLocation += file.exists() ? te.getEntity().getCurrentProgress() : 0; - } - if (isNeedLoadFileSize) { - mTotalLen += te.getEntity().getFileSize(); - } - mTasksMap.put(te.getGroupHash(), te); - } - } - @Override public void start() { new Thread(this).start(); } @@ -295,8 +217,8 @@ public abstract class AbsGroupUtil implements IUtil, Runnable { closeTimer(); return; } - clearState(); onStart(); + startRunningFlow(); } protected void onStart() { @@ -308,20 +230,15 @@ public abstract class AbsGroupUtil implements IUtil, Runnable { } @Override public void setMaxSpeed(int speed) { - Set keys = mDownloaderMap.keySet(); + Set keys = mSubQueue.getExec().keySet(); for (String key : keys) { - Downloader dt = mDownloaderMap.get(key); + SubDownloadLoader dt = mSubQueue.getExec().get(key); if (dt != null) { dt.setMaxSpeed(speed); } } } - private void clearState() { - mDownloaderMap.clear(); - mFailMap.clear(); - } - synchronized void closeTimer() { if (mTimer != null && !mTimer.isShutdown()) { mTimer.shutdown(); @@ -331,9 +248,9 @@ public abstract class AbsGroupUtil implements IUtil, Runnable { /** * 开始进度流程 */ - void startRunningFlow() { + private void startRunningFlow() { closeTimer(); - mListener.onPostPre(mTotalLen); + mListener.onPostPre(mGTWrapper.getEntity().getFileSize()); if (mCurrentLocation > 0) { mListener.onResume(mCurrentLocation); } else { @@ -343,11 +260,11 @@ public abstract class AbsGroupUtil implements IUtil, Runnable { } private synchronized void startTimer() { - isRunning = true; + mState.isRunning = true; mTimer = new ScheduledThreadPoolExecutor(1); mTimer.scheduleWithFixedDelay(new Runnable() { @Override public void run() { - if (!isRunning) { + if (!mState.isRunning) { closeTimer(); } else if (mCurrentLocation >= 0) { long t = 0; @@ -359,6 +276,7 @@ public abstract class AbsGroupUtil implements IUtil, Runnable { } } mCurrentLocation = t; + mState.updateProgress(mCurrentLocation); mListener.onProgress(t); } } @@ -366,201 +284,12 @@ public abstract class AbsGroupUtil implements IUtil, Runnable { } /** - * 创建子任务下载器，默认创建完成自动启动 + * 创建并启动子任务下载器 */ - void createChildDownload(DTaskWrapper taskEntity) { - createChildDownload(taskEntity, true); - } - - /** - * 创建子任务下载器，启动子任务下载器 - * - * @param start 是否启动下载 - */ - private Downloader createChildDownload(DTaskWrapper taskEntity, boolean start) { - ChildDownloadListener listener = new ChildDownloadListener(taskEntity); - Downloader dt = new Downloader(listener, taskEntity); - mDownloaderMap.put(taskEntity.getEntity().getUrl(), dt); - if (start) { - dt.start(); - } - return dt; - } - - /** - * 子任务事件监听 - */ - private class ChildDownloadListener implements IDownloadListener { - private DTaskWrapper subTaskWrapper; - private DownloadEntity subEntity; - private int RUN_SAVE_INTERVAL = 5 * 1000; //5s保存一次下载中的进度 - private long lastSaveTime; - private long lastLen; - private ScheduledThreadPoolExecutor timer; - private boolean isNotNetRetry; - - ChildDownloadListener(DTaskWrapper entity) { - subTaskWrapper = entity; - subEntity = subTaskWrapper.getEntity(); - subEntity.setFailNum(0); - lastLen = subEntity.getCurrentProgress(); - lastSaveTime = System.currentTimeMillis(); - isNotNetRetry = AriaManager.getInstance(AriaManager.APP).getAppConfig().isNotNetRetry(); - } - - @Override public void onPre() { - saveData(IEntity.STATE_PRE, -1); - } - - @Override public void onPostPre(long fileSize) { - subEntity.setFileSize(fileSize); - subEntity.setConvertFileSize(CommonUtil.formatFileSize(fileSize)); - saveData(IEntity.STATE_POST_PRE, -1); - mListener.onSubPre(subEntity); - } - - @Override public void onResume(long resumeLocation) { - saveData(IEntity.STATE_POST_PRE, IEntity.STATE_RUNNING); - lastLen = resumeLocation; - mListener.onSubStart(subEntity); - } - - @Override public void onStart(long startLocation) { - saveData(IEntity.STATE_POST_PRE, IEntity.STATE_RUNNING); - lastLen = startLocation; - mListener.onSubStart(subEntity); - } - - @Override public void onProgress(long currentLocation) { - long speed = currentLocation - lastLen; - //mCurrentLocation += speed; - subEntity.setCurrentProgress(currentLocation); - handleSpeed(speed); - mListener.onSubRunning(subEntity); - if (System.currentTimeMillis() - lastSaveTime >= RUN_SAVE_INTERVAL) { - saveData(IEntity.STATE_RUNNING, currentLocation); - lastSaveTime = System.currentTimeMillis(); - } - lastLen = currentLocation; - } - - @Override public void onStop(long stopLocation) { - saveData(IEntity.STATE_STOP, stopLocation); - handleSpeed(0); - mListener.onSubStop(subEntity); - synchronized (AbsGroupUtil.LOCK) { - mStopNum++; - if (mStopNum + mCompleteNum + mFailMap.size() == mGroupSize && !isStop) { - closeTimer(); - mListener.onStop(mCurrentLocation); - } - } - } - - @Override public void onCancel() { - saveData(IEntity.STATE_CANCEL, -1); - handleSpeed(0); - mListener.onSubCancel(subEntity); - } - - @Override public void onComplete() { - subEntity.setComplete(true); - saveData(IEntity.STATE_COMPLETE, subEntity.getFileSize()); - handleSpeed(0); - mListener.onSubComplete(subEntity); - synchronized (AbsGroupUtil.LOCK) { - mCompleteNum++; - //如果子任务完成的数量和总任务数一致，表示任务组任务已经完成 - if (mCompleteNum == mGroupSize) { - closeTimer(); - mListener.onComplete(); - } else if (mFailMap.size() > 0 - && mStopNum + mCompleteNum + mFailMap.size() >= mGroupSize - && !isStop) { - //如果子任务完成数量加上失败的数量和总任务数一致，则任务组停止下载 - closeTimer(); - mListener.onStop(mCurrentLocation); - } - } - } - - @Override public void onFail(boolean needRetry, BaseException e) { - subEntity.setFailNum(subEntity.getFailNum() + 1); - saveData(IEntity.STATE_FAIL, lastLen); - handleSpeed(0); - reTry(needRetry); - } - - /** - * 重试下载，只有全部都下载失败才会执行任务组的整体重试，否则只会执行单个子任务的重试 - */ - private void reTry(boolean needRetry) { - synchronized (AbsGroupUtil.LOCK) { - Downloader dt = mDownloaderMap.get(subEntity.getUrl()); - if (!isCancel && !isStop && dt != null - && !dt.isBreak() - && needRetry - && subEntity.getFailNum() < 3 - && (NetUtils.isConnected(AriaManager.APP) || isNotNetRetry)) { - ALog.d(TAG, "downloader retry"); - reStartTask(dt); - } else { - mFailMap.put(subTaskWrapper.getKey(), subTaskWrapper); - mListener.onSubFail(subEntity, new TaskException(TAG, - String.format("任务组子任务【%s】下载失败，下载地址【%s】", subEntity.getFileName(), - subEntity.getUrl()))); - if (mFailMap.size() == mExeMap.size() || mFailMap.size() + mCompleteNum == mGroupSize) { - closeTimer(); - } - if (mFailMap.size() == mGroupSize) { - mListener.onFail(true, new TaskException(TAG, - String.format("任务组【%s】下载失败", mGTWrapper.getEntity().getGroupHash()))); - } else if (mFailMap.size() + mCompleteNum >= mExeMap.size()) { - mListener.onStop(mCurrentLocation); - } - } - } - } - - private void reStartTask(final Downloader dt) { - if (timer == null || timer.isShutdown()) { - timer = new ScheduledThreadPoolExecutor(1); - } - timer.schedule(new Runnable() { - @Override public void run() { - if (dt != null) { - dt.retryTask(); - } - } - }, 5, TimeUnit.SECONDS); - } - - private void handleSpeed(long speed) { - subEntity.setSpeed(speed); - subEntity.setConvertSpeed( - speed <= 0 ? "" : String.format("%s/s", CommonUtil.formatFileSize(speed))); - subEntity.setPercent((int) (subEntity.getFileSize() <= 0 ? 0 - : subEntity.getCurrentProgress() * 100 / subEntity.getFileSize())); - } - - private void saveData(int state, long location) { - subTaskWrapper.setState(state); - subEntity.setState(state); - subEntity.setComplete(state == IEntity.STATE_COMPLETE); - if (state == IEntity.STATE_CANCEL) { - subEntity.deleteData(); - } else if (state == IEntity.STATE_STOP) { - subEntity.setStopTime(System.currentTimeMillis()); - } else if (subEntity.isComplete()) { - subEntity.setCompleteTime(System.currentTimeMillis()); - subEntity.setCurrentProgress(subEntity.getFileSize()); - } else if (location > 0) { - subEntity.setCurrentProgress(location); - } - } - - @Override public void supportBreakpoint(boolean support) { - - } + SubDownloadLoader createSubLoader(DTaskWrapper taskWrapper) { + SubDownloadLoader loader = new SubDownloadLoader(mScheduler, taskWrapper); + mExeLoader.put(loader.getKey(), loader); + mSubQueue.startTask(loader); + return loader; } } diff --git a/Aria/src/main/java/com/arialyy/aria/core/download/group/ChildDownloadListener.java b/Aria/src/main/java/com/arialyy/aria/core/download/group/ChildDownloadListener.java new file mode 100644 index 00000000..dcccf5be --- /dev/null +++ b/Aria/src/main/java/com/arialyy/aria/core/download/group/ChildDownloadListener.java @@ -0,0 +1,146 @@ +/* + * Copyright (C) 2016 AriaLyy(https://github.com/AriaLyy/Aria) + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package com.arialyy.aria.core.download.group; + +import android.os.Handler; +import com.arialyy.aria.core.download.DownloadEntity; +import com.arialyy.aria.core.inf.IDownloadListener; +import com.arialyy.aria.core.inf.IEntity; +import com.arialyy.aria.core.scheduler.ISchedulers; +import com.arialyy.aria.exception.BaseException; +import com.arialyy.aria.util.CommonUtil; + +/** + * 子任务事件监听 + */ +class ChildDownloadListener implements IDownloadListener { + private DownloadEntity subEntity; + private int RUN_SAVE_INTERVAL = 5 * 1000; //5s保存一次下载中的进度 + private long lastSaveTime; + private long lastLen; + private Handler schedulers; + private SubDownloadLoader loader; + + ChildDownloadListener(Handler schedulers, SubDownloadLoader loader) { + this.loader = loader; + this.schedulers = schedulers; + subEntity = loader.getEntity(); + subEntity.setFailNum(0); + lastLen = subEntity.getCurrentProgress(); + lastSaveTime = System.currentTimeMillis(); + } + + @Override public void supportBreakpoint(boolean support) { + + } + + @Override public void onPre() { + saveData(IEntity.STATE_PRE, -1); + } + + @Override public void onPostPre(long fileSize) { + subEntity.setFileSize(fileSize); + subEntity.setConvertFileSize(CommonUtil.formatFileSize(fileSize)); + saveData(IEntity.STATE_POST_PRE, -1); + sendToTarget(ISchedulers.POST_PRE, loader); + } + + @Override public void onResume(long resumeLocation) { + lastLen = resumeLocation; + saveData(IEntity.STATE_POST_PRE, IEntity.STATE_RUNNING); + sendToTarget(ISchedulers.START, loader); + } + + @Override public void onStart(long startLocation) { + lastLen = startLocation; + saveData(IEntity.STATE_POST_PRE, IEntity.STATE_RUNNING); + sendToTarget(ISchedulers.START, loader); + } + + @Override public void onProgress(long currentLocation) { + long diff = currentLocation - lastLen; + //mCurrentLocation += speed; + subEntity.setCurrentProgress(currentLocation); + handleSpeed(diff); + sendToTarget(ISchedulers.RUNNING, loader); + if (System.currentTimeMillis() - lastSaveTime >= RUN_SAVE_INTERVAL) { + saveData(IEntity.STATE_RUNNING, currentLocation); + lastSaveTime = System.currentTimeMillis(); + } + lastLen = currentLocation; + } + + @Override public void onStop(long stopLocation) { + handleSpeed(0); + saveData(IEntity.STATE_STOP, stopLocation); + sendToTarget(ISchedulers.STOP, loader); + } + + /** + * 组合任务子任务不允许删除 + */ + @Deprecated + @Override public void onCancel() { + + } + + @Override public void onComplete() { + subEntity.setComplete(true); + saveData(IEntity.STATE_COMPLETE, subEntity.getFileSize()); + handleSpeed(0); + sendToTarget(ISchedulers.COMPLETE, loader); + } + + @Override public void onFail(boolean needRetry, BaseException e) { + subEntity.setFailNum(subEntity.getFailNum() + 1); + saveData(IEntity.STATE_FAIL, lastLen); + handleSpeed(0); + sendToTarget(ISchedulers.FAIL, loader); + } + + private void handleSpeed(long speed) { + subEntity.setSpeed(speed); + subEntity.setConvertSpeed( + speed <= 0 ? "" : String.format("%s/s", CommonUtil.formatFileSize(speed))); + subEntity.setPercent((int) (subEntity.getFileSize() <= 0 ? 0 + : subEntity.getCurrentProgress() * 100 / subEntity.getFileSize())); + } + + private void saveData(int state, long location) { + loader.getWrapper().setState(state); + subEntity.setState(state); + subEntity.setComplete(state == IEntity.STATE_COMPLETE); + if (state == IEntity.STATE_CANCEL) { + subEntity.deleteData(); + } else if (state == IEntity.STATE_STOP) { + subEntity.setStopTime(System.currentTimeMillis()); + } else if (subEntity.isComplete()) { + subEntity.setCompleteTime(System.currentTimeMillis()); + subEntity.setCurrentProgress(subEntity.getFileSize()); + } else if (location > 0) { + subEntity.setCurrentProgress(location); + } + } + + /** + * 发送状态到子任务调度器{@link SimpleSchedulers}，让调度器处理任务调度 + * + * @param state {@link ISchedulers} + */ + private void sendToTarget(int state, SubDownloadLoader util) { + schedulers.obtainMessage(state, util).sendToTarget(); + } +} \ No newline at end of file diff --git a/Aria/src/main/java/com/arialyy/aria/core/download/group/DownloadGroupUtil.java b/Aria/src/main/java/com/arialyy/aria/core/download/group/DownloadGroupUtil.java index 82a52f48..09a48055 100644 --- a/Aria/src/main/java/com/arialyy/aria/core/download/group/DownloadGroupUtil.java +++ b/Aria/src/main/java/com/arialyy/aria/core/download/group/DownloadGroupUtil.java @@ -15,48 +15,19 @@ */ package com.arialyy.aria.core.download.group; -import android.util.SparseArray; -import com.arialyy.aria.core.common.CompleteInfo; import com.arialyy.aria.core.common.IUtil; -import com.arialyy.aria.core.common.OnFileInfoCallback; import com.arialyy.aria.core.download.DGTaskWrapper; import com.arialyy.aria.core.download.DTaskWrapper; -import com.arialyy.aria.core.download.downloader.HttpFileInfoThread; import com.arialyy.aria.core.inf.IEntity; -import com.arialyy.aria.exception.BaseException; -import com.arialyy.aria.exception.TaskException; -import com.arialyy.aria.util.ALog; -import java.util.Set; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; /** * Created by AriaL on 2017/6/30. * 任务组下载工具 */ public class DownloadGroupUtil extends AbsGroupUtil implements IUtil { - private final String TAG = "DownloadGroupUtil"; - private ExecutorService mInfoPool; - /** - * 初始化完成的任务数 - */ - private int mInitCompleteNum; - /** - * 初始化失败的任务数 - */ - private int mInitFailNum; - private boolean isStop = false; - private boolean isStart = false; - private int mExeNum; - - /** - * 文件信息回调组 - */ - private SparseArray mFileInfoCallbacks = new SparseArray<>(); public DownloadGroupUtil(IDownloadGroupListener listener, DGTaskWrapper taskWrapper) { super(listener, taskWrapper); - mInfoPool = Executors.newCachedThreadPool(); } @Override int getTaskType() { @@ -65,121 +36,22 @@ public class DownloadGroupUtil extends AbsGroupUtil implements IUtil { @Override public void onCancel() { super.onCancel(); - isStop = true; - if (!mInfoPool.isShutdown()) { - mInfoPool.shutdown(); - } } @Override protected void onStop() { super.onStop(); - isStop = true; - if (!mInfoPool.isShutdown()) { - mInfoPool.shutdown(); - } } @Override protected void onStart() { - onPre(); - isStop = false; - if (mCompleteNum == mGroupSize) { + super.onStart(); + if (mState.getCompleteNum() == mState.getSubSize()) { mListener.onComplete(); - return; - } - - if (mExeMap.size() == 0) { - mListener.onFail(false, new TaskException(TAG, - String.format("任务组【%s】无可执行任务", mGTWrapper.getEntity().getGroupHash()))); - return; - } - Set keys = mExeMap.keySet(); - mExeNum = mExeMap.size(); - for (String key : keys) { - DTaskWrapper taskEntity = mExeMap.get(key); - if (taskEntity != null) { - if (taskEntity.getState() != IEntity.STATE_FAIL - && taskEntity.getState() != IEntity.STATE_WAIT) { - mInitCompleteNum++; - createChildDownload(taskEntity); - checkStartFlow(); - } else { - mInfoPool.execute(createFileInfoThread(taskEntity)); + } else { + for (DTaskWrapper wrapper : mGTWrapper.getSubTaskWrapper()) { + if (wrapper.getState() != IEntity.STATE_COMPLETE) { + createSubLoader(wrapper); } } } - if (mCurrentLocation == mTotalLen) { - mListener.onComplete(); - } - } - - /** - * 创建文件信息获取线程 - */ - private HttpFileInfoThread createFileInfoThread(DTaskWrapper taskEntity) { - OnFileInfoCallback callback = mFileInfoCallbacks.get(taskEntity.hashCode()); - - if (callback == null) { - callback = new OnFileInfoCallback() { - int failNum = 0; - - @Override public void onComplete(String url, CompleteInfo info) { - if (isStop) return; - DTaskWrapper te = mExeMap.get(url); - if (te != null) { - if (isNeedLoadFileSize) { - mTotalLen += te.getEntity().getFileSize(); - } - createChildDownload(te); - } - mInitCompleteNum++; - - checkStartFlow(); - } - - @Override public void onFail(String url, BaseException e, boolean needRetry) { - if (isStop) return; - ALog.e(TAG, String.format("任务【%s】初始化失败", url)); - DTaskWrapper te = mExeMap.get(url); - if (te != null) { - mFailMap.put(url, te); - mFileInfoCallbacks.put(te.hashCode(), this); - mExeMap.remove(url); - } - //404链接不重试下载 - //if (failNum < 3 && !errorMsg.contains("错误码：404") && !errorMsg.contains( - // "UnknownHostException")) { - // mInfoPool.execute(createFileInfoThread(te)); - //} else { - // mInitFailNum++; - //} - //failNum++; - mInitFailNum++; - checkStartFlow(); - } - }; - } - return new HttpFileInfoThread(taskEntity, callback); - } - - /** - * 检查能否启动下载流程 - */ - private void checkStartFlow() { - synchronized (DownloadGroupUtil.class) { - if (isStop) { - closeTimer(); - return; - } - if (mInitFailNum == mExeNum) { - closeTimer(); - mListener.onFail(true, new TaskException(TAG, - String.format("任务组【%s】初始化失败", mGTWrapper.getEntity().getGroupHash()))); - } - if (!isStart && mInitCompleteNum + mInitFailNum == mExeNum || !isNeedLoadFileSize) { - startRunningFlow(); - updateFileSize(); - isStart = true; - } - } } } \ No newline at end of file diff --git a/Aria/src/main/java/com/arialyy/aria/core/download/group/FtpDirDownloadUtil.java b/Aria/src/main/java/com/arialyy/aria/core/download/group/FtpDirDownloadUtil.java index bbf1d9dc..e04ce09b 100644 --- a/Aria/src/main/java/com/arialyy/aria/core/download/group/FtpDirDownloadUtil.java +++ b/Aria/src/main/java/com/arialyy/aria/core/download/group/FtpDirDownloadUtil.java @@ -19,8 +19,8 @@ import com.arialyy.aria.core.common.CompleteInfo; import com.arialyy.aria.core.common.OnFileInfoCallback; import com.arialyy.aria.core.download.DGTaskWrapper; import com.arialyy.aria.core.download.DTaskWrapper; +import com.arialyy.aria.core.inf.IEntity; import com.arialyy.aria.exception.BaseException; -import java.util.Set; /** * Created by Aria.Lao on 2017/7/27. ftp文件夹下载工具 @@ -39,23 +39,16 @@ public class FtpDirDownloadUtil extends AbsGroupUtil { @Override protected void onStart() { super.onStart(); if (mGTWrapper.getEntity().getFileSize() > 1) { - onPre(); startDownload(); } else { new FtpDirInfoThread(mGTWrapper, new OnFileInfoCallback() { @Override public void onComplete(String url, CompleteInfo info) { if (info.code >= 200 && info.code < 300) { - onPre(); startDownload(); } } @Override public void onFail(String url, BaseException e, boolean needRetry) { - DTaskWrapper te = mExeMap.get(url); - if (te != null) { - mFailMap.put(url, te); - mExeMap.remove(url); - } mListener.onFail(needRetry, e); } }).start(); @@ -63,23 +56,10 @@ public class FtpDirDownloadUtil extends AbsGroupUtil { } private void startDownload() { - if (mCompleteNum == mGroupSize) { - mListener.onComplete(); - return; - } - int i = 0; - Set keys = mExeMap.keySet(); - for (String key : keys) { - DTaskWrapper taskEntity = mExeMap.get(key); - if (taskEntity != null) { - createChildDownload(taskEntity); - i++; + for (DTaskWrapper wrapper : mGTWrapper.getSubTaskWrapper()) { + if (wrapper.getState() != IEntity.STATE_COMPLETE) { + createSubLoader(wrapper); } } - if (mExeMap.size() == 0) { - mListener.onComplete(); - } else if (i == mExeMap.size()) { - startRunningFlow(); - } } } diff --git a/Aria/src/main/java/com/arialyy/aria/core/download/group/GroupRunState.java b/Aria/src/main/java/com/arialyy/aria/core/download/group/GroupRunState.java new file mode 100644 index 00000000..28131784 --- /dev/null +++ b/Aria/src/main/java/com/arialyy/aria/core/download/group/GroupRunState.java @@ -0,0 +1,169 @@ +/* + * Copyright (C) 2016 AriaLyy(https://github.com/AriaLyy/Aria) + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package com.arialyy.aria.core.download.group; + +import com.arialyy.aria.core.inf.AbsTaskWrapper; +import java.util.HashSet; +import java.util.Set; + +/** + * 组合任务执行中的状态信息 + */ +class GroupRunState { + /** + * 子任务数 + */ + private int mSubSize; + + /** + * 已经完成的任务数 + */ + private int mCompleteNum; + + /** + * 失败的任务数 + */ + private int mFailNum; + + /** + * 停止的任务数 + */ + private int mStopNum; + + /** + * 当前进度 + */ + private long mProgress; + + /** + * 组合任务监听 + */ + IDownloadGroupListener listener; + + /** + * 子任务队列 + */ + SimpleSubQueue queue; + + /** + * 是否在执行 + */ + boolean isRunning = false; + + /** + * 子任务失败、停止记录，用于当子任务失败重新被用户点击开始时，更新{@link #mStopNum}或{@link #mFailNum} + * 保存的数据为：子任务key + */ + private Set mFailTemp = new HashSet<>(), mStopTemp = new HashSet<>(); + + private String mGroupHash; + + GroupRunState(String groupHash, IDownloadGroupListener listener, int subSize, + SimpleSubQueue queue) { + this.listener = listener; + this.queue = queue; + mSubSize = subSize; + mGroupHash = groupHash; + } + + String getGroupHash() { + return mGroupHash; + } + + /** + * 获取组合任务子任务数 + */ + int getSubSize() { + return mSubSize; + } + + /** + * 获取失败的数量 + */ + int getFailNum() { + return mFailNum; + } + + /** + * 获取停止的数量 + */ + int getStopNum() { + return mStopNum; + } + + /** + * 获取完成的数量 + */ + int getCompleteNum() { + return mCompleteNum; + } + + /** + * 获取当前组合任务总进度 + */ + long getProgress() { + return mProgress; + } + + /** + * 更新完成的数量，mCompleteNum + 1 + */ + void updateCompleteNum() { + mCompleteNum++; + } + + /** + * 更新任务进度 + */ + void updateProgress(long newProgress) { + this.mProgress = newProgress; + } + + /** + * 当子任务开始时，更新停止\失败的任务数 + * + * @param key {@link AbsTaskWrapper#getKey()} + */ + void updateCount(String key) { + if (mFailTemp.contains(key)) { + mFailTemp.remove(key); + mFailNum--; + } else if (mStopTemp.contains(key)) { + mStopTemp.remove(key); + mStopNum--; + } + } + + /** + * 统计子任务停止的数量 + * + * @param key {@link AbsTaskWrapper#getKey()} + */ + void countStopNum(String key) { + mStopTemp.add(key); + mStopNum++; + } + + /** + * 统计子任务失败的数量 + * + * @param key {@link AbsTaskWrapper#getKey()} + */ + void countFailNum(String key) { + mFailTemp.add(key); + mFailNum++; + } +} diff --git a/Aria/src/main/java/com/arialyy/aria/core/download/group/ISubQueue.java b/Aria/src/main/java/com/arialyy/aria/core/download/group/ISubQueue.java index d94ffa82..d8196380 100644 --- a/Aria/src/main/java/com/arialyy/aria/core/download/group/ISubQueue.java +++ b/Aria/src/main/java/com/arialyy/aria/core/download/group/ISubQueue.java @@ -16,13 +16,15 @@ package com.arialyy.aria.core.download.group; import com.arialyy.aria.core.common.AbsFileer; +import com.arialyy.aria.core.common.IUtil; +import com.arialyy.aria.core.config.DGroupConfig; /** * 组合任务子任务队列 * * @param {@link AbsFileer}下载器 */ -interface ISubQueue { +interface ISubQueue { /** * 添加任务 @@ -32,6 +34,9 @@ interface ISubQueue { /** * 开始任务 + * 如果执行队列没有达到上限，则启动任务。 + * 如果执行队列已经到达上限，则将任务添加到等待队列总。 + * 队列上限配置{@link DGroupConfig#setSubMaxTaskNum(int)} */ void startTask(Fileer fileer); @@ -61,4 +66,9 @@ interface ISubQueue { * 获取下一个任务 */ Fileer getNextTask(); + + /** + * 清空缓存队列和执行队列 + */ + void clear(); } diff --git a/Aria/src/main/java/com/arialyy/aria/core/download/group/SimpleSchedulers.java b/Aria/src/main/java/com/arialyy/aria/core/download/group/SimpleSchedulers.java index a5aa761a..e3ccde58 100644 --- a/Aria/src/main/java/com/arialyy/aria/core/download/group/SimpleSchedulers.java +++ b/Aria/src/main/java/com/arialyy/aria/core/download/group/SimpleSchedulers.java @@ -16,100 +16,164 @@ package com.arialyy.aria.core.download.group; -import android.os.CountDownTimer; import android.os.Message; import com.arialyy.aria.core.AriaManager; import com.arialyy.aria.core.config.Configuration; -import com.arialyy.aria.core.download.DownloadTask; -import com.arialyy.aria.core.download.downloader.Downloader; import com.arialyy.aria.core.inf.AbsEntity; import com.arialyy.aria.core.scheduler.ISchedulers; +import com.arialyy.aria.exception.TaskException; import com.arialyy.aria.util.ALog; import com.arialyy.aria.util.NetUtils; +import java.util.concurrent.ScheduledThreadPoolExecutor; +import java.util.concurrent.TimeUnit; /** * 组合任务子任务调度器，用于调度任务的开始、停止、失败、完成等情况 * 该调度器生命周期和{@link AbsGroupUtil}生命周期一致 */ -public class SimpleSchedulers implements ISchedulers { +class SimpleSchedulers implements ISchedulers { private static final String TAG = "SimpleSchedulers"; - private SimpleSubQueue mQueue = SimpleSubQueue.newInstance(); - - private SimpleSchedulers() { + private SimpleSubQueue mQueue; + private GroupRunState mGState; + private SimpleSchedulers(GroupRunState state) { + mQueue = state.queue; + mGState = state; } - public static SimpleSchedulers newInstance() { + public static SimpleSchedulers newInstance(GroupRunState state) { - return new SimpleSchedulers(); + return new SimpleSchedulers(state); } @Override public boolean handleMessage(Message msg) { - Downloader loader = (Downloader) msg.obj; + SubDownloadLoader loader = (SubDownloadLoader) msg.obj; switch (msg.what) { - case ADD: - mQueue.addTask(loader); + case RUNNING: + mGState.listener.onSubRunning(loader.getEntity()); + break; + case PRE: + mGState.listener.onSubPre(loader.getEntity()); + mGState.updateCount(loader.getKey()); break; case START: - mQueue.startTask(loader); + mGState.listener.onSubStart(loader.getEntity()); break; case STOP: - mQueue.stopTask(loader); - startNext(); + handleStop(loader); break; case COMPLETE: - mQueue.removeTaskFromExecQ(loader); - startNext(); + handleComplete(loader); + break; case FAIL: + handleFail(loader); break; } return true; } /** - * 如果有等待中的任务，则启动下一任务 + * 处理子任务失败的情况 + * 1、子任务失败次数大于等于配置的重试次数，才能认为子任务停止 + * 2、stopNum + failNum + completeNum == subSize，则认为组合任务停止 + * 3、failNum == subSize，只有全部的子任务都失败了，才能任务组合任务失败 */ - private void startNext() { - Downloader next = mQueue.getNextTask(); - if (next != null) { - mQueue.startTask(next); - } else { - ALog.i(TAG, "没有下一任务"); - } - } - - /** - * 处理失败的任务 - */ - private void handleFail(final Downloader loader) { + private synchronized void handleFail(final SubDownloadLoader loader) { Configuration config = Configuration.getInstance(); - long interval = config.downloadCfg.getReTryInterval(); - int num = config.downloadCfg.getReTryNum(); + long interval = config.dGroupCfg.getSubReTryInterval(); + int num = config.dGroupCfg.getSubReTryNum(); boolean isNotNetRetry = config.appCfg.isNotNetRetry(); final int reTryNum = num; if ((!NetUtils.isConnected(AriaManager.APP) && !isNotNetRetry) || loader.getEntity().getFailNum() > reTryNum) { - startNext(); + mQueue.removeTaskFromExecQ(loader); + mGState.listener.onSubFail(loader.getEntity(), new TaskException(TAG, + String.format("任务组子任务【%s】下载失败，下载地址【%s】", loader.getEntity().getFileName(), + loader.getEntity().getUrl()))); + mGState.countFailNum(loader.getKey()); + if (mGState.getFailNum() == mGState.getSubSize() + || mGState.getStopNum() + mGState.getFailNum() + mGState.getCompleteNum() + == mGState.getSubSize()) { + mGState.isRunning = false; + mGState.listener.onFail(true, new TaskException(TAG, + String.format("任务组【%s】下载失败", mGState.getGroupHash()))); + } else { + startNext(); + } return; } + final ScheduledThreadPoolExecutor timer = new ScheduledThreadPoolExecutor(1); - CountDownTimer timer = new CountDownTimer(interval, 1000) { - @Override public void onTick(long millisUntilFinished) { - - } - - @Override public void onFinish() { + timer.schedule(new Runnable() { + @Override public void run() { AbsEntity entity = loader.getEntity(); if (entity.getFailNum() <= reTryNum) { ALog.d(TAG, String.format("任务【%s】开始重试", loader.getEntity().getFileName())); - loader.retryTask(); + loader.reStart(); } else { startNext(); } } - }; - timer.start(); + }, interval, TimeUnit.MILLISECONDS); + } + + /** + * 处理子任务停止的情况 + * 1、所有的子任务已经停止，则认为组合任务停止 + * 2、completeNum + failNum + stopNum = subSize，则认为组合任务停止 + */ + private synchronized void handleStop(SubDownloadLoader loader) { + mGState.listener.onSubStop(loader.getEntity()); + mGState.countStopNum(loader.getKey()); + if (mGState.getStopNum() == mGState.getSubSize() + || mGState.getStopNum() + mGState.getCompleteNum() + mGState.getFailNum() + == mGState.getSubSize()) { + mGState.isRunning = false; + mGState.listener.onStop(mGState.getProgress()); + } else { + startNext(); + } + } + + /** + * 处理子任务完成的情况，有以下三种情况 + * 1、已经没有缓存的子任务，并且停止的子任务是数{@link GroupRunState#getStopNum()} ()}为0，失败的子任数{@link + * GroupRunState#getFailNum()}为0，则认为组合任务已经完成 + * 2、已经没有缓存的子任务，并且停止的子任务是数{@link GroupRunState#getCompleteNum()}不为0，或者失败的子任数{@link + * GroupRunState#getFailNum()}不为0，则认为组合任务被停止 + * 3、只有有缓存的子任务，则任务组合任务没有完成 + */ + private synchronized void handleComplete(SubDownloadLoader loader) { + ALog.d(TAG, String.format("子任务【%s】完成", loader.getEntity().getFileName())); + mGState.listener.onSubComplete(loader.getEntity()); + mQueue.removeTaskFromExecQ(loader); + mGState.updateCompleteNum(); + ALog.d(TAG, String.format("总任务数：%s，完成的任务数：%s，失败的任务数：%s，停止的任务数：%s", mGState.getSubSize(), + mGState.getCompleteNum(), mGState.getFailNum(), mGState.getStopNum())); + if (mGState.getCompleteNum() == mGState.getSubSize()) { + if (mGState.getStopNum() == 0 && mGState.getFailNum() == 0) { + mGState.listener.onComplete(); + } else { + mGState.listener.onStop(mGState.getProgress()); + } + mGState.isRunning = false; + } else { + startNext(); + } + } + + /** + * 如果有等待中的任务，则启动下一任务 + */ + private void startNext() { + SubDownloadLoader next = mQueue.getNextTask(); + if (next != null) { + ALog.d(TAG, String.format("启动任务：%s", next.getEntity().getFileName())); + mQueue.startTask(next); + } else { + ALog.i(TAG, "没有下一子任务"); + } } } diff --git a/Aria/src/main/java/com/arialyy/aria/core/download/group/SimpleSubQueue.java b/Aria/src/main/java/com/arialyy/aria/core/download/group/SimpleSubQueue.java index d4b083af..2e261867 100644 --- a/Aria/src/main/java/com/arialyy/aria/core/download/group/SimpleSubQueue.java +++ b/Aria/src/main/java/com/arialyy/aria/core/download/group/SimpleSubQueue.java @@ -16,7 +16,6 @@ package com.arialyy.aria.core.download.group; import com.arialyy.aria.core.config.Configuration; -import com.arialyy.aria.core.download.downloader.Downloader; import com.arialyy.aria.util.ALog; import java.util.ArrayList; import java.util.Collection; @@ -29,16 +28,16 @@ import java.util.Set; /** * 组合任务队列，该队列生命周期和{@link AbsGroupUtil}生命周期一致 */ -class SimpleSubQueue implements ISubQueue { +class SimpleSubQueue implements ISubQueue { private static final String TAG = "SimpleSubQueue"; /** * 缓存下载器 */ - private Map mCache = new LinkedHashMap<>(); + private Map mCache = new LinkedHashMap<>(); /** * 执行中的下载器 */ - private Map mExec = new LinkedHashMap<>(); + private Map mExec = new LinkedHashMap<>(); /** * 最大执行任务数 @@ -53,22 +52,26 @@ class SimpleSubQueue implements ISubQueue { return new SimpleSubQueue(); } - @Override public void addTask(Downloader fileer) { + public Map getExec() { + return mExec; + } + + @Override public void addTask(SubDownloadLoader fileer) { mCache.put(fileer.getKey(), fileer); } - @Override public void startTask(Downloader fileer) { + @Override public void startTask(SubDownloadLoader fileer) { if (mExec.size() < mExecSize) { mCache.remove(fileer.getKey()); mExec.put(fileer.getKey(), fileer); fileer.start(); } else { - ALog.d(TAG, String.format("执行队列已满，任务见缓冲到缓存器中，key: %s", fileer.getKey())); + ALog.d(TAG, String.format("执行队列已满，任务进入缓存器中，key: %s", fileer.getKey())); addTask(fileer); } } - @Override public void stopTask(Downloader fileer) { + @Override public void stopTask(SubDownloadLoader fileer) { fileer.stop(); mExec.remove(fileer.getKey()); } @@ -89,7 +92,7 @@ class SimpleSubQueue implements ISubQueue { if (oldSize < num) { // 处理队列变小的情况，该情况下将停止队尾任务，并将这些任务添加到缓存队列中 if (mExec.size() > num) { Set keys = mExec.keySet(); - List caches = new ArrayList<>(); + List caches = new ArrayList<>(); int i = 0; for (String key : keys) { if (i > num) { @@ -97,20 +100,20 @@ class SimpleSubQueue implements ISubQueue { } i++; } - Collection temp = mCache.values(); + Collection temp = mCache.values(); mCache.clear(); ALog.d(TAG, String.format("测试, map size: %s", mCache.size())); - for (Downloader cache : caches) { + for (SubDownloadLoader cache : caches) { addTask(cache); } - for (Downloader t : temp) { + for (SubDownloadLoader t : temp) { addTask(t); } } } else { // 处理队列变大的情况，该情况下将增加任务 if (mExec.size() < num) { for (int i = 0; i < diff; i++) { - Downloader next = getNextTask(); + SubDownloadLoader next = getNextTask(); if (next != null) { startTask(next); } else { @@ -121,7 +124,7 @@ class SimpleSubQueue implements ISubQueue { } } - @Override public void removeTaskFromExecQ(Downloader fileer) { + @Override public void removeTaskFromExecQ(SubDownloadLoader fileer) { if (mExec.containsKey(fileer.getKey())) { if (fileer.isRunning()) { fileer.stop(); @@ -130,16 +133,21 @@ class SimpleSubQueue implements ISubQueue { } } - @Override public void removeTask(Downloader fileer) { + @Override public void removeTask(SubDownloadLoader fileer) { removeTaskFromExecQ(fileer); mCache.remove(fileer.getKey()); } - @Override public Downloader getNextTask() { + @Override public SubDownloadLoader getNextTask() { Iterator keys = mCache.keySet().iterator(); if (keys.hasNext()) { return mCache.get(keys.next()); } return null; } + + @Override public void clear() { + mCache.clear(); + mExec.clear(); + } } diff --git a/Aria/src/main/java/com/arialyy/aria/core/download/group/SubDownloadLoader.java b/Aria/src/main/java/com/arialyy/aria/core/download/group/SubDownloadLoader.java new file mode 100644 index 00000000..34d6e6f1 --- /dev/null +++ b/Aria/src/main/java/com/arialyy/aria/core/download/group/SubDownloadLoader.java @@ -0,0 +1,113 @@ +/* + * Copyright (C) 2016 AriaLyy(https://github.com/AriaLyy/Aria) + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package com.arialyy.aria.core.download.group; + +import android.os.Handler; +import com.arialyy.aria.core.common.CompleteInfo; +import com.arialyy.aria.core.common.IUtil; +import com.arialyy.aria.core.common.OnFileInfoCallback; +import com.arialyy.aria.core.download.DTaskWrapper; +import com.arialyy.aria.core.download.DownloadEntity; +import com.arialyy.aria.core.download.downloader.Downloader; +import com.arialyy.aria.core.download.downloader.HttpFileInfoThread; +import com.arialyy.aria.core.scheduler.ISchedulers; +import com.arialyy.aria.exception.BaseException; + +/** + * 子任务下载工具，负责创建{@link Downloader} + */ +class SubDownloadLoader implements IUtil { + + private Downloader mDownloader; + private DTaskWrapper mWrapper; + private Handler mSchedulers; + private ChildDownloadListener mListener; + + SubDownloadLoader(Handler schedulers, DTaskWrapper taskWrapper) { + mWrapper = taskWrapper; + mSchedulers = schedulers; + mListener = new ChildDownloadListener(mSchedulers, SubDownloadLoader.this); + } + + @Override public String getKey() { + return mWrapper.getKey(); + } + + public DTaskWrapper getWrapper() { + return mWrapper; + } + + public DownloadEntity getEntity() { + return mWrapper.getEntity(); + } + + /** + * 重新开始任务 + */ + void reStart() { + if (mDownloader != null) { + mDownloader.retryTask(); + } + } + + @Override public long getFileSize() { + return mDownloader == null ? -1 : mDownloader.getFileSize(); + } + + @Override public long getCurrentLocation() { + return mDownloader == null ? -1 : mDownloader.getCurrentLocation(); + } + + @Override public boolean isRunning() { + return mDownloader != null && mDownloader.isRunning(); + } + + @Override public void cancel() { + if (mDownloader != null) { + mDownloader.cancel(); + } + } + + @Override public void stop() { + if (mDownloader != null) { + mDownloader.stop(); + } + } + + @Override public void start() { + new Thread(new HttpFileInfoThread(mWrapper, new OnFileInfoCallback() { + + @Override public void onComplete(String url, CompleteInfo info) { + mDownloader = new Downloader(mListener, mWrapper); + mDownloader.start(); + } + + @Override public void onFail(String url, BaseException e, boolean needRetry) { + mSchedulers.obtainMessage(ISchedulers.FAIL, SubDownloadLoader.this); + } + })).start(); + } + + @Override public void resume() { + start(); + } + + @Override public void setMaxSpeed(int speed) { + if (mDownloader != null) { + mDownloader.setMaxSpeed(speed); + } + } +} diff --git a/Aria/src/main/java/com/arialyy/aria/core/inf/AbsEntity.java b/Aria/src/main/java/com/arialyy/aria/core/inf/AbsEntity.java index 58e2b0c9..5046623d 100644 --- a/Aria/src/main/java/com/arialyy/aria/core/inf/AbsEntity.java +++ b/Aria/src/main/java/com/arialyy/aria/core/inf/AbsEntity.java @@ -44,7 +44,7 @@ public abstract class AbsEntity extends DbEntity implements IEntity, Parcelable, /** * 文件大小 */ - private long fileSize = 1; + private long fileSize = 0; /** * 转换后的文件大小 */ diff --git a/Aria/src/main/java/com/arialyy/aria/core/inf/AbsGroupTask.java b/Aria/src/main/java/com/arialyy/aria/core/inf/AbsGroupTask.java index 1029b335..75336c0c 100644 --- a/Aria/src/main/java/com/arialyy/aria/core/inf/AbsGroupTask.java +++ b/Aria/src/main/java/com/arialyy/aria/core/inf/AbsGroupTask.java @@ -49,15 +49,4 @@ public abstract class AbsGroupTask> implements ISchedulers { + QUEUE extends ITaskQueue> implements ISchedulers { private final String TAG = "AbsSchedulers"; protected QUEUE mQueue; @@ -344,12 +346,10 @@ abstract class AbsSchedulers extends Handler.Callback { +public interface ISchedulers extends Handler.Callback { String ARIA_TASK_INFO_ACTION = "ARIA_TASK_INFO_ACTION"; /** @@ -113,10 +113,6 @@ public interface ISchedulers extends Handler.Callback { * 等待 */ int WAIT = 10; - /** - * 添加任务 - */ - int ADD = 11; /** * 组合任务子任务预处理 diff --git a/Aria/src/main/java/com/arialyy/aria/core/upload/uploader/SimpleUploadUtil.java b/Aria/src/main/java/com/arialyy/aria/core/upload/uploader/SimpleUploadUtil.java index 810de951..fe185f48 100644 --- a/Aria/src/main/java/com/arialyy/aria/core/upload/uploader/SimpleUploadUtil.java +++ b/Aria/src/main/java/com/arialyy/aria/core/upload/uploader/SimpleUploadUtil.java @@ -33,26 +33,26 @@ public class SimpleUploadUtil implements IUtil, Runnable { private static final String TAG = "SimpleUploadUtil"; private UploadEntity mUploadEntity; - private UTaskWrapper mTaskEntity; + private UTaskWrapper mTaskWrapper; private IUploadListener mListener; private Uploader mUploader; - public SimpleUploadUtil(UTaskWrapper taskEntity, IUploadListener listener) { - mTaskEntity = taskEntity; - CheckUtil.checkTaskEntity(taskEntity); - mUploadEntity = taskEntity.getEntity(); + public SimpleUploadUtil(UTaskWrapper taskWrapper, IUploadListener listener) { + mTaskWrapper = taskWrapper; + CheckUtil.checkTaskEntity(taskWrapper); + mUploadEntity = taskWrapper.getEntity(); if (listener == null) { throw new IllegalArgumentException("上传监听不能为空"); } mListener = listener; - mUploader = new Uploader(mListener, taskEntity); + mUploader = new Uploader(mListener, taskWrapper); } @Override public void run() { mListener.onPre(); - switch (mTaskEntity.getRequestType()) { + switch (mTaskWrapper.getRequestType()) { case AbsTaskWrapper.U_FTP: - new FtpFileInfoThread(mTaskEntity, new OnFileInfoCallback() { + new FtpFileInfoThread(mTaskWrapper, new OnFileInfoCallback() { @Override public void onComplete(String url, CompleteInfo info) { if (info.code == FtpFileInfoThread.CODE_COMPLETE) { mListener.onComplete(); @@ -72,6 +72,10 @@ public class SimpleUploadUtil implements IUtil, Runnable { } } + @Override public String getKey() { + return mTaskWrapper.getKey(); + } + @Override public long getFileSize() { return mUploader.getFileSize(); } diff --git a/app/src/main/assets/aria_config.xml b/app/src/main/assets/aria_config.xml index 6f1d56b7..eabeec56 100644 --- a/app/src/main/assets/aria_config.xml +++ b/app/src/main/assets/aria_config.xml @@ -131,7 +131,7 @@ - + @@ -155,7 +155,7 @@ - + \ No newline at end of file diff --git a/app/src/main/java/com/arialyy/simple/download/group/DownloadGroupActivity.java b/app/src/main/java/com/arialyy/simple/download/group/DownloadGroupActivity.java index 85e3d66a..3bb2920f 100644 --- a/app/src/main/java/com/arialyy/simple/download/group/DownloadGroupActivity.java +++ b/app/src/main/java/com/arialyy/simple/download/group/DownloadGroupActivity.java @@ -91,6 +91,7 @@ public class DownloadGroupActivity extends BaseActivity