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

修复ftp服务器无法响应abor命令导致的无法停止上传的问题

Browse Source 
修复ftp上传时，服务器有长度为0的文件导致上传失败的问题
修复下载任务和上传任务的文件路径是同一个时，导致的记录混乱问题
This commit is contained in:
laoyuyu
2019-12-17 22:04:21 +08:00
parent 0d93953cd9
commit 27c889e171
43 changed files with 295 additions and 145 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
FtpComponent/src/main/java/com/arialyy/aria/ftp/AbsFtpInfoThread.java
View File

@@ -50,7 +50,7 @@ import javax.net.ssl.SSLContext;
public abstract class AbsFtpInfoThread<ENTITY extends AbsEntity, TASK_WRAPPER extends AbsTaskWrapper<ENTITY>>
implements Runnable {
private final String TAG = CommonUtil.getClassName(getClass());
protected final String TAG = CommonUtil.getClassName(getClass());
protected ENTITY mEntity;
protected TASK_WRAPPER mTaskWrapper;
protected FtpTaskOption mTaskOption;
@@ -123,7 +123,7 @@ public abstract class AbsFtpInfoThread<ENTITY extends AbsEntity, TASK_WRAPPER ex
// 处理拦截功能
if (!onInterceptor(client, files)) {
closeClient(client);
ALog.d(TAG, "拦截器处理完成任务，任务将不再执行");
ALog.d(TAG, "拦截器处理完成任务");
return;
}
@@ -390,8 +390,7 @@ public abstract class AbsFtpInfoThread<ENTITY extends AbsEntity, TASK_WRAPPER ex
needRetry = needRetry && !CheckUtil.ftpIsBadRequest(client.getReplyCode());
}
//mCallback.onFail(mEntity, new AriaIOException(TAG, msg), needRetry);
mCallback.onFail(mEntity, new AriaIOException(TAG, msg, e), false);
mCallback.onFail(mEntity, new AriaIOException(TAG, msg), needRetry);
}
}

4
FtpComponent/src/main/java/com/arialyy/aria/ftp/FtpRecordAdapter.java
View File

@@ -58,7 +58,7 @@ public class FtpRecordAdapter extends AbsRecordHandlerAdapter {
tr.threadId = threadId;
tr.startLocation = startL;
tr.isComplete = false;
tr.threadType = TaskRecord.TYPE_HTTP_FTP;
tr.threadType = getWrapper().getEntity().getTaskType();
//最后一个线程的结束位置即为文件的总长度
if (threadId == (record.threadNum - 1)) {
endL = getEntity().getFileSize();
@@ -81,7 +81,7 @@ public class FtpRecordAdapter extends AbsRecordHandlerAdapter {
} else {
record.isBlock = false;
}
record.taskType = TaskRecord.TYPE_HTTP_FTP;
record.taskType = getWrapper().getEntity().getTaskType();
record.isGroupRecord = getEntity().isGroupChild();
if (record.isGroupRecord) {
if (getEntity() instanceof DownloadEntity) {

4
FtpComponent/src/main/java/com/arialyy/aria/ftp/download/FtpDThreadTaskAdapter.java
View File

@@ -180,7 +180,9 @@ final class FtpDThreadTaskAdapter extends BaseFtpThreadTaskAdapter {
try {
file =
new BufferedRandomAccessFile(getThreadConfig().tempFile, "rwd", getTaskConfig().getBuffSize());
file.seek(getThreadRecord().startLocation);
if (getThreadRecord().startLocation > 0){
file.seek(getThreadRecord().startLocation);
}
byte[] buffer = new byte[getTaskConfig().getBuffSize()];
int len;
while (getThreadTask().isLive() && (len = is.read(buffer)) != -1) {

22
FtpComponent/src/main/java/com/arialyy/aria/ftp/upload/FtpUFileInfoThread.java
View File

@@ -22,11 +22,11 @@ import com.arialyy.aria.core.TaskRecord;
import com.arialyy.aria.core.ThreadRecord;
import com.arialyy.aria.core.common.CompleteInfo;
import com.arialyy.aria.core.inf.OnFileInfoCallback;
import com.arialyy.aria.core.processor.FtpInterceptHandler;
import com.arialyy.aria.core.processor.IFtpUploadInterceptor;
import com.arialyy.aria.core.upload.UTaskWrapper;
import com.arialyy.aria.core.upload.UploadEntity;
import com.arialyy.aria.ftp.AbsFtpInfoThread;
import com.arialyy.aria.core.processor.FtpInterceptHandler;
import com.arialyy.aria.core.processor.IFtpUploadInterceptor;
import com.arialyy.aria.util.ALog;
import com.arialyy.aria.util.CommonUtil;
import com.arialyy.aria.util.DbDataHelper;
@@ -39,7 +39,6 @@ import java.util.List;
* 单任务上传远程服务器文件信息
*/
class FtpUFileInfoThread extends AbsFtpInfoThread<UploadEntity, UTaskWrapper> {
private static final String TAG = "FtpUploadFileInfoThread";
static final int CODE_COMPLETE = 0xab1;
private boolean isComplete = false;
private String remotePath;
@@ -59,10 +58,10 @@ class FtpUFileInfoThread extends AbsFtpInfoThread<UploadEntity, UTaskWrapper> {
@Override protected boolean onInterceptor(FTPClient client, FTPFile[] ftpFiles) {
// 旧任务将不做处理，否则断点续传上传将失效
if (!mTaskWrapper.isNewTask()) {
ALog.d(TAG, "任务是旧任务，忽略该拦截器");
return true;
}
//if (!mTaskWrapper.isNewTask()) {
// ALog.d(TAG, "任务是旧任务，忽略该拦截器");
// return true;
//}
try {
IFtpUploadInterceptor interceptor = mTaskOption.getUploadInterceptor();
if (interceptor != null) {
@@ -78,7 +77,7 @@ class FtpUFileInfoThread extends AbsFtpInfoThread<UploadEntity, UTaskWrapper> {
FtpInterceptHandler interceptHandler = interceptor.onIntercept(mEntity, files);
/*
处理远端有同名文件的情况
* 处理远端有同名文件的情况
*/
if (files.contains(mEntity.getFileName())) {
if (interceptHandler.isCoverServerFile()) {
@@ -94,6 +93,7 @@ class FtpUFileInfoThread extends AbsFtpInfoThread<UploadEntity, UTaskWrapper> {
+ "/"
+ interceptHandler.getNewFileName();
mTaskOption.setNewFileName(interceptHandler.getNewFileName());
closeClient(client);
run();
return false;
@@ -123,6 +123,9 @@ class FtpUFileInfoThread extends AbsFtpInfoThread<UploadEntity, UTaskWrapper> {
if (ftpFile.getSize() == mEntity.getFileSize()) {
isComplete = true;
ALog.d(TAG, "FTP服务器上已存在该文件【" + ftpFile.getName() + "】");
} else if (ftpFile.getSize() == 0) {
mTaskWrapper.setNewTask(true);
ALog.d(TAG, "FTP服务器上已存在该文件【" + ftpFile.getName() + "】，但文件长度为0，重新上传该文件");
} else {
ALog.w(TAG, "FTP服务器已存在未完成的文件【"
+ ftpFile.getName()
@@ -135,7 +138,8 @@ class FtpUFileInfoThread extends AbsFtpInfoThread<UploadEntity, UTaskWrapper> {
mTaskWrapper.setNewTask(false);
// 修改记录
TaskRecord record = DbDataHelper.getTaskRecord(mTaskWrapper.getKey());
TaskRecord record = DbDataHelper.getTaskRecord(mTaskWrapper.getKey(),
mTaskWrapper.getEntity().getTaskType());
if (record == null) {
record = new TaskRecord();
record.fileName = mEntity.getFileName();

31
FtpComponent/src/main/java/com/arialyy/aria/ftp/upload/FtpUThreadTaskAdapter.java
View File

@@ -34,6 +34,7 @@ import java.io.UnsupportedEncodingException;
*/
class FtpUThreadTaskAdapter extends BaseFtpThreadTaskAdapter {
private String dir, remotePath;
private boolean storeFail = false;
FtpUThreadTaskAdapter(SubThreadConfig config) {
super(config);
@@ -67,8 +68,7 @@ class FtpUThreadTaskAdapter extends BaseFtpThreadTaskAdapter {
file =
new BufferedRandomAccessFile(getThreadConfig().tempFile, "rwd",
getTaskConfig().getBuffSize());
if (getThreadRecord().startLocation != 0) {
//file.skipBytes((int) getThreadConfig().START_LOCATION);
if (getThreadRecord().startLocation > 0) {
file.seek(getThreadRecord().startLocation);
}
boolean complete = upload(client, file);
@@ -79,6 +79,7 @@ class FtpUThreadTaskAdapter extends BaseFtpThreadTaskAdapter {
String.format("任务【%s】线程__%s__上传完毕", getEntity().getKey(), getThreadRecord().threadId));
complete();
} catch (IOException e) {
e.printStackTrace();
fail(new AriaIOException(TAG,
String.format("上传失败，filePath: %s, uploadUrl: %s", getEntity().getFilePath(),
getThreadConfig().url)), true);
@@ -118,10 +119,11 @@ class FtpUThreadTaskAdapter extends BaseFtpThreadTaskAdapter {
*/
private boolean upload(final FTPClient client, final BufferedRandomAccessFile bis)
throws IOException {
final FtpFISAdapter fa = new FtpFISAdapter(bis);
storeFail = false;
try {
ALog.d(TAG, String.format("remotePath: %s", remotePath));
client.storeFile(remotePath, new FtpFISAdapter(bis), new OnFtpInputStreamListener() {
client.storeFile(remotePath, fa, new OnFtpInputStreamListener() {
boolean isStoped = false;
@Override public void onFtpInputStream(FTPClient client, long totalBytesTransferred,
@@ -137,23 +139,30 @@ class FtpUThreadTaskAdapter extends BaseFtpThreadTaskAdapter {
progress(bytesTransferred);
} catch (IOException e) {
e.printStackTrace();
storeFail = true;
try {
fa.close();
} catch (IOException e1) {
e1.printStackTrace();
}
closeClient(client);
}
}
});
} catch (IOException e) {
String msg = String.format("文件上传错误，错误码为：%s, msg：%s, filePath: %s", client.getReplyCode(),
client.getReplyString(), getEntity().getFilePath());
if (client.isConnected()) {
client.disconnect();
}
closeClient(client);
if (e.getMessage().contains("AriaIOException caught while copying")) {
e.printStackTrace();
} else {
fail(new AriaIOException(TAG, msg, e), true);
fail(new AriaIOException(TAG, msg, e), !storeFail);
}
return false;
}
if (storeFail) {
return false;
}
int reply = client.getReplyCode();
if (!FTPReply.isPositiveCompletion(reply)) {
if (reply != FTPReply.TRANSFER_ABORTED) {
@@ -161,9 +170,7 @@ class FtpUThreadTaskAdapter extends BaseFtpThreadTaskAdapter {
String.format("文件上传错误，错误码为：%s, msg：%s, filePath: %s", reply, client.getReplyString(),
getEntity().getFilePath())), false);
}
if (client.isConnected()) {
client.disconnect();
}
closeClient(client);
return false;
}
return true;