tt/Aria
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

修复重构loader模块后出现的组合任务的bug

Browse Source
This commit is contained in:
laoyuyu
2020-01-02 21:04:44 +08:00
parent 102d3ba0f6
commit c39bfbdd88
29 changed files with 390 additions and 262 deletions
Download Patch File Download Diff File Expand all files Collapse all files

68
PublicComponent/src/main/java/com/arialyy/aria/core/common/SubThreadConfig.java
View File

@@ -16,14 +16,21 @@
package com.arialyy.aria.core.common;
import android.os.Handler;
import com.arialyy.aria.core.AriaConfig;
import com.arialyy.aria.core.ThreadRecord;
import com.arialyy.aria.core.wrapper.AbsTaskWrapper;
import com.arialyy.aria.core.wrapper.ITaskWrapper;
import java.io.File;
/**
* 子线程下载信息类
*/
public class SubThreadConfig {
public static final int TYPE_HTTP = 1;
public static final int TYPE_FTP = 2;
public static final int TYPE_M3U8_PEER = 3;
public static final int TYPE_HTTP_DG_SUB = 4;
public static final int TYPE_FTP_DG_SUB = 5;
public AbsTaskWrapper taskWrapper;
public boolean isBlock = false;
@@ -37,4 +44,65 @@ public class SubThreadConfig {
public Handler stateHandler;
// m3u8切片索引
public int peerIndex;
// 线程任务类型
public int threadType = TYPE_HTTP;
// 更新间隔，单位：毫秒
public long updateInterval = 1000;
/**
* 转换线程任务类型
*
* @param requestType {@link AbsTaskWrapper#getRequestType()}
* @return {@link #threadType}
*/
public static int getThreadType(int requestType) {
int threadType = SubThreadConfig.TYPE_HTTP;
switch (requestType) {
case ITaskWrapper.D_HTTP:
case ITaskWrapper.U_HTTP:
threadType = SubThreadConfig.TYPE_HTTP;
break;
case ITaskWrapper.D_FTP:
case ITaskWrapper.U_FTP:
threadType = SubThreadConfig.TYPE_FTP;
break;
case ITaskWrapper.D_FTP_DIR:
threadType = SubThreadConfig.TYPE_FTP_DG_SUB;
break;
case ITaskWrapper.DG_HTTP:
threadType = SubThreadConfig.TYPE_HTTP_DG_SUB;
break;
case ITaskWrapper.M3U8_LIVE:
case ITaskWrapper.M3U8_VOD:
threadType = SubThreadConfig.TYPE_M3U8_PEER;
break;
}
return threadType;
}
/**
* 根据配置肚脐更新间隔
*
* @param requestType {@link AbsTaskWrapper#getRequestType()}
* @return {@link #updateInterval}
*/
public static long getUpdateInterval(int requestType) {
long updateInterval = 1000;
switch (requestType) {
case ITaskWrapper.D_HTTP:
case ITaskWrapper.D_FTP:
case ITaskWrapper.M3U8_LIVE:
case ITaskWrapper.M3U8_VOD:
updateInterval = AriaConfig.getInstance().getDConfig().getUpdateInterval();
break;
case ITaskWrapper.D_FTP_DIR:
case ITaskWrapper.DG_HTTP:
updateInterval = AriaConfig.getInstance().getDGConfig().getUpdateInterval();
break;
case ITaskWrapper.U_HTTP:
case ITaskWrapper.U_FTP:
updateInterval = AriaConfig.getInstance().getUConfig().getUpdateInterval();
}
return updateInterval;
}
}

30
PublicComponent/src/main/java/com/arialyy/aria/core/group/AbsGroupLoader.java
View File

@@ -45,7 +45,7 @@ public abstract class AbsGroupLoader implements ILoaderVisitor, ILoader {
protected final String TAG = CommonUtil.getClassName(getClass());
private long mCurrentLocation = 0;
protected IDGroupListener mListener;
private IDGroupListener mListener;
private ScheduledThreadPoolExecutor mTimer;
private long mUpdateInterval;
private boolean isStop = false, isCancel = false;
@@ -76,6 +76,10 @@ public abstract class AbsGroupLoader implements ILoaderVisitor, ILoader {
*/
protected abstract AbsSubDLoadUtil createSubLoader(DTaskWrapper wrapper, boolean needGetFileInfo);
protected IDGroupListener getListener() {
return mListener;
}
protected DGTaskWrapper getWrapper() {
return mGTWrapper;
}
@@ -113,7 +117,7 @@ public abstract class AbsGroupLoader implements ILoaderVisitor, ILoader {
getWrapper().setState(IEntity.STATE_POST_PRE);
}
mState.updateProgress(mCurrentLocation);
mScheduler = new Handler(looper, SimpleSchedulers.newInstance(mState));
mScheduler = new Handler(looper, SimpleSchedulers.newInstance(mState, mGTWrapper.getKey()));
}
@Override public String getKey() {
@@ -248,18 +252,26 @@ public abstract class AbsGroupLoader implements ILoaderVisitor, ILoader {
mListener.onComplete();
return;
}
mListener.onPostPre(mGTWrapper.getEntity().getFileSize());
if (mCurrentLocation > 0) {
mListener.onResume(mCurrentLocation);
} else {
mListener.onStart(mCurrentLocation);
}
startTimer();
handlerTask(looper);
Looper.loop();
}
/**
* 组合任务获取完成子任务的信息后调用
*/
protected void onPostStart() {
if (isBreak()) {
return;
}
getListener().onPostPre(getWrapper().getEntity().getFileSize());
if (getWrapper().getEntity().getFileSize() > 0) {
getListener().onResume(getWrapper().getEntity().getCurrentProgress());
} else {
getListener().onStart(getWrapper().getEntity().getCurrentProgress());
}
}
private synchronized void startTimer() {
mState.isRunning = true;
mTimer = new ScheduledThreadPoolExecutor(1);

21
PublicComponent/src/main/java/com/arialyy/aria/core/group/AbsSubDLoadUtil.java
View File

@@ -34,19 +34,19 @@ public abstract class AbsSubDLoadUtil implements IUtil, Runnable {
protected SubLoader mDLoader;
private DTaskWrapper mWrapper;
private Handler mSchedulers;
private ChildDLoadListener mListener;
private boolean needGetInfo;
private boolean isStop = false, isCancel = false;
private String parentKey;
/**
* @param schedulers 调度器
* @param needGetInfo {@code true} 需要获取文件信息。{@code false} 不需要获取文件信息
*/
protected AbsSubDLoadUtil(DTaskWrapper taskWrapper, Handler schedulers, boolean needGetInfo) {
protected AbsSubDLoadUtil(DTaskWrapper taskWrapper, Handler schedulers, boolean needGetInfo, String parentKey) {
mWrapper = taskWrapper;
mSchedulers = schedulers;
this.parentKey = parentKey;
this.needGetInfo = needGetInfo;
mListener = new ChildDLoadListener(mSchedulers, AbsSubDLoadUtil.this);
mDLoader = getLoader();
}
@@ -57,6 +57,10 @@ public abstract class AbsSubDLoadUtil implements IUtil, Runnable {
protected abstract LoaderStructure buildLoaderStructure();
public String getParentKey() {
return parentKey;
}
protected boolean isNeedGetInfo() {
return needGetInfo;
}
@@ -66,7 +70,7 @@ public abstract class AbsSubDLoadUtil implements IUtil, Runnable {
}
@Override public String getKey() {
return mWrapper.getKey();
return mDLoader.getKey();
}
public DTaskWrapper getWrapper() {
@@ -77,15 +81,10 @@ public abstract class AbsSubDLoadUtil implements IUtil, Runnable {
return mWrapper.getEntity();
}
public ChildDLoadListener getListener() {
return mListener;
}
@Override public void run() {
if (isStop || isCancel) {
return;
}
mListener.onPre();
buildLoaderStructure();
mDLoader.run();
}
@@ -107,10 +106,6 @@ public abstract class AbsSubDLoadUtil implements IUtil, Runnable {
}
}
public SubLoader getDownloader() {
return mDLoader;
}
/**
* @deprecated 子任务不实现这个
*/

146
PublicComponent/src/main/java/com/arialyy/aria/core/group/ChildDLoadListener.java
View File

@@ -1,146 +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.group;
import android.os.Handler;
import com.arialyy.aria.core.download.DownloadEntity;
import com.arialyy.aria.core.inf.IEntity;
import com.arialyy.aria.core.listener.IDLoadListener;
import com.arialyy.aria.core.listener.ISchedulers;
import com.arialyy.aria.exception.BaseException;
import com.arialyy.aria.util.CommonUtil;
/**
* 子任务事件监听
*/
public class ChildDLoadListener implements IDLoadListener {
private DownloadEntity subEntity;
private int RUN_SAVE_INTERVAL = 5 * 1000; //5s保存一次下载中的进度
private long lastSaveTime;
private long lastLen;
private Handler schedulers;
private AbsSubDLoadUtil loader;
ChildDLoadListener(Handler schedulers, AbsSubDLoadUtil 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, subEntity.getCurrentProgress());
}
@Override public void onPostPre(long fileSize) {
subEntity.setFileSize(fileSize);
subEntity.setConvertFileSize(CommonUtil.formatFileSize(fileSize));
saveData(IEntity.STATE_POST_PRE, subEntity.getCurrentProgress());
sendToTarget(ISchedulers.POST_PRE, loader);
}
@Override public void onResume(long resumeLocation) {
lastLen = resumeLocation;
saveData(IEntity.STATE_POST_PRE, subEntity.getCurrentProgress());
sendToTarget(ISchedulers.START, loader);
}
@Override public void onStart(long startLocation) {
lastLen = startLocation;
saveData(IEntity.STATE_POST_PRE, subEntity.getCurrentProgress());
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, subEntity.getCurrentProgress());
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, AbsSubDLoadUtil util) {
schedulers.obtainMessage(state, util).sendToTarget();
}
}

56
PublicComponent/src/main/java/com/arialyy/aria/core/group/SimpleSchedulers.java
View File

@@ -16,6 +16,7 @@
package com.arialyy.aria.core.group;
import android.os.Bundle;
import android.os.Handler;
import android.os.Message;
import com.arialyy.aria.core.AriaConfig;
@@ -37,45 +38,56 @@ class SimpleSchedulers implements Handler.Callback {
private static final String TAG = "SimpleSchedulers";
private SimpleSubQueue mQueue;
private GroupRunState mGState;
private String mKey; // 组合任务的key
private SimpleSchedulers(GroupRunState state) {
private SimpleSchedulers(GroupRunState state, String key) {
mQueue = state.queue;
mGState = state;
mKey = key;
}
static SimpleSchedulers newInstance(GroupRunState state) {
return new SimpleSchedulers(state);
static SimpleSchedulers newInstance(GroupRunState state, String key) {
return new SimpleSchedulers(state, key);
}
@Override public boolean handleMessage(Message msg) {
// todo key 应该从bundle中获取
String key = (String) msg.obj;
AbsSubDLoadUtil loader = mQueue.getLoaderUtil(key);
if (loader == null) {
ALog.e(TAG, "子任务loder不存在，key：" + key);
Bundle b = msg.getData();
if (b == null) {
ALog.w(TAG, "组合任务子任务调度数据为空");
return true;
}
// todo 处理的是子任务的线程，需要删除 ThreadTaskManager.removeSingleTaskThread 删除线程任务
String threadName = b.getString(IThreadStateManager.DATA_THREAD_NAME);
AbsSubDLoadUtil loaderUtil = mQueue.getLoaderUtil(threadName);
if (loaderUtil == null) {
ALog.e(TAG, String.format("子任务loader不存在，state：%s，key：%s", msg.what, threadName));
return true;
}
long curLocation = b.getLong(IThreadStateManager.DATA_THREAD_LOCATION,
loaderUtil.getLoader().getWrapper().getEntity().getCurrentProgress());
// 处理状态
switch (msg.what) {
case IThreadStateManager.STATE_RUNNING:
mGState.listener.onSubRunning(loader.getEntity());
long range = (long) msg.obj;
mGState.listener.onSubRunning(loaderUtil.getEntity(), range);
break;
case PRE:
mGState.listener.onSubPre(loader.getEntity());
mGState.updateCount(loader.getKey());
case IThreadStateManager.STATE_PRE:
mGState.listener.onSubPre(loaderUtil.getEntity());
mGState.updateCount(loaderUtil.getKey());
break;
case START:
mGState.listener.onSubStart(loader.getEntity());
case IThreadStateManager.STATE_START:
mGState.listener.onSubStart(loaderUtil.getEntity());
break;
case IThreadStateManager.STATE_STOP:
handleStop(loader);
handleStop(loaderUtil, curLocation);
ThreadTaskManager.getInstance().removeSingleTaskThread(mKey, threadName);
break;
case IThreadStateManager.STATE_COMPLETE:
handleComplete(loader);
handleComplete(loaderUtil);
ThreadTaskManager.getInstance().removeSingleTaskThread(mKey, threadName);
break;
case IThreadStateManager.STATE_FAIL:
handleFail(loader);
handleFail(loaderUtil);
ThreadTaskManager.getInstance().removeSingleTaskThread(mKey, threadName);
break;
}
return true;
@@ -96,7 +108,7 @@ class SimpleSchedulers implements Handler.Callback {
final int reTryNum = num;
if ((!NetUtils.isConnected(AriaConfig.getInstance().getAPP()) && !isNotNetRetry)
|| loaderUtil.getDownloader() == null // 如果获取不到文件信息，loader为空
|| loaderUtil.getLoader() == null // 如果获取不到文件信息，loader为空
|| loaderUtil.getEntity().getFailNum() > reTryNum) {
mQueue.removeTaskFromExecQ(loaderUtil);
mGState.listener.onSubFail(loaderUtil.getEntity(), new TaskException(TAG,
@@ -136,8 +148,8 @@ class SimpleSchedulers implements Handler.Callback {
* 1、所有的子任务已经停止，则认为组合任务停止
* 2、completeNum + failNum + stopNum = subSize，则认为组合任务停止
*/
private synchronized void handleStop(AbsSubDLoadUtil loadUtil) {
mGState.listener.onSubStop(loadUtil.getEntity());
private synchronized void handleStop(AbsSubDLoadUtil loadUtil, long curLocation) {
mGState.listener.onSubStop(loadUtil.getEntity(), curLocation);
mGState.countStopNum(loadUtil.getKey());
if (mGState.getStopNum() == mGState.getSubSize()
|| mGState.getStopNum()

10
PublicComponent/src/main/java/com/arialyy/aria/core/group/SimpleSubQueue.java
View File

@@ -17,6 +17,7 @@ package com.arialyy.aria.core.group;
import com.arialyy.aria.core.config.Configuration;
import com.arialyy.aria.util.ALog;
import com.arialyy.aria.util.CommonUtil;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Iterator;
@@ -29,7 +30,7 @@ import java.util.Set;
* 组合任务队列，该队列生命周期和{@link AbsGroupLoaderUtil}生命周期一致
*/
class SimpleSubQueue implements ISubQueue<AbsSubDLoadUtil> {
private static final String TAG = "SimpleSubQueue";
private final String TAG = CommonUtil.getClassName(getClass());
/**
* 缓存下载器
*/
@@ -162,12 +163,7 @@ class SimpleSubQueue implements ISubQueue<AbsSubDLoadUtil> {
}
@Override public void removeTaskFromExecQ(AbsSubDLoadUtil fileer) {
if (mExec.containsKey(fileer.getKey())) {
if (fileer.isRunning()) {
fileer.stop();
}
mExec.remove(fileer.getKey());
}
mExec.remove(fileer.getKey());
}
@Override public void removeTask(AbsSubDLoadUtil fileer) {

6
PublicComponent/src/main/java/com/arialyy/aria/core/inf/IThreadStateManager.java
View File

@@ -32,8 +32,10 @@ public interface IThreadStateManager extends ILoaderComponent {
int STATE_UPDATE_PROGRESS = 0x06;
int STATE_PRE = 0x07;
int STATE_START = 0x08;
String KEY_RETRY = "KEY_RETRY";
String KEY_ERROR_INFO = "KEY_ERROR_INFO";
String DATA_RETRY = "DATA_RETRY";
String DATA_ERROR_INFO = "DATA_ERROR_INFO";
String DATA_THREAD_NAME = "DATA_THREAD_NAME";
String DATA_THREAD_LOCATION = "DATA_THREAD_LOCATION";
/**
* 任务是否已经失败

30
PublicComponent/src/main/java/com/arialyy/aria/core/listener/DownloadGroupListener.java
View File

@@ -21,8 +21,8 @@ import com.arialyy.aria.core.download.DownloadEntity;
import com.arialyy.aria.core.download.DownloadGroupEntity;
import com.arialyy.aria.core.group.GroupSendParams;
import com.arialyy.aria.core.inf.IEntity;
import com.arialyy.aria.core.loader.IRecordHandler;
import com.arialyy.aria.core.inf.TaskSchedulerType;
import com.arialyy.aria.core.loader.IRecordHandler;
import com.arialyy.aria.core.task.AbsTask;
import com.arialyy.aria.core.task.DownloadGroupTask;
import com.arialyy.aria.exception.BaseException;
@@ -47,6 +47,7 @@ public class DownloadGroupListener
@Override
public void onSubPre(DownloadEntity subEntity) {
handleSpeed(subEntity, 0);
saveSubState(IEntity.STATE_PRE, subEntity);
sendInState2Target(ISchedulers.SUB_PRE, subEntity);
}
@@ -58,12 +59,15 @@ public class DownloadGroupListener
@Override
public void onSubStart(DownloadEntity subEntity) {
handleSpeed(subEntity, 0);
saveSubState(IEntity.STATE_RUNNING, subEntity);
sendInState2Target(ISchedulers.SUB_START, subEntity);
}
@Override
public void onSubStop(DownloadEntity subEntity) {
public void onSubStop(DownloadEntity subEntity, long stopLocation) {
subEntity.setCurrentProgress(stopLocation);
handleSpeed(subEntity, 0);
saveSubState(IEntity.STATE_STOP, subEntity);
saveCurrentLocation();
sendInState2Target(ISchedulers.SUB_STOP, subEntity);
@@ -71,6 +75,7 @@ public class DownloadGroupListener
@Override
public void onSubComplete(DownloadEntity subEntity) {
handleSpeed(subEntity, 0);
saveSubState(IEntity.STATE_COMPLETE, subEntity);
saveCurrentLocation();
sendInState2Target(ISchedulers.SUB_COMPLETE, subEntity);
@@ -78,6 +83,7 @@ public class DownloadGroupListener
@Override
public void onSubFail(DownloadEntity subEntity, BaseException e) {
handleSpeed(subEntity, 0);
saveSubState(IEntity.STATE_FAIL, subEntity);
saveCurrentLocation();
sendInState2Target(ISchedulers.SUB_FAIL, subEntity);
@@ -89,13 +95,16 @@ public class DownloadGroupListener
@Override
public void onSubCancel(DownloadEntity subEntity) {
handleSpeed(subEntity, 0);
saveSubState(IEntity.STATE_CANCEL, subEntity);
saveCurrentLocation();
sendInState2Target(ISchedulers.SUB_CANCEL, subEntity);
}
@Override
public void onSubRunning(DownloadEntity subEntity) {
public void onSubRunning(DownloadEntity subEntity, long currentProgress) {
handleSpeed(subEntity, currentProgress);
if (System.currentTimeMillis() - mLastSaveTime >= RUN_SAVE_INTERVAL) {
saveSubState(IEntity.STATE_RUNNING, subEntity);
mLastSaveTime = System.currentTimeMillis();
@@ -103,6 +112,21 @@ public class DownloadGroupListener
sendInState2Target(ISchedulers.SUB_RUNNING, subEntity);
}
private void handleSpeed(DownloadEntity subEntity, long currentProgress) {
if (currentProgress == 0){
subEntity.setSpeed(0);
subEntity.setConvertSpeed("0kb/s");
return;
}
long range = currentProgress - subEntity.getCurrentProgress() ;
subEntity.setSpeed(range);
subEntity.setConvertSpeed(
range <= 0 ? "" : String.format("%s/s", CommonUtil.formatFileSize(range)));
subEntity.setPercent((int) (subEntity.getFileSize() <= 0 ? 0
: subEntity.getCurrentProgress() * 100 / subEntity.getFileSize()));
subEntity.setCurrentProgress(currentProgress);
}
/**
* 将任务状态发送给下载器
*

7
PublicComponent/src/main/java/com/arialyy/aria/core/listener/IDGroupListener.java
View File

@@ -16,7 +16,6 @@
package com.arialyy.aria.core.listener;
import com.arialyy.aria.core.download.DownloadEntity;
import com.arialyy.aria.core.listener.IDLoadListener;
import com.arialyy.aria.exception.BaseException;
/**
@@ -45,7 +44,7 @@ public interface IDGroupListener extends IDLoadListener {
/**
* 子任务停止下载
*/
void onSubStop(DownloadEntity subEntity);
void onSubStop(DownloadEntity subEntity, long stopLocation);
/**
* 子任务下载完成
@@ -64,6 +63,8 @@ public interface IDGroupListener extends IDLoadListener {
/**
* 子任务执行中
*
* @param currentProgress 当前进度
*/
void onSubRunning(DownloadEntity subEntity);
void onSubRunning(DownloadEntity subEntity, long currentProgress);
}

2
PublicComponent/src/main/java/com/arialyy/aria/core/loader/AbsNormalTTBuilder.java
View File

@@ -84,6 +84,8 @@ public abstract class AbsNormalTTBuilder implements IThreadTaskBuilder {
config.taskWrapper = mWrapper;
config.record = record;
config.stateHandler = mStateHandler;
config.threadType = SubThreadConfig.getThreadType(mWrapper.getRequestType());
config.updateInterval = SubThreadConfig.getUpdateInterval(mWrapper.getRequestType());
return createThreadTask(config);
}

4
PublicComponent/src/main/java/com/arialyy/aria/core/loader/NormalThreadStateManager.java
View File

@@ -90,8 +90,8 @@ public class NormalThreadStateManager implements IThreadStateManager {
mFailNum++;
if (isFail()) {
Bundle b = msg.getData();
mListener.onFail(b.getBoolean(KEY_RETRY, false),
(BaseException) b.getSerializable(KEY_ERROR_INFO));
mListener.onFail(b.getBoolean(DATA_RETRY, false),
(BaseException) b.getSerializable(DATA_ERROR_INFO));
quitLooper();
}
break;

86
PublicComponent/src/main/java/com/arialyy/aria/core/loader/SubLoader.java
View File

@@ -15,13 +15,17 @@
*/
package com.arialyy.aria.core.loader;
import android.os.Bundle;
import android.os.Handler;
import android.os.Message;
import android.text.TextUtils;
import com.arialyy.aria.core.TaskRecord;
import com.arialyy.aria.core.common.AbsEntity;
import com.arialyy.aria.core.common.CompleteInfo;
import com.arialyy.aria.core.inf.IThreadStateManager;
import com.arialyy.aria.core.listener.ISchedulers;
import com.arialyy.aria.core.manager.ThreadTaskManager;
import com.arialyy.aria.core.task.IThreadTask;
import com.arialyy.aria.core.task.ThreadTask;
import com.arialyy.aria.core.wrapper.AbsTaskWrapper;
import com.arialyy.aria.exception.BaseException;
import com.arialyy.aria.util.ALog;
@@ -42,29 +46,84 @@ public final class SubLoader implements ILoader, ILoaderVisitor {
private IThreadTaskBuilder ttBuild;
private IRecordHandler recordHandler;
private IThreadTask threadTask;
private String parentKey;
public SubLoader(AbsTaskWrapper wrapper, Handler schedulers) {
this.wrapper = wrapper;
this.schedulers = schedulers;
}
public AbsTaskWrapper getWrapper() {
return wrapper;
}
/**
* 发送状态到调度器
*
* @param state {@link IThreadStateManager}
*/
private void sendNormalState(int state) {
Message msg = schedulers.obtainMessage();
Bundle b = msg.getData();
if (b == null) {
b = new Bundle();
}
b.putString(IThreadStateManager.DATA_THREAD_NAME, getKey());
msg.what = state;
msg.setData(b);
msg.sendToTarget();
}
/**
* 发送失败的状态
*/
private void sendFailState(boolean needRetry) {
Message msg = schedulers.obtainMessage();
Bundle b = msg.getData();
if (b == null) {
b = new Bundle();
}
b.putString(IThreadStateManager.DATA_THREAD_NAME, getKey());
b.putBoolean(IThreadStateManager.DATA_RETRY, needRetry);
msg.what = IThreadStateManager.STATE_FAIL;
msg.setData(b);
msg.sendToTarget();
}
private void handlerTask() {
List<IThreadTask> task =
ttBuild.buildThreadTask(recordHandler.getRecord(wrapper.getEntity().getFileSize()),
schedulers);
if (task == null || task.isEmpty()) {
ALog.e(TAG, "创建子任务的线程任务失败，key：" + getKey());
//schedulers.obtainMessage(ISchedulers.FAIL, SubLoader.this).sendToTarget();
TaskRecord record = recordHandler.getRecord(wrapper.getEntity().getFileSize());
if (record.threadRecords != null
&& !record.threadRecords.isEmpty()
&& record.threadRecords.get(0).isComplete) {
ALog.d(TAG, "子任务已完成，key：" + wrapper.getKey());
sendNormalState(IThreadStateManager.STATE_COMPLETE);
return;
}
List<IThreadTask> task = ttBuild.buildThreadTask(record, schedulers);
if (task == null || task.isEmpty()) {
ALog.e(TAG, "创建子任务的线程任务失败，key：" + wrapper.getKey());
sendFailState(false);
return;
}
if (TextUtils.isEmpty(parentKey)) {
ALog.e(TAG, "parentKey为空");
sendFailState(false);
return;
}
sendNormalState(IThreadStateManager.STATE_PRE);
threadTask = task.get(0);
try {
ThreadTaskManager.getInstance().startThread(getKey(), threadTask);
ThreadTaskManager.getInstance().startThread(parentKey, threadTask);
sendNormalState(IThreadStateManager.STATE_START);
} catch (Exception e) {
e.printStackTrace();
}
}
public void setParentKey(String parentKey) {
this.parentKey = parentKey;
}
public void setNeedGetInfo(boolean needGetInfo) {
this.needGetInfo = needGetInfo;
}
@@ -91,7 +150,7 @@ public final class SubLoader implements ILoader, ILoaderVisitor {
}
@Override public boolean isRunning() {
return !threadTask.isBreak();
return threadTask != null && !threadTask.isBreak();
}
@Override public void cancel() {
@@ -112,8 +171,13 @@ public final class SubLoader implements ILoader, ILoaderVisitor {
return false;
}
/**
* 线程名，一个子任务的loader只有一个线程，使用线程名标示key
*
* @return {@link ThreadTask#getThreadName()}
*/
@Override public String getKey() {
return wrapper.getKey();
return CommonUtil.getThreadName(wrapper.getKey(), 0);
}
/**
@@ -136,7 +200,7 @@ public final class SubLoader implements ILoader, ILoaderVisitor {
}
@Override public void onFail(AbsEntity entity, BaseException e, boolean needRetry) {
schedulers.obtainMessage(ISchedulers.FAIL, SubLoader.this).sendToTarget();
sendFailState(needRetry);
}
});
}

45
PublicComponent/src/main/java/com/arialyy/aria/core/manager/ThreadTaskManager.java
View File

@@ -16,6 +16,7 @@
package com.arialyy.aria.core.manager;
import android.text.TextUtils;
import com.arialyy.aria.core.task.IThreadTask;
import com.arialyy.aria.core.wrapper.AbsTaskWrapper;
import com.arialyy.aria.util.ALog;
@@ -24,7 +25,6 @@ import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
@@ -153,6 +153,49 @@ public class ThreadTaskManager {
}
}
/**
* 根据线程名删除任务的中的线程
*
* @param key 任务的key，如果是组合任务，则为组合任务的key
* @param threadName 线程名
* @return true 删除线程成功；false 删除线程失败
*/
public boolean removeSingleTaskThread(String key, String threadName) {
try {
LOCK.tryLock(2, TimeUnit.SECONDS);
if (mExePool.isShutdown()) {
ALog.e(TAG, "线程池已经关闭");
return false;
}
if (TextUtils.isEmpty(threadName)) {
ALog.e(TAG, "线程名为空");
return false;
}
key = getKey(key);
Set<FutureContainer> temp = mThreadTasks.get(key);
if (temp != null && temp.size() > 0) {
FutureContainer tempC = null;
for (FutureContainer container : temp) {
if (container.threadTask.getThreadName().equals(threadName)) {
tempC = container;
break;
}
}
if (tempC != null) {
tempC.threadTask.destroy();
temp.remove(tempC);
return true;
}
}
} catch (Exception e) {
e.printStackTrace();
} finally {
LOCK.unlock();
}
return false;
}
/**
* 删除单个线程任务
*

6
PublicComponent/src/main/java/com/arialyy/aria/core/task/IThreadTask.java
View File

@@ -90,7 +90,11 @@ public interface IThreadTask extends Callable<IThreadTask> {
/**
* 获取线程id
* @return
*/
int getThreadId();
/**
* 线程名字，命名规则：md5(任务地址 + 线程id)
*/
String getThreadName();
}

51
PublicComponent/src/main/java/com/arialyy/aria/core/task/ThreadTask.java
View File

@@ -53,7 +53,7 @@ public class ThreadTask implements IThreadTask, IThreadTaskObserver {
private IEntity mEntity;
protected AbsTaskWrapper mTaskWrapper;
private int mFailTimes = 0;
private long mLastSaveTime;
private long mLastSaveTime, mLastSendProgressTime;
private boolean isNotNetRetry; //断网情况是否重试
private boolean taskBreak = false; //任务跳出
private boolean isDestroy = false;
@@ -67,6 +67,8 @@ public class ThreadTask implements IThreadTask, IThreadTaskObserver {
private long mRangeProgress;
private IThreadTaskAdapter mAdapter;
private ThreadRecord mRecord;
private String mThreadNmae;
private long updateInterval = 1000; // 更新间隔
private Thread mConfigThread = new Thread(new Runnable() {
@Override public void run() {
@@ -87,6 +89,7 @@ public class ThreadTask implements IThreadTask, IThreadTaskObserver {
isNotNetRetry = AriaConfig.getInstance().getAConfig().isNotNetRetry();
mRangeProgress = mRecord.startLocation;
updateInterval = config.updateInterval;
}
/**
@@ -219,7 +222,7 @@ public class ThreadTask implements IThreadTask, IThreadTaskObserver {
blockFile.getName(), blockFile.length(), mRecord.blockLen, mRecord.startLocation,
mRecord.endLocation));
if (blockFile.exists()) {
blockFile.delete();
FileUtil.deleteFile(blockFile);
ALog.i(TAG, String.format("删除分块【%s】成功", blockFile.getName()));
}
retryBlockTask(isBreak());
@@ -232,19 +235,24 @@ public class ThreadTask implements IThreadTask, IThreadTaskObserver {
return mRecord.threadId;
}
@Override public String getThreadName() {
return mThreadNmae == null ? CommonUtil.getThreadName(getConfig().url, getThreadId())
: mThreadNmae;
}
/**
* 停止任务
*/
@Override
public void stop() {
isStop = true;
final long stopLocation = mRangeProgress;
updateState(IThreadStateManager.STATE_STOP, null);
if (mTaskWrapper.getRequestType() == ITaskWrapper.M3U8_VOD) {
writeConfig(false, getConfig().tempFile.length());
ALog.i(TAG, String.format("任务【%s】已停止", getFileName()));
} else {
if (mTaskWrapper.isSupportBP()) {
final long stopLocation = mRangeProgress;
ALog.d(TAG,
String.format("任务【%s】thread__%s__停止【当前线程停止位置：%s】", getFileName(),
mRecord.threadId, stopLocation));
@@ -264,12 +272,14 @@ public class ThreadTask implements IThreadTask, IThreadTaskObserver {
@Override
public synchronized void updateState(int state, Bundle bundle) {
Message msg = mStateHandler.obtainMessage();
msg.what = state;
if (bundle != null) {
msg.setData(bundle);
if (bundle == null) {
bundle = new Bundle();
}
int reqType = mTaskWrapper.getRequestType();
msg.setData(bundle);
bundle.putString(IThreadStateManager.DATA_THREAD_NAME, getThreadName());
bundle.putLong(IThreadStateManager.DATA_THREAD_LOCATION, mRangeProgress);
msg.what = state;
int reqType = getConfig().threadType;
if (reqType == ITaskWrapper.M3U8_VOD || reqType == ITaskWrapper.M3U8_LIVE) {
sendM3U8Info(state, msg);
}
@@ -287,9 +297,6 @@ public class ThreadTask implements IThreadTask, IThreadTaskObserver {
}
Bundle bundle = msg.getData();
if ((state == IThreadStateManager.STATE_COMPLETE || state == IThreadStateManager.STATE_FAIL)) {
if (bundle == null) {
bundle = new Bundle();
}
bundle.putString(ISchedulers.DATA_M3U8_URL, getConfig().url);
bundle.putString(ISchedulers.DATA_M3U8_PEER_PATH, getConfig().tempFile.getPath());
bundle.putInt(ISchedulers.DATA_M3U8_PEER_INDEX, getConfig().peerIndex);
@@ -318,7 +325,21 @@ public class ThreadTask implements IThreadTask, IThreadTaskObserver {
if (!loopThread.isAlive() || loopThread.isInterrupted()) {
return;
}
mStateHandler.obtainMessage(IThreadStateManager.STATE_RUNNING, len).sendToTarget();
// 不需要太频繁发送，以减少消息队列的压力
if (System.currentTimeMillis() - mLastSendProgressTime > updateInterval) {
Message msg = mStateHandler.obtainMessage();
Bundle b = msg.getData();
if (b == null) {
b = new Bundle();
msg.setData(b);
}
b.putString(IThreadStateManager.DATA_THREAD_NAME, getThreadName());
msg.what = IThreadStateManager.STATE_RUNNING;
msg.obj = mRangeProgress;
msg.sendToTarget();
mLastSendProgressTime = System.currentTimeMillis();
}
if (System.currentTimeMillis() - mLastSaveTime > 5000
&& mRangeProgress < mRecord.endLocation) {
mLastSaveTime = System.currentTimeMillis();
@@ -435,7 +456,7 @@ public class ThreadTask implements IThreadTask, IThreadTaskObserver {
*/
if (blockFileLen > threadRect) {
ALog.i(TAG, String.format("分块【%s】错误，将重新下载该分块", temp.getName()));
temp.delete();
FileUtil.deleteFile(temp);
mRecord.startLocation = mRecord.endLocation - mRecord.blockLen;
mRecord.isComplete = false;
} else if (blockFileLen < mRecord.blockLen) {
@@ -459,9 +480,9 @@ public class ThreadTask implements IThreadTask, IThreadTaskObserver {
*/
private void sendFailMsg(BaseException e, boolean needRetry) {
Bundle b = new Bundle();
b.putBoolean(IThreadStateManager.KEY_RETRY, needRetry);
b.putBoolean(IThreadStateManager.DATA_RETRY, needRetry);
if (e != null) {
b.putSerializable(IThreadStateManager.KEY_ERROR_INFO, e);
b.putSerializable(IThreadStateManager.DATA_ERROR_INFO, e);
updateState(IThreadStateManager.STATE_FAIL, b);
} else {
updateState(IThreadStateManager.STATE_FAIL, b);

5
PublicComponent/src/main/java/com/arialyy/aria/core/wrapper/ITaskWrapper.java
View File

@@ -97,7 +97,10 @@ public interface ITaskWrapper {
IEntity getEntity();
/**
* 获取任务唯一标志
* 获取任务唯一标志：
* 下载任务，对应文件的下载地址
* 上传任务，对应需要上传的文件路径
* 组合任务，对应组合任务的groupHash
*/
String getKey();
}

14
PublicComponent/src/main/java/com/arialyy/aria/util/CommonUtil.java
View File

@@ -61,6 +61,16 @@ public class CommonUtil {
public static final String SERVER_CHARSET = "ISO-8859-1";
private static long lastClickTime;
/**
* 获取线程名称，命名规则：md5(任务地址 + 线程id)
*
* @param url 任务地址
* @param threadId 线程id
*/
public static String getThreadName(String url, int threadId) {
return getStrMd5(url.concat(String.valueOf(threadId)));
}
/**
* 检查sql的expression是否合法
*
@@ -436,7 +446,7 @@ public class CommonUtil {
md.update(str.getBytes());
return new BigInteger(1, md.digest()).toString(16);
} catch (NoSuchAlgorithmException e) {
ALog.e(TAG, e.getMessage());
e.printStackTrace();
}
return "";
}
@@ -633,7 +643,7 @@ public class CommonUtil {
ignore.add(field);
}
}
if (!ignore.isEmpty()){
if (!ignore.isEmpty()) {
fields.removeAll(ignore);
}
return fields;

2
PublicComponent/src/main/java/com/arialyy/aria/util/RecordUtil.java
View File

@@ -85,7 +85,7 @@ public class RecordUtil {
DbEntity.findRelationData(RecordWrapper.class, "dGroupHash=?", groupEntity.getGroupHash());
if (records == null || records.isEmpty()) {
ALog.w(TAG, "组任务记录删除失败，记录为null");
ALog.w(TAG, "组任务记录已删除");
} else {
for (RecordWrapper record : records) {
if (record == null || record.taskRecord == null) {