修复一个表创建失败的问题 https://github.com/AriaLyy/Aria/issues/570

修复一个非分块模式下导致下载失败的问题 https://github.com/AriaLyy/Aria/issues/571 修复一个服务器端无法创建socket连接，却没有返回码导致客户端卡住的问题 https://github.com/AriaLyy/Aria/issues/569 修复文件删除后，组合任务没有重新下载的问题 https://github.com/AriaLyy/Aria/issues/574 优化缓存队列和执行队列
2019-12-22 11:30:29 +08:00
parent 27c889e171
commit 8dc16ce302
38 changed files with 563 additions and 718 deletions
--- a/FtpComponent/src/main/java/aria/apache/commons/net/ftp/FTPClient.java
+++ b/FtpComponent/src/main/java/aria/apache/commons/net/ftp/FTPClient.java
@@ -3662,7 +3662,6 @@ public class FTPClient extends FTP implements Configurable {
    private long time = System.currentTimeMillis();
    private int notAcked;
    private OnFtpInputStreamListener listener;
-    private long lTime = time;

    CSL(FTPClient parent, long idleTime, int maxWait) throws SocketException {
      this(parent, idleTime, maxWait, null);
@@ -3696,12 +3695,9 @@ public class FTPClient extends FTP implements Configurable {

        time = now;
      }
-      //if (now - lTime > 100) {
      if (listener != null) {
        listener.onFtpInputStream(parent, totalBytesTransferred, bytesTransferred, streamSize);
      }
-      //lTime = now;
-      //}
    }

    void cleanUp() throws IOException {
--- a/FtpComponent/src/main/java/com/arialyy/aria/ftp/BaseFtpThreadTaskAdapter.java
+++ b/FtpComponent/src/main/java/com/arialyy/aria/ftp/BaseFtpThreadTaskAdapter.java
@@ -117,7 +117,8 @@ public abstract class BaseFtpThreadTaskAdapter extends AbsThreadTaskAdapter {
        }
      }
      client.setControlEncoding(charSet);
-      client.setDataTimeout(getTaskConfig().getIOTimeOut());
+      //client.setDataTimeout(getTaskConfig().getIOTimeOut());
+      client.setDataTimeout(1000);
      client.setConnectTimeout(getTaskConfig().getConnectTimeOut());
      if (mTaskOption.getConnMode() == FtpConnectionMode.DATA_CONNECTION_MODE_ACTIVITY) {
        client.enterLocalActiveMode();
--- a/FtpComponent/src/main/java/com/arialyy/aria/ftp/upload/FtpUFileInfoThread.java
+++ b/FtpComponent/src/main/java/com/arialyy/aria/ftp/upload/FtpUFileInfoThread.java
@@ -42,10 +42,6 @@ class FtpUFileInfoThread extends AbsFtpInfoThread<UploadEntity, UTaskWrapper> {
  static final int CODE_COMPLETE = 0xab1;
  private boolean isComplete = false;
  private String remotePath;
-  /**
-   * true 使用拦截器，false 不使用拦截器
-   */
-  private boolean useInterceptor = false;

  FtpUFileInfoThread(UTaskWrapper taskEntity, OnFileInfoCallback callback) {
    super(taskEntity, callback);
@@ -65,7 +61,9 @@ class FtpUFileInfoThread extends AbsFtpInfoThread<UploadEntity, UTaskWrapper> {
    try {
      IFtpUploadInterceptor interceptor = mTaskOption.getUploadInterceptor();
      if (interceptor != null) {
-        useInterceptor = true;
+        /**
+         * true 使用拦截器，false 不使用拦截器
+         */
        List<String> files = new ArrayList<>();
        for (FTPFile ftpFile : ftpFiles) {
          if (ftpFile.isDirectory()) {
@@ -118,7 +116,7 @@ class FtpUFileInfoThread extends AbsFtpInfoThread<UploadEntity, UTaskWrapper> {
   */
  @Override protected void handleFile(String remotePath, FTPFile ftpFile) {
    super.handleFile(remotePath, ftpFile);
-    if (ftpFile != null && !useInterceptor) {
+    if (ftpFile != null) {
      //远程文件已完成
      if (ftpFile.getSize() == mEntity.getFileSize()) {
        isComplete = true;
--- a/FtpComponent/src/main/java/com/arialyy/aria/ftp/upload/FtpUThreadTaskAdapter.java
+++ b/FtpComponent/src/main/java/com/arialyy/aria/ftp/upload/FtpUThreadTaskAdapter.java
@@ -28,6 +28,8 @@ import com.arialyy.aria.util.BufferedRandomAccessFile;
 import com.arialyy.aria.util.CommonUtil;
 import java.io.IOException;
 import java.io.UnsupportedEncodingException;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;

 /**
 * Created by Aria.Lao on 2017/7/28. D_FTP 单线程上传任务，需要FTP 服务器给用户打开append和write的权限
@@ -35,13 +37,16 @@ import java.io.UnsupportedEncodingException;
 class FtpUThreadTaskAdapter extends BaseFtpThreadTaskAdapter {
  private String dir, remotePath;
  private boolean storeFail = false;
+  private ScheduledThreadPoolExecutor timer;
+  private FTPClient client = null;
+  private boolean isTimeOut = true;
+  private FtpFISAdapter fa;

  FtpUThreadTaskAdapter(SubThreadConfig config) {
    super(config);
  }

  @Override protected void handlerThreadTask() {
-    FTPClient client = null;
    BufferedRandomAccessFile file = null;
    try {
      ALog.d(TAG,
@@ -71,7 +76,7 @@ class FtpUThreadTaskAdapter extends BaseFtpThreadTaskAdapter {
      if (getThreadRecord().startLocation > 0) {
        file.seek(getThreadRecord().startLocation);
      }
-      boolean complete = upload(client, file);
+      boolean complete = upload(file);
      if (!complete || getThreadTask().isBreak()) {
        return;
      }
@@ -94,6 +99,7 @@ class FtpUThreadTaskAdapter extends BaseFtpThreadTaskAdapter {
        e.printStackTrace();
      }
      closeClient(client);
+      closeTimer();
    }
  }

@@ -112,15 +118,46 @@ class FtpUThreadTaskAdapter extends BaseFtpThreadTaskAdapter {
            String.format("%s/%s", mTaskOption.getUrlEntity().remotePath, fileName));
  }

+  /**
+   * 启动监听定时器，当网络断开时，如果该任务的FTP服务器的传输线程没有断开，当客户端重新连接时，客户端将无法发送数据到服务端
+   * 每隔10s检查一次。
+   */
+  private void startTimer() {
+    timer = new ScheduledThreadPoolExecutor(1);
+    timer.scheduleWithFixedDelay(new Runnable() {
+      @Override public void run() {
+        try {
+          if (isTimeOut) {
+            fail(new AriaIOException(TAG, "socket连接失败，该问题一般出现于网络断开，客户端重新连接，"
+                + "但是服务器端无法创建socket缺没有返回错误码的情况。"), false);
+          }
+          if (fa != null) {
+            fa.close();
+          }
+          isTimeOut = true;
+        } catch (IOException e) {
+          e.printStackTrace();
+        }
+      }
+    }, 10, 10, TimeUnit.SECONDS);
+  }
+
+  private void closeTimer() {
+    if (timer != null && !timer.isShutdown()) {
+      timer.shutdown();
+    }
+  }
+
  /**
   * 上传
   *
   * @return {@code true}上传成功、{@code false} 上传失败
   */
-  private boolean upload(final FTPClient client, final BufferedRandomAccessFile bis)
+  private boolean upload(final BufferedRandomAccessFile bis)
      throws IOException {
-    final FtpFISAdapter fa = new FtpFISAdapter(bis);
+    fa = new FtpFISAdapter(bis);
    storeFail = false;
+    startTimer();
    try {
      ALog.d(TAG, String.format("remotePath: %s", remotePath));
      client.storeFile(remotePath, fa, new OnFtpInputStreamListener() {
@@ -129,6 +166,7 @@ class FtpUThreadTaskAdapter extends BaseFtpThreadTaskAdapter {
        @Override public void onFtpInputStream(FTPClient client, long totalBytesTransferred,
            int bytesTransferred, long streamSize) {
          try {
+            isTimeOut = false;
            if (getThreadTask().isBreak() && !isStoped) {
              isStoped = true;
              client.abor();
@@ -150,9 +188,10 @@ class FtpUThreadTaskAdapter extends BaseFtpThreadTaskAdapter {
        }
      });
    } catch (IOException e) {
-      String msg = String.format("文件上传错误，错误码为：%s, msg：%s, filePath: %s", client.getReplyCode(),
+      String msg = String.format("文件上传错误，错误码为：%s, msg：%s, filePath: %s", client.getReply(),
          client.getReplyString(), getEntity().getFilePath());
      closeClient(client);
+      e.printStackTrace();
      if (e.getMessage().contains("AriaIOException caught while copying")) {
        e.printStackTrace();
      } else {
@@ -163,7 +202,7 @@ class FtpUThreadTaskAdapter extends BaseFtpThreadTaskAdapter {
    if (storeFail) {
      return false;
    }
-    int reply = client.getReplyCode();
+    int reply = client.getReply();
    if (!FTPReply.isPositiveCompletion(reply)) {
      if (reply != FTPReply.TRANSFER_ABORTED) {
        fail(new AriaIOException(TAG,