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

sftp上传实现

Browse Source
This commit is contained in:
laoyuyu
2020-01-18 10:00:06 +08:00
parent 50b265f22a
commit d217ebcb25
59 changed files with 1616 additions and 387 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
PublicComponent/src/main/java/com/arialyy/aria/core/loader/AbsNormalLoader.java
View File

@@ -39,10 +39,10 @@ import java.util.concurrent.TimeUnit;
* 3、创建文件信息获取器，获取文件信息，根据文件信息执行任务
* 4、创建线程任务执行下载、上传操作
*/
public abstract class AbsNormalLoader implements ILoaderVisitor, ILoader {
public abstract class AbsNormalLoader<T extends AbsTaskWrapper> implements ILoaderVisitor, ILoader {
protected final String TAG = CommonUtil.getClassName(getClass());
private IEventListener mListener;
protected AbsTaskWrapper mTaskWrapper;
protected T mTaskWrapper;
protected File mTempFile;
private List<IThreadTask> mTask = new ArrayList<>();
@@ -61,7 +61,7 @@ public abstract class AbsNormalLoader implements ILoaderVisitor, ILoader {
protected IInfoTask mInfoTask;
protected IThreadTaskBuilder mTTBuilder;
protected AbsNormalLoader(AbsTaskWrapper wrapper, IEventListener listener) {
protected AbsNormalLoader(T wrapper, IEventListener listener) {
mListener = listener;
mTaskWrapper = wrapper;
}
@@ -123,7 +123,7 @@ public abstract class AbsNormalLoader implements ILoaderVisitor, ILoader {
}
Looper.prepare();
Looper looper = Looper.myLooper();
if (looper == Looper.getMainLooper()){
if (looper == Looper.getMainLooper()) {
throw new IllegalThreadStateException("不能在主线程程序中调用Loader");
}
isRuning = true;

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

@@ -34,12 +34,12 @@ import java.io.File;
/**
* 单文件
*/
public class NormalLoader extends AbsNormalLoader {
public class NormalLoader<T extends AbsTaskWrapper> extends AbsNormalLoader<T> {
private int startThreadNum; //启动的线程数
protected boolean isComplete = false;
private Looper looper;
public NormalLoader(AbsTaskWrapper wrapper, IEventListener listener) {
public NormalLoader(T wrapper, IEventListener listener) {
super(wrapper, listener);
mTempFile = new File(getEntity().getFilePath());
EventMsgUtil.getDefault().register(this);

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

@@ -22,7 +22,6 @@ import android.os.Message;
import com.arialyy.aria.core.TaskRecord;
import com.arialyy.aria.core.inf.IThreadStateManager;
import com.arialyy.aria.core.listener.IEventListener;
import com.arialyy.aria.core.wrapper.ITaskWrapper;
import com.arialyy.aria.exception.BaseException;
import com.arialyy.aria.util.ALog;
import com.arialyy.aria.util.FileUtil;
@@ -100,6 +99,11 @@ public class NormalThreadStateManager implements IThreadStateManager {
mCompleteNum++;
if (isComplete()) {
ALog.d(TAG, "isComplete, completeNum = " + mCompleteNum);
//if (mTaskRecord.taskType == ITaskWrapper.D_SFTP) {
// mergerSFtp();
// mListener.onComplete();
//} else
if (mTaskRecord.isBlock) {
if (mergeFile()) {
mListener.onComplete();
@@ -215,6 +219,26 @@ public class NormalThreadStateManager implements IThreadStateManager {
return size;
}
/**
* 合并sftp的分块
*/
private boolean mergerSFtp() {
if (mTaskRecord.threadNum == 1) {
File partFile = new File(String.format(IRecordHandler.SUB_PATH, mTaskRecord.filePath, 0));
return partFile.renameTo(new File(mTaskRecord.filePath));
}
List<String> partPath = new ArrayList<>();
for (int i = 0, len = mTaskRecord.threadNum; i < len; i++) {
partPath.add(String.format(IRecordHandler.SUB_PATH, mTaskRecord.filePath, i));
}
FileUtil.mergeSFtpFile(mTaskRecord.filePath, partPath, mTaskRecord.fileLength);
for (String pp : partPath) {
FileUtil.deleteFile(pp);
}
return true;
}
/**
* 合并文件
*
@@ -230,15 +254,10 @@ public class NormalThreadStateManager implements IThreadStateManager {
for (int i = 0, len = mTaskRecord.threadNum; i < len; i++) {
partPath.add(String.format(IRecordHandler.SUB_PATH, mTaskRecord.filePath, i));
}
boolean isSuccess = mTaskRecord.taskType == ITaskWrapper.D_SFTP ?
FileUtil.mergeSFtpFile(mTaskRecord.filePath, partPath, mTaskRecord.fileLength)
: FileUtil.mergeFile(mTaskRecord.filePath, partPath);
boolean isSuccess = FileUtil.mergeFile(mTaskRecord.filePath, partPath);
if (isSuccess) {
for (String pp : partPath) {
File f = new File(pp);
if (f.exists()) {
f.delete();
}
FileUtil.deleteFile(pp);
}
File targetFile = new File(mTaskRecord.filePath);
if (targetFile.exists() && targetFile.length() > mTaskRecord.fileLength) {

3
PublicComponent/src/main/java/com/arialyy/aria/core/processor/IKeyUrlConverter.java
View File

@@ -25,8 +25,9 @@ public interface IKeyUrlConverter extends IEventHandler {
/**
* 将被加密的密钥下载地址转换为可使用的http下载地址
*
* @param m3u8Url m3u8文件的下载地址
* @param keyUrl 加密的url地址
* @return 可正常访问的http地址
*/
String convert(String keyUrl);
String convert(String m3u8Url, String keyUrl);
}

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

@@ -367,6 +367,11 @@ public class ThreadTask implements IThreadTask, IThreadTaskObserver {
b.putLong(IThreadStateManager.DATA_ADD_LEN, mRangeProgress - mLastRangeProgress);
msg.what = IThreadStateManager.STATE_RUNNING;
msg.obj = mRangeProgress;
Thread loopThread = mStateHandler.getLooper().getThread();
if (!loopThread.isAlive() || loopThread.isInterrupted()) {
return;
}
msg.sendToTarget();
}
@@ -451,6 +456,7 @@ public class ThreadTask implements IThreadTask, IThreadTaskObserver {
handleBlockRecord();
ThreadTaskManager.getInstance().retryThread(this);
} else {
ALog.e(TAG, String.format("任务【%s】执行失败", getFileName()));
sendFailMsg(null, false);
}
}

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

@@ -84,6 +84,11 @@ public interface ITaskWrapper {
*/
int D_SFTP = 12;
/**
* SFTP 上传
*/
int U_SFTP = 13;
/**
* 获取任务类型
*

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

@@ -30,7 +30,6 @@ import java.io.File;
import java.io.FileFilter;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.UnsupportedEncodingException;
@@ -136,6 +135,18 @@ public class CommonUtil {
return new String(str.getBytes(charSet), SERVER_CHARSET);
}
/**
* 将字符串转换为Ftp服务器默认的ISO-8859-1编码
*
* @param charSet 字符串编码
* @param str 需要转换的字符串
* @return 转换后的字符串
*/
public static String convertSFtpChar(String charSet, String str)
throws UnsupportedEncodingException {
return new String(str.getBytes(), charSet);
}
/**
* 获取某包下所有类
*
@@ -481,7 +492,6 @@ public class CommonUtil {
}
}
/**
* 校验文件MD5码
*/

9
PublicComponent/src/main/java/com/arialyy/aria/util/ComponentUtil.java
View File

@@ -40,6 +40,7 @@ public class ComponentUtil {
public static final int COMPONENT_TYPE_HTTP = 1;
public static final int COMPONENT_TYPE_FTP = 2;
public static final int COMPONENT_TYPE_M3U8 = 3;
public static final int COMPONENT_TYPE_SFTP = 4;
private String TAG = CommonUtil.getClassName(getClass());
private static volatile ComponentUtil INSTANCE = null;
@@ -82,6 +83,10 @@ public class ComponentUtil {
className = "com.arialyy.aria.http.HttpTaskOption";
errorStr = "http插件不存在，请添加http插件";
break;
case COMPONENT_TYPE_SFTP:
className = "com.arialyy.aria.sftp.SFtpTaskOption";
errorStr = "sftp插件不存在，请添加sftp插件";
break;
}
try {
@@ -130,6 +135,9 @@ public class ComponentUtil {
case ITaskWrapper.D_SFTP:
className = "com.arialyy.aria.sftp.download.SFtpDLoaderUtil";
break;
case ITaskWrapper.U_SFTP:
className = "com.arialyy.aria.sftp.upload.SFtpULoaderUtil";
break;
}
if (className == null) {
ALog.e(TAG, "不识别的类名：" + className);
@@ -178,6 +186,7 @@ public class ComponentUtil {
break;
case ITaskWrapper.U_FTP:
case ITaskWrapper.U_HTTP:
case ITaskWrapper.U_SFTP:
className = "com.arialyy.aria.core.listener.BaseUListener";
break;
case ITaskWrapper.DG_HTTP:

31
PublicComponent/src/main/java/com/arialyy/aria/util/FileUtil.java
View File

@@ -37,17 +37,13 @@ import java.io.IOException;
import java.io.InputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.SequenceInputStream;
import java.io.Serializable;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.nio.ByteBuffer;
import java.nio.channels.Channels;
import java.nio.channels.FileChannel;
import java.nio.channels.ReadableByteChannel;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Enumeration;
import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.List;
@@ -342,21 +338,12 @@ public class FileUtil {
return false;
}
streams.add(new FileInputStream(subPath));
FileInputStream fis = new FileInputStream(subPath);
FileChannel fic = fis.getChannel();
foc.transferFrom(fic, fileLen, f.length());
fileLen += f.length();
fis.close();
}
Enumeration<FileInputStream> en = Collections.enumeration(streams);
SequenceInputStream sis = new SequenceInputStream(en);
ReadableByteChannel fic = Channels.newChannel(sis);
//ByteBuffer bf = ByteBuffer.allocate(8196);
//while (fic.read(bf) != -1) {
// bf.flip();
// foc.write(bf);
// bf.compact();
//}
foc.transferFrom(fic, 0, fileLen);
fic.close();
sis.close();
ALog.d(TAG, String.format("合并文件耗时：%sms", (System.currentTimeMillis() - startTime)));
return true;
} catch (IOException e) {
@@ -405,7 +392,6 @@ public class FileUtil {
int i = 0;
int threadNum = subPaths.size();
long tempLen = targetFileSize / threadNum;
ALog.d(TAG, "fileSize = " + targetFileSize);
for (String subPath : subPaths) {
File f = new File(subPath);
if (!f.exists()) {
@@ -421,14 +407,13 @@ public class FileUtil {
long blockLen = i == (threadNum - 1) ? targetFileSize - tempLen * i : tempLen;
FileInputStream fis = new FileInputStream(subPath);
FileChannel fic = fis.getChannel();
ALog.d(TAG, "blcokLen = " + blockLen);
long rLen = foc.transferFrom(fic, 0, blockLen);
ALog.d(TAG, "writeLen = " + rLen);
foc.transferFrom(fic, blockLen * i, blockLen);
fis.close();
i ++;
i++;
}
ALog.d(TAG, String.format("合并文件耗时：%sms", (System.currentTimeMillis() - startTime)));
ALog.d(TAG, String.format("合并文件耗时：%sms，合并后的文件长度：%s", (System.currentTimeMillis() - startTime),
file.length()));
return true;
} catch (IOException e) {
e.printStackTrace();