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-15 20:28:10 +08:00
parent 6f53235807
commit 50b265f22a
37 changed files with 934 additions and 257 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -33,6 +33,7 @@ import com.arialyy.aria.core.config.UploadConfig;
import com.arialyy.aria.core.config.XMLReader;
import com.arialyy.aria.util.ALog;
import com.arialyy.aria.util.CommonUtil;
import com.arialyy.aria.util.FileUtil;
import java.io.File;
import java.io.IOException;
import javax.xml.parsers.ParserConfigurationException;
@@ -136,7 +137,7 @@ public class AriaConfig {
.addTransportType(NetworkCapabilities.TRANSPORT_WIFI)
.addTransportType(NetworkCapabilities.TRANSPORT_CELLULAR)
.build();
if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.LOLLIPOP){
if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.LOLLIPOP) {
cm.registerNetworkCallback(request, new ConnectivityManager.NetworkCallback() {
@Override public void onLost(Network network) {
@@ -152,7 +153,6 @@ public class AriaConfig {
}
});
}
}
public boolean isConnectedNet() {
@@ -179,7 +179,7 @@ public class AriaConfig {
if (file.exists()) {
file.delete();
}
CommonUtil.createFileFormInputStream(APP.getAssets().open("aria_config.xml"),
FileUtil.createFileFormInputStream(APP.getAssets().open("aria_config.xml"),
file.getPath());
if (!CommonUtil.checkMD5(md5Code, file) || !Configuration.getInstance().configExists()) {
loadConfig();
@@ -204,7 +204,7 @@ public class AriaConfig {
SAXParserFactory factory = SAXParserFactory.newInstance();
SAXParser parser = factory.newSAXParser();
parser.parse(APP.getAssets().open("aria_config.xml"), helper);
CommonUtil.createFileFormInputStream(APP.getAssets().open("aria_config.xml"),
FileUtil.createFileFormInputStream(APP.getAssets().open("aria_config.xml"),
APP.getFilesDir().getPath() + Configuration.XML_FILE);
} catch (ParserConfigurationException | IOException | SAXException e) {
ALog.e(TAG, e.toString());

18
PublicComponent/src/main/java/com/arialyy/aria/core/IdEntity.java
View File

@@ -21,7 +21,7 @@ package com.arialyy.aria.core;
public class IdEntity {
/**
* 私钥证书内容（非路径）
* 私钥证书路径
*/
public String prvKey;
@@ -31,17 +31,27 @@ public class IdEntity {
public String prvPass;
/**
* 公钥证书内容（非路径）
* 公钥证书路径
*/
public String pubKey;
/**
* 私钥证书路径
* knowhost文件路径
*/
public String knowHost;
/**
* ca 证书密码
*/
public String storePass;
/**
* ca证书路径
*/
public String storePath;
/**
* 私钥别名
* ca证书别名
*/
public String keyAlias;
}

2
PublicComponent/src/main/java/com/arialyy/aria/core/group/SubRecordHandler.java
View File

@@ -58,7 +58,7 @@ public class SubRecordHandler extends RecordHandler {
record.threadRecords = new ArrayList<>();
record.threadNum = threadNum;
record.isBlock = false;
record.taskType = getEntity().getTaskType();
record.taskType = getWrapper().getRequestType();
record.isGroupRecord = true;
if (getEntity() instanceof DownloadEntity) {
record.dGroupHash = ((DownloadEntity) getEntity()).getGroupHash();

3
PublicComponent/src/main/java/com/arialyy/aria/core/listener/BaseListener.java
View File

@@ -136,7 +136,8 @@ public abstract class BaseListener<ENTITY extends AbsEntity, TASK_WRAPPER extend
mEntity.setConvertSpeed(CommonUtil.formatFileSize(speed < 0 ? 0 : speed) + "/s");
}
mEntity.setSpeed(speed < 0 ? 0 : speed);
if (mTaskWrapper.getRequestType() != ITaskWrapper.M3U8_VOD) {
int taskType = mTaskWrapper.getRequestType();
if (taskType != ITaskWrapper.M3U8_VOD && taskType != ITaskWrapper.M3U8_LIVE) {
mEntity.setPercent((int) (mEntity.getFileSize() <= 0 ? 0
: mEntity.getCurrentProgress() * 100 / mEntity.getFileSize()));
}

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

@@ -22,6 +22,7 @@ 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;
@@ -229,7 +230,9 @@ 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 = FileUtil.mergeFile(mTaskRecord.filePath, partPath);
boolean isSuccess = mTaskRecord.taskType == ITaskWrapper.D_SFTP ?
FileUtil.mergeSFtpFile(mTaskRecord.filePath, partPath, mTaskRecord.fileLength)
: FileUtil.mergeFile(mTaskRecord.filePath, partPath);
if (isSuccess) {
for (String pp : partPath) {
File f = new File(pp);

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

@@ -327,6 +327,9 @@ public class ThreadTask implements IThreadTask, IThreadTaskObserver {
fail(mRangeProgress, e, needRetry);
}
/**
* @param len 新增的长度
*/
@Override
public synchronized void updateProgress(long len) {
mRangeProgress += len;

28
PublicComponent/src/main/java/com/arialyy/aria/exception/AriaException.java Normal file
View File

@@ -0,0 +1,28 @@
/*
* 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.exception;
public class AriaException extends BaseException {
private static final String ARIA_NET_EXCEPTION = "Aria Exception:";
public AriaException(String tag, String message) {
super(tag, String.format("%s%s", ARIA_NET_EXCEPTION, message));
}
public AriaException(String tag, String message, Exception e) {
super(tag, message, e);
}
}

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

@@ -481,24 +481,6 @@ public class CommonUtil {
}
}
/**
* 通过流创建文件
*/
public static void createFileFormInputStream(InputStream is, String path) {
try {
FileOutputStream fos = new FileOutputStream(path);
byte[] buf = new byte[1024];
int len;
while ((len = is.read(buf)) > 0) {
fos.write(buf, 0, len);
}
is.close();
fos.flush();
fos.close();
} catch (IOException e) {
e.printStackTrace();
}
}
/**
* 校验文件MD5码

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

@@ -127,6 +127,9 @@ public class ComponentUtil {
case ITaskWrapper.DG_HTTP:
className = "com.arialyy.aria.http.download.HttpDGLoaderUtil";
break;
case ITaskWrapper.D_SFTP:
className = "com.arialyy.aria.sftp.download.SFtpDLoaderUtil";
break;
}
if (className == null) {
ALog.e(TAG, "不识别的类名：" + className);
@@ -170,6 +173,7 @@ public class ComponentUtil {
break;
case ITaskWrapper.D_FTP:
case ITaskWrapper.D_HTTP:
case ITaskWrapper.D_SFTP:
className = "com.arialyy.aria.core.listener.BaseDListener";
break;
case ITaskWrapper.U_FTP:

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

@@ -34,6 +34,7 @@ import java.io.FileOutputStream;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.IOException;
import java.io.InputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.SequenceInputStream;
@@ -63,6 +64,27 @@ public class FileUtil {
private static final String EXTERNAL_STORAGE_PATH =
Environment.getExternalStorageDirectory().getPath();
/**
* 通过流创建文件
*
* @param dest 输出路径
*/
public static void createFileFormInputStream(InputStream is, String dest) {
try {
FileOutputStream fos = new FileOutputStream(dest);
byte[] buf = new byte[1024];
int len;
while ((len = is.read(buf)) > 0) {
fos.write(buf, 0, len);
}
is.close();
fos.flush();
fos.close();
} catch (IOException e) {
e.printStackTrace();
}
}
/**
* 获取m3u8 ts文件的缓存目录。
* 缓存文件夹格式：父文件夹/.文件名_码率
@@ -295,6 +317,7 @@ public class FileUtil {
File file = new File(targetPath);
FileOutputStream fos = null;
FileChannel foc = null;
long startTime = System.currentTimeMillis();
try {
if (file.exists() && file.isDirectory()) {
ALog.w(TAG, String.format("路径【%s】是文件夹，将删除该文件夹", targetPath));
@@ -307,6 +330,7 @@ public class FileUtil {
fos = new FileOutputStream(targetPath);
foc = fos.getChannel();
List<FileInputStream> streams = new LinkedList<>();
long fileLen = 0;
for (String subPath : subPaths) {
File f = new File(subPath);
if (!f.exists()) {
@@ -319,18 +343,92 @@ public class FileUtil {
return false;
}
streams.add(new FileInputStream(subPath));
fileLen += f.length();
}
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();
}
//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) {
e.printStackTrace();
} finally {
try {
if (foc != null) {
foc.close();
}
if (fos != null) {
fos.close();
}
} catch (IOException e) {
e.printStackTrace();
}
}
return false;
}
/**
* 合并sftp的分块文件，sftp的分块可能会超出规定的长度，因此需要使用本方法
*
* @param targetPath 目标文件
* @param subPaths 碎片文件路径
* @param targetFileSize 文件长度
* @return {@code true} 合并成功，{@code false}合并失败
*/
public static boolean mergeSFtpFile(String targetPath, List<String> subPaths,
long targetFileSize) {
File file = new File(targetPath);
FileOutputStream fos = null;
FileChannel foc = null;
long startTime = System.currentTimeMillis();
try {
if (file.exists() && file.isDirectory()) {
ALog.w(TAG, String.format("路径【%s】是文件夹，将删除该文件夹", targetPath));
FileUtil.deleteDir(file);
}
if (!file.exists()) {
FileUtil.createFile(file);
}
fos = new FileOutputStream(targetPath);
foc = fos.getChannel();
List<FileInputStream> streams = new LinkedList<>();
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()) {
ALog.d(TAG, String.format("合并文件失败，文件【%s】不存在", subPath));
for (FileInputStream fis : streams) {
fis.close();
}
streams.clear();
return false;
}
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);
fis.close();
i ++;
}
ALog.d(TAG, String.format("合并文件耗时：%sms", (System.currentTimeMillis() - startTime)));
return true;
} catch (IOException e) {
e.printStackTrace();