thread task

v4
laoyuyu 2 years ago
parent e137263b2c
commit 75d22ea028
  1. 22
      Http/src/main/java/com/arialyy/aria/http/download/HttpBlockThreadInterceptor.kt
  2. 5
      Http/src/main/java/com/arialyy/aria/http/download/HttpDBlockInterceptor.kt
  3. 78
      Http/src/main/java/com/arialyy/aria/http/download/HttpDCTTaskAdapter.kt
  4. 5
      Http/src/main/java/com/arialyy/aria/http/download/HttpDHeaderInterceptor.kt
  5. 2
      Http/src/main/java/com/arialyy/aria/http/download/HttpDStartController.kt
  6. 96
      Http/src/main/java/com/arialyy/aria/http/download/HttpDTTaskAdapter.kt
  7. 13
      Http/src/main/java/com/arialyy/aria/http/download/HttpDTaskOption.kt
  8. 18
      Http/src/main/java/com/arialyy/aria/http/download/HttpDTaskUtil.kt
  9. 82
      Http/src/main/java/com/arialyy/aria/http/download/HttpDThreadTaskAdapter.java
  10. 5
      PublicComponent/src/main/java/com/arialyy/aria/core/DuaContext.kt
  11. 6
      PublicComponent/src/main/java/com/arialyy/aria/core/inf/IBlockManager.java
  12. 84
      PublicComponent/src/main/java/com/arialyy/aria/core/task/AbsThreadTaskAdapter.java
  13. 13
      PublicComponent/src/main/java/com/arialyy/aria/core/task/DBlockManager.kt
  14. 3
      PublicComponent/src/main/java/com/arialyy/aria/core/task/IThreadTask.java
  15. 13
      PublicComponent/src/main/java/com/arialyy/aria/core/task/IThreadTaskAdapter.java
  16. 6
      PublicComponent/src/main/java/com/arialyy/aria/core/task/IThreadTaskObserver.java
  17. 11
      PublicComponent/src/main/java/com/arialyy/aria/core/task/TaskState.kt
  18. 26
      PublicComponent/src/main/java/com/arialyy/aria/core/task/ThreadConfig.kt
  19. 2
      PublicComponent/src/main/java/com/arialyy/aria/core/task/ThreadTask.java
  20. 4
      PublicComponent/src/main/java/com/arialyy/aria/core/task/ThreadTask2.kt

@ -15,11 +15,13 @@
*/
package com.arialyy.aria.http.download
import com.arialyy.aria.core.DuaContext
import com.arialyy.aria.core.task.ITaskInterceptor
import com.arialyy.aria.core.task.IThreadTask
import com.arialyy.aria.core.task.TaskChain
import com.arialyy.aria.core.task.TaskResp
import com.arialyy.aria.core.task.ThreadTask
import com.arialyy.aria.core.task.ThreadConfig
import com.arialyy.aria.core.task.ThreadTask2
import com.arialyy.aria.orm.entity.BlockRecord
/**
@ -34,13 +36,25 @@ class HttpBlockThreadInterceptor : ITaskInterceptor {
if (unfinishedBlockList.isEmpty()) {
return TaskResp(TaskResp.CODE_BLOCK_QUEUE_NULL)
}
createThreadTask(unfinishedBlockList)
createThreadTask(unfinishedBlockList, chain)
return TaskResp(TaskResp.CODE_SUCCESS)
}
private fun createThreadTask(blockRecordList: List<BlockRecord>) {
private fun createThreadTask(blockRecordList: List<BlockRecord>, chain: TaskChain) {
val threadTaskList = mutableListOf<IThreadTask>()
blockRecordList.forEach{
blockRecordList.forEach {
val option = chain.getTask().getTaskOption(HttpDTaskOption::class.java)
val threadConfig = ThreadConfig(it, option, DuaContext.getDConfig().maxSpeed)
threadTaskList.add(
ThreadTask2(
adapter = if (option.isChunkTask) HttpDCTTaskAdapter(threadConfig) else HttpDThreadTaskAdapter(
threadConfig
),
handler = chain.blockManager.handler,
record = it
)
)
}
}
}

@ -15,8 +15,6 @@
*/
package com.arialyy.aria.http.download
import android.os.Handler
import android.os.Looper
import com.arialyy.aria.core.DuaContext
import com.arialyy.aria.core.inf.IBlockManager
import com.arialyy.aria.core.task.BlockUtil
@ -126,7 +124,7 @@ internal class HttpDBlockInterceptor : ITaskInterceptor {
* if block already exist, upload progress
*/
private suspend fun checkBlock(): Int {
val handler = Handler(Looper.myLooper()!!, blockManager.handlerCallback)
val handler = blockManager.handler
val needUpdateBlockRecord = mutableSetOf<BlockRecord>()
for (br in taskRecord.blockList) {
val blockF = File(br.blockPath)
@ -145,7 +143,6 @@ internal class HttpDBlockInterceptor : ITaskInterceptor {
handler.obtainMessage(IBlockManager.STATE_UPDATE_PROGRESS, br.curProgress)
}
}
handler.removeCallbacksAndMessages(null)
// update block record
val dao = DuaContext.getServiceManager().getDbService().getDuaDb().getRecordDao()

@ -0,0 +1,78 @@
/*
* 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.http.download
import com.arialyy.aria.core.task.AbsThreadTaskAdapter
import com.arialyy.aria.core.task.ThreadConfig
import com.arialyy.aria.http.ConnectionHelp
import com.arialyy.aria.http.request.IRequest.Companion.getRequest
import java.io.BufferedInputStream
import java.io.FileOutputStream
import java.io.InputStream
import java.nio.ByteBuffer
import java.nio.channels.Channels
/**
* http chunk download thread task
* @Author laoyuyu
* @Description
* @Date 23:07 PM 2023/2/8
**/
class HttpDCTTaskAdapter(threadConfig: ThreadConfig) : AbsThreadTaskAdapter(threadConfig) {
private fun getTaskOption(): HttpDTaskOption {
return threadConfig.option as HttpDTaskOption
}
override fun handlerThreadTask() {
val taskOption = getTaskOption()
val option = taskOption.httpOption!!
val conn = getRequest(option).getDConnection(taskOption.sourUrl!!, option)
conn.doInput = true
conn.setChunkedStreamingMode(0)
conn.connect()
BufferedInputStream(ConnectionHelp.convertInputStream(conn)).use {
readBytes(it)
}
}
private fun readBytes(ips: InputStream) {
val fos = FileOutputStream(blockRecord.blockPath, true)
val foc = fos.channel
val fic = Channels.newChannel(ips)
val bf = ByteBuffer.allocate(BUF_SIZE)
fic.use {
var len: Long
while (fic.read(bf).also { len = it.toLong() } != -1) {
if (isBreakTask) {
break
}
if (mSpeedBandUtil != null) {
mSpeedBandUtil.limitNextBytes(len.toInt())
}
bf.flip()
foc.write(bf)
bf.compact()
progress(len)
}
// 将数据刷出到磁盘
foc.force(true)
foc.close()
fos.close()
}
}
}

@ -62,10 +62,9 @@ internal class HttpDHeaderInterceptor : ITaskInterceptor {
try {
val fileSize = getFileSize()
if (fileSize >= 0) {
task.taskState.isSupportResume = fileSize != 0L
task.taskState.isSupportBlock =
task.taskState.isSupportResume && fileSize > BlockRecord.BLOCK_SIZE
task.taskState.fileSize = fileSize
taskOption.isSupportResume = fileSize != 0L
taskOption.isSupportBlock = taskOption.isSupportResume && fileSize > BlockRecord.BLOCK_SIZE
return chain.proceed(task)
}
} catch (e: IOException) {

@ -50,7 +50,7 @@ class HttpDStartController(target: Any, val url: String) : HttpBaseController(ta
* use multi-threaded download file, if file size <= 5m, this setting is not valid
* @param threadNum range [1 - 32]
*/
fun setThreadNum(threadNum: Long): HttpDStartController {
fun setThreadNum(threadNum: Int): HttpDStartController {
if (threadNum !in 1..32) {
Timber.e("set thread num fail, only 0 < threadNum < 33, threadNum: $threadNum")
return this

@ -0,0 +1,96 @@
/*
* 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.http.download
import com.arialyy.aria.core.task.AbsThreadTaskAdapter
import com.arialyy.aria.core.task.ThreadConfig
import com.arialyy.aria.http.ConnectionHelp
import com.arialyy.aria.http.request.IRequest.Companion.getRequest
import timber.log.Timber
import java.io.BufferedInputStream
import java.io.FileOutputStream
import java.io.InputStream
import java.nio.ByteBuffer
import java.nio.channels.Channels
/**
* http download thread task, not support chunk
* @Author laoyuyu
* @Description
* @Date 23:07 PM 2023/2/8
**/
class HttpDTTaskAdapter(threadConfig: ThreadConfig) : AbsThreadTaskAdapter(threadConfig) {
private fun getTaskOption(): HttpDTaskOption {
return threadConfig.option as HttpDTaskOption
}
override fun handlerThreadTask() {
val taskOption = getTaskOption()
val option = taskOption.httpOption!!
val conn = getRequest(option).getDConnection(taskOption.sourUrl!!, option)
if (!taskOption.isSupportResume) {
Timber.w("this task not support resume, url: %s", taskOption.sourUrl)
} else {
conn.setRequestProperty(
"Range", String.format(
"bytes=%s-%s", threadConfig.blockRecord.startLocation,
threadConfig.blockRecord.endLocation - 1
)
)
}
conn.connect()
BufferedInputStream(ConnectionHelp.convertInputStream(conn)).use {
readBytes(it)
}
}
private fun readBytes(ips: InputStream) {
val fos = FileOutputStream(blockRecord.blockPath, true)
val foc = fos.channel
val fic = Channels.newChannel(ips)
val bf = ByteBuffer.allocate(BUF_SIZE)
fic.use {
var len: Long
while (fic.read(bf).also { len = it.toLong() } != -1) {
if (isBreakTask) {
break
}
if (mSpeedBandUtil != null) {
mSpeedBandUtil.limitNextBytes(len.toInt())
}
if (rangeProgress + len >= blockRecord.endLocation) {
len = (blockRecord.endLocation - rangeProgress)
bf.flip()
fos.write(bf.array(), 0, len.toInt())
bf.compact()
progress(len)
break
} else {
bf.flip()
foc.write(bf)
bf.compact()
progress(len)
}
}
// 将数据刷出到磁盘
foc.force(true)
foc.close()
fos.close()
}
}
}

@ -27,9 +27,20 @@ import com.arialyy.aria.http.HttpOption
**/
class HttpDTaskOption : DTaskOption() {
var httpOption: HttpOption? = null
var fileSizeAdapter: IHttpFileLenAdapter? = null
var taskInterceptor = mutableListOf<ITaskInterceptor>()
var isChunkTask = false
/**
* whether block is supported, true: supported
*/
var isSupportResume = true
/**
* whether resume task is supported
* 1. in download task, if file length not obtained, isSupportResume = false
* 2. in upload task, if service not supported resume, isSupportResume = false
*/
var isSupportBlock = true
}

@ -1,3 +1,18 @@
/*
* 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.http.download
import android.os.Looper
@ -43,10 +58,11 @@ internal class HttpDTaskUtil : AbsTaskUtil() {
}
DuaContext.duaScope.launch(Dispatchers.IO) {
Looper.prepare()
blockManager?.setLopper(Looper.myLooper()!!)
blockManager?.setLooper()
addCoreInterceptor(TimerInterceptor())
addCoreInterceptor(HttpDHeaderInterceptor())
addCoreInterceptor(HttpDBlockInterceptor())
addCoreInterceptor(HttpBlockThreadInterceptor())
val resp = interceptor()
Looper.loop()
}

@ -16,11 +16,12 @@
package com.arialyy.aria.http.download;
import com.arialyy.aria.core.common.RequestEnum;
import com.arialyy.aria.core.common.SubThreadConfig;
import com.arialyy.aria.core.download.DTaskWrapper;
import com.arialyy.aria.core.task.AbsThreadTaskAdapter;
import com.arialyy.aria.core.task.ThreadConfig;
import com.arialyy.aria.exception.AriaHTTPException;
import com.arialyy.aria.http.BaseHttpThreadTaskAdapter;
import com.arialyy.aria.http.ConnectionHelp;
import com.arialyy.aria.http.HttpOption;
import com.arialyy.aria.http.request.IRequest;
import com.arialyy.aria.util.ALog;
import com.arialyy.aria.util.BufferedRandomAccessFile;
import java.io.BufferedInputStream;
@ -30,7 +31,6 @@ import java.io.InputStream;
import java.io.OutputStreamWriter;
import java.net.HttpURLConnection;
import java.net.MalformedURLException;
import java.net.URL;
import java.net.URLEncoder;
import java.nio.ByteBuffer;
import java.nio.channels.Channels;
@ -38,41 +38,37 @@ import java.nio.channels.FileChannel;
import java.nio.channels.ReadableByteChannel;
import java.util.Map;
import java.util.Set;
import timber.log.Timber;
/**
* Created by lyy on 2017/1/18. 下载线程
*/
final class HttpDThreadTaskAdapter extends BaseHttpThreadTaskAdapter {
private final String TAG = "HttpDThreadTaskAdapter";
private DTaskWrapper mTaskWrapper;
final class HttpDThreadTaskAdapter extends AbsThreadTaskAdapter {
HttpDThreadTaskAdapter(SubThreadConfig config) {
super(config);
HttpDThreadTaskAdapter(ThreadConfig threadConfig) {
super(threadConfig);
}
private HttpDTaskOption getTaskOption() {
return (HttpDTaskOption) getThreadConfig().getOption();
}
@Override protected void handlerThreadTask() {
mTaskWrapper = (DTaskWrapper) getTaskWrapper();
if (getThreadRecord().isComplete) {
handleComplete();
return;
}
HttpURLConnection conn = null;
BufferedInputStream is = null;
BufferedRandomAccessFile file = null;
try {
URL url = ConnectionHelp.handleUrl(getThreadConfig().url, mTaskOption);
conn = ConnectionHelp.handleConnection(url, mTaskOption);
if (mTaskWrapper.isSupportBP()) {
ALog.d(TAG,
String.format("任务【%s】线程__%s__开始下载【开始位置 : %s,结束位置:%s】", getFileName(),
getThreadRecord().threadId, getThreadRecord().startLocation,
getThreadRecord().endLocation));
HttpDTaskOption taskOption = getTaskOption();
HttpOption option = taskOption.getHttpOption();
conn = IRequest.Companion.getRequest(option).getDConnection(taskOption.getSourUrl(), option);
if (!taskOption.isSupportResume()) {
Timber.w("this task not support resume, url: %s", taskOption.getSourUrl());
}else {
conn.setRequestProperty("Range",
String.format("bytes=%s-%s", getThreadRecord().startLocation,
(getThreadRecord().endLocation - 1)));
} else {
ALog.w(TAG, "该下载不支持断点");
String.format("bytes=%s-%s", getThreadConfig().getBlockRecord().getStartLocation(),
(getThreadConfig().getBlockRecord().getEndLocation() - 1)));
}
ConnectionHelp.setConnectParam(mTaskOption, conn);
conn.setConnectTimeout(getTaskConfig().getConnectTimeOut());
conn.setReadTimeout(getTaskConfig().getIOTimeOut()); //设置读取流的等待时间,必须设置该参数
@ -81,23 +77,6 @@ final class HttpDThreadTaskAdapter extends BaseHttpThreadTaskAdapter {
conn.setChunkedStreamingMode(0);
}
conn.connect();
// 传递参数
if (mTaskOption.getRequestEnum() == RequestEnum.POST) {
Map<String, String> params = mTaskOption.getParams();
if (params != null) {
OutputStreamWriter dos = new OutputStreamWriter(conn.getOutputStream());
Set<String> keys = params.keySet();
StringBuilder sb = new StringBuilder();
for (String key : keys) {
sb.append(key).append("=").append(URLEncoder.encode(params.get(key))).append("&");
}
String paramStr = sb.toString();
paramStr = paramStr.substring(0, paramStr.length() - 1);
dos.write(paramStr);
dos.flush();
dos.close();
}
}
is = new BufferedInputStream(ConnectionHelp.convertInputStream(conn));
if (mTaskOption.isChunked()) {
@ -220,7 +199,8 @@ final class HttpDThreadTaskAdapter extends BaseHttpThreadTaskAdapter {
}
handleComplete();
} catch (IOException e) {
fail(new AriaHTTPException(String.format("文件下载失败,savePath: %s, url: %s", getEntity().getFilePath(),
fail(new AriaHTTPException(
String.format("文件下载失败,savePath: %s, url: %s", getEntity().getFilePath(),
getThreadConfig().url), e), true);
} finally {
try {
@ -259,17 +239,11 @@ final class HttpDThreadTaskAdapter extends BaseHttpThreadTaskAdapter {
}
}
/**
* 处理完成配置文件的更新或事件回调
*/
private void handleComplete() {
if (getThreadTask().isBreak()) {
return;
}
if (!getThreadTask().checkBlock()) {
return;
}
@Override public void cancel() {
}
@Override public void stop() {
complete();
}
}

@ -20,7 +20,6 @@ import android.content.Context
import com.arialyy.aria.core.service.LifecycleManager
import com.arialyy.aria.core.service.ServiceManager
import kotlinx.coroutines.MainScope
import java.util.logging.Handler
/**
* @Author laoyuyu
@ -44,4 +43,8 @@ object DuaContext {
fun getServiceManager() = ServiceManager
fun getLifeManager() = LifecycleManager
fun getDConfig() = AriaConfig.getInstance().dConfig
fun getUConfig() = AriaConfig.getInstance().uConfig
}

@ -17,10 +17,8 @@ package com.arialyy.aria.core.inf;
import android.os.Handler;
import com.arialyy.aria.core.task.IThreadTask;
import com.arialyy.aria.core.task.ThreadTask;
import com.arialyy.aria.orm.entity.BlockRecord;
import java.util.List;
import kotlinx.coroutines.channels.Channel;
/**
* 线程任务状态
@ -40,6 +38,8 @@ public interface IBlockManager {
String DATA_THREAD_LOCATION = "DATA_THREAD_LOCATION";
String DATA_ADD_LEN = "DATA_ADD_LEN"; // 增加的长度
void setLooper();
void start(List<IThreadTask> threadTaskList);
void setBlockNum(int blockNum);
@ -78,5 +78,5 @@ public interface IBlockManager {
/**
* 创建handler 回调
*/
Handler.Callback getHandlerCallback();
Handler getHandler();
}

@ -15,13 +15,8 @@
*/
package com.arialyy.aria.core.task;
import com.arialyy.aria.core.ThreadRecord;
import com.arialyy.aria.core.common.SubThreadConfig;
import com.arialyy.aria.core.config.BaseTaskConfig;
import com.arialyy.aria.core.wrapper.AbsTaskWrapper;
import com.arialyy.aria.exception.AriaException;
import com.arialyy.aria.orm.entity.BlockRecord;
import com.arialyy.aria.util.BandwidthLimiter;
import com.arialyy.aria.util.CommonUtil;
/**
* @author lyy
@ -29,67 +24,56 @@ import com.arialyy.aria.util.CommonUtil;
*/
public abstract class AbsThreadTaskAdapter implements IThreadTaskAdapter {
protected String TAG = CommonUtil.getClassName(getClass());
/**
* 速度限制工具
*/
protected BandwidthLimiter mSpeedBandUtil;
private ThreadRecord mThreadRecord;
private IThreadTaskObserver mObserver;
private AbsTaskWrapper mWrapper;
private SubThreadConfig mThreadConfig;
private IThreadTask mThreadTask;
protected AbsThreadTaskAdapter(SubThreadConfig config) {
mThreadRecord = config.record;
mWrapper = config.taskWrapper;
mThreadConfig = config;
if (getTaskConfig().getMaxSpeed() > 0) {
mSpeedBandUtil = new BandwidthLimiter(getTaskConfig().getMaxSpeed(), config.startThreadNum);
private final ThreadConfig mThreadConfig;
private boolean breakTask = false;
protected AbsThreadTaskAdapter(ThreadConfig threadConfig) {
mThreadConfig = threadConfig;
if (threadConfig.getSpeed() > 0) {
mSpeedBandUtil =
new BandwidthLimiter(threadConfig.getSpeed(), mThreadConfig.getOption().threadNum);
}
}
@Override public void call(IThreadTask threadTask) throws Exception {
mThreadTask = threadTask;
handlerThreadTask();
@Override public void breakTask() {
breakTask = true;
}
/**
* 开始处理线程任务
*/
protected abstract void handlerThreadTask();
protected boolean isBreakTask() {
return breakTask;
}
/**
* 当前线程的下去区间的进度
*/
protected long getRangeProgress() {
return mObserver.getThreadProgress();
protected BlockRecord getBlockRecord() {
return mThreadConfig.getBlockRecord();
}
protected ThreadRecord getThreadRecord() {
return mThreadRecord;
protected ThreadConfig getThreadConfig() {
return mThreadConfig;
}
protected AbsTaskWrapper getTaskWrapper() {
return mWrapper;
@Override public void run() {
try {
handlerThreadTask();
} catch (Exception e) {
fail(e);
}
}
/**
* 获取任务配置信息
* 开始处理线程任务
*/
protected BaseTaskConfig getTaskConfig() {
return getTaskWrapper().getConfig();
}
protected IThreadTask getThreadTask() {
return mThreadTask;
}
protected abstract void handlerThreadTask();
/**
* 获取线程配置信息
* 当前线程的下去区间的进度
*/
protected SubThreadConfig getThreadConfig() {
return mThreadConfig;
protected long getRangeProgress() {
return mObserver.getThreadProgress();
}
@Override public void attach(IThreadTaskObserver observer) {
@ -99,26 +83,26 @@ public abstract class AbsThreadTaskAdapter implements IThreadTaskAdapter {
@Override public void setMaxSpeed(int speed) {
if (mSpeedBandUtil == null) {
mSpeedBandUtil =
new BandwidthLimiter(getTaskConfig().getMaxSpeed(), getThreadConfig().startThreadNum);
new BandwidthLimiter(mThreadConfig.getSpeed(), mThreadConfig.getOption().threadNum);
}
mSpeedBandUtil.setMaxRate(speed);
}
protected void complete() {
if (mObserver != null) {
mObserver.updateCompleteState();
mObserver.onComplete();
}
}
protected void fail(AriaException ex, boolean needRetry) {
protected void fail(Exception ex) {
if (mObserver != null) {
mObserver.updateFailState(ex, needRetry);
mObserver.onFail(ex);
}
}
protected void progress(long len) {
if (mObserver != null) {
mObserver.updateProgress(len);
mObserver.onProgress(len);
}
}
}

@ -15,6 +15,7 @@
*/
package com.arialyy.aria.core.task
import android.os.Handler
import android.os.Handler.Callback
import android.os.Looper
import com.arialyy.aria.core.inf.IBlockManager
@ -49,6 +50,7 @@ class BlockManager(task: ITask) : IBlockManager {
private var progress: Long = 0 //当前总进度
private lateinit var looper: Looper
private lateinit var handler: Handler
private var blockNum: Int = 1
private var eventListener: IEventListener =
task.getTaskOption(ITaskOption::class.java).taskListener
@ -105,6 +107,7 @@ class BlockManager(task: ITask) : IBlockManager {
*/
private fun quitLooper() {
looper.quit()
handler.removeCallbacksAndMessages(null)
scope.cancel()
}
@ -116,11 +119,15 @@ class BlockManager(task: ITask) : IBlockManager {
return unfinishedBlock
}
override fun start(threadTaskList: List<IThreadTask>) {
override fun setLooper() {
if (Looper.myLooper() == Looper.getMainLooper()) {
throw IllegalThreadStateException("io operations cannot be in the main thread")
}
looper = Looper.myLooper()!!
handler = Handler(looper, callback)
}
override fun start(threadTaskList: List<IThreadTask>) {
threadTaskList.forEach { tt ->
scope.launch(dispatcher) {
tt.run()
@ -163,7 +170,7 @@ class BlockManager(task: ITask) : IBlockManager {
return scope.isActive
}
override fun getHandlerCallback(): Callback {
return callback
override fun getHandler(): Handler {
return handler
}
}

@ -45,7 +45,4 @@ public interface IThreadTask extends Runnable {
*/
boolean isRunning();
void onFail(Exception e);
void onComplete();
}

@ -23,20 +23,17 @@ package com.arialyy.aria.core.task;
*/
public interface IThreadTaskAdapter {
/**
* 执行任务
*/
void run(IThreadTask threadTask);
int BUF_SIZE = 4096;
/**
* 取消任务
* 执行任务
*/
void cancel();
void run();
/**
* 停止任务
* 中断任务
*/
void stop();
void breakTask();
/**
* 设置当前线程最大下载速度

@ -28,10 +28,14 @@ public interface IThreadTaskObserver {
*
* @param len 新增的长度
*/
void updateProgress(long len);
void onProgress(long len);
/**
* 获取线程当前进度
*/
long getThreadProgress();
void onFail(Exception e);
void onComplete();
}

@ -43,17 +43,6 @@ class TaskState {
*/
var curProgress: Long = 0
/**
* whether block is supported, true: supported
*/
var isSupportBlock = false
/**
* whether resume task is supported
* 1. in download task, if file length not obtained, isSupportResume = false
* 2. in upload task, if service not supported resume, isSupportResume = false
*/
var isSupportResume = false
/**
* Bytes transferred in 1 second, if file size 0, return 0

@ -0,0 +1,26 @@
/*
* 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.task
import com.arialyy.aria.core.inf.ITaskOption
import com.arialyy.aria.orm.entity.BlockRecord
/**
* @Author laoyuyu
* @Description
* @Date 23:12 PM 2023/2/8
**/
data class ThreadConfig(val blockRecord: BlockRecord, val option: ITaskOption, val speed: Int)

@ -331,7 +331,7 @@ public class ThreadTask implements IThreadTask, IThreadTaskObserver {
* @param len 新增的长度
*/
@Override
public synchronized void updateProgress(long len) {
public synchronized void onProgress(long len) {
mRangeProgress += len;
Thread loopThread = mStateHandler.getLooper().getThread();
if (!loopThread.isAlive() || loopThread.isInterrupted()) {

@ -47,7 +47,7 @@ class ThreadTask2(
}
override fun run() {
adapter.run(this)
adapter.run()
}
override fun cancel() {
@ -117,7 +117,7 @@ class ThreadTask2(
/**
* update current thread progress, once a second
*/
override fun updateProgress(len: Long) {
override fun onProgress(len: Long) {
record.curProgress += len
if (System.currentTimeMillis() - lastUpdateTime > 1000) {
lastUpdateTime = System.currentTimeMillis()

Loading…
Cancel
Save