Multipart uploads. Fix #5487.

Changelog.txt

@@ -9,6 +9,8 @@ Changelog. Last modified: $Date$
 - [Feature] Replaced protocol implementation (FTP)
 - [Feature] Connecting to Windows Azure Blob Storage (Azure) (#3938)
 - [Feature] Copy and paste files using menu item to duplicate
+- [Feature] Multipart Uploads with parallelism (S3) (#5487)
+- [Feature] Support new 5TB Object Size Limit (S3)
 - [Bugfix] Limit number of concurrent transfers (#5539)
 - [Bugfix] Dragging into topmost folder in browser (#1945)
 - [Bugfix] ACLs getting dropped when updating metadata (S3) (#5571)

source/ch/cyberduck/core/Path.java

@@ -44,7 +44,9 @@
 import java.net.URLEncoder;
 import java.text.MessageFormat;
 import java.util.Collection;
-import java.util.*;
+import java.util.LinkedHashSet;
+import java.util.Set;
+import java.util.StringTokenizer;
 import java.util.regex.Pattern;
 import java.util.regex.PatternSyntaxException;
 
@@ -444,7 +446,7 @@ public boolean add(Path path) {
 
             @Override
             public boolean addAll(Collection<? extends Path> c) {
-                for(Path path: c) {
+                for(Path path : c) {
                     this.add(path);
                 }
                 return true;
@@ -742,37 +744,42 @@ protected void upload(BandwidthThrottle throttle, StreamListener listener) {
      */
     protected abstract void upload(BandwidthThrottle throttle, StreamListener listener, boolean check);
 
+    protected void upload(OutputStream out, InputStream in, BandwidthThrottle throttle, final StreamListener l) throws IOException {
+        this.upload(out, in, throttle, l, status().getCurrent(), -1);
+    }
+
     /**
      * Will copy from in to out. Will attempt to skip Status#getCurrent
      * from the inputstream but not from the outputstream. The outputstream
      * is asssumed to append to a already existing file if
      * Status#getCurrent > 0
      *
-     * @param in       The stream to read from
      * @param out      The stream to write to
+     * @param in       The stream to read from
      * @param throttle The bandwidth limit
      * @param l        The stream listener to notify about bytes received and sent
+     * @param offset
      * @throws IOResumeException           If the input stream fails to skip the appropriate
      *                                     number of bytes
      * @throws IOException                 Write not completed due to a I/O problem
      * @throws ConnectionCanceledException When transfer is interrupted by user setting the
      *                                     status flag to cancel.
      */
-    protected void upload(OutputStream out, InputStream in, BandwidthThrottle throttle, final StreamListener l) throws IOException {
+    protected void upload(OutputStream out, InputStream in, BandwidthThrottle throttle, final StreamListener l, long offset, final long limit) throws IOException {
         if(log.isDebugEnabled()) {
             log.debug("upload(" + out.toString() + ", " + in.toString());
         }
         this.getSession().message(MessageFormat.format(Locale.localizedString("Uploading {0}", "Status"),
                 this.getName()));
 
-        if(status().isResume()) {
-            long skipped = in.skip(status().getCurrent());
+        if(offset > 0) {
+            long skipped = in.skip(offset);
             log.info("Skipping " + skipped + " bytes");
             if(skipped < status().getCurrent()) {
                 throw new IOResumeException("Skipped " + skipped + " bytes instead of " + status().getCurrent());
             }
         }
-        this.transfer(in, new ThrottledOutputStream(out, throttle), l);
+        this.transfer(in, new ThrottledOutputStream(out, throttle), l, limit);
     }
 
     /**
@@ -836,7 +843,7 @@ public void bytesReceived(long bytes) {
                 }
             }
         };
-        this.transfer(new ThrottledInputStream(in, throttle), out, listener);
+        this.transfer(new ThrottledInputStream(in, throttle), out, listener, -1);
     }
 
     /**
@@ -845,25 +852,36 @@ public void bytesReceived(long bytes) {
      * @param in       The stream to read from
      * @param out      The stream to write to
      * @param listener The stream listener to notify about bytes received and sent
+     * @param limit
      * @throws IOException                 Write not completed due to a I/O problem
      * @throws ConnectionCanceledException When transfer is interrupted by user setting the
      *                                     status flag to cancel.
      */
-    private void transfer(InputStream in, OutputStream out, StreamListener listener) throws IOException {
+    private void transfer(InputStream in, OutputStream out, StreamListener listener, final long limit) throws IOException {
         byte[] chunk = new byte[CHUNKSIZE];
-        long bytesTransferred = status().getCurrent();
+        long bytesTransferred = 0;
         while(!status().isCanceled()) {
             int read = in.read(chunk, 0, CHUNKSIZE);
             listener.bytesReceived(read);
             if(-1 == read) {
+                log.debug("End of file reached");
                 // End of file
                 status().setComplete(true);
                 break;
             }
             out.write(chunk, 0, read);
             listener.bytesSent(read);
+            status().addCurrent(read);
             bytesTransferred += read;
-            status().setCurrent(bytesTransferred);
+            if(limit == bytesTransferred) {
+                log.debug("Limit reached reading from stream:" + limit);
+                // Part reached
+                if(0 == in.available()) {
+                    // End of file
+                    status().setComplete(true);
+                }
+                break;
+            }
         }
         out.flush();
         if(status().isCanceled()) {

source/ch/cyberduck/core/Preferences.java

@@ -523,7 +523,7 @@ protected void setDefaults() {
          * If set calculate MD5 sum of uploaded file and set metadata header Content-MD5
          */
         defaults.put("s3.upload.metadata.md5", String.valueOf(false));
-        defaults.put("cf.upload.metadata.md5", String.valueOf(false));
+        defaults.put("s3.upload.concurency", String.valueOf(10));
         /**
          * A prefix to apply to log file names
          */
@@ -537,14 +537,15 @@ protected void setDefaults() {
          * Default metadata for uploads. Format must be "key1=value1 key2=value2"
          */
         defaults.put("s3.metadata.default", "");
-        defaults.put("cf.metadata.default", "");
-        defaults.put("azure.metadata.default", "");
 
-        defaults.put("webdav.followRedirects", String.valueOf(true));
+        defaults.put("azure.metadata.default", "");
 
         defaults.put("cf.authentication.host", "auth.api.rackspacecloud.com");
         defaults.put("cf.authentication.context", "/v1.0");
 
+        defaults.put("cf.upload.metadata.md5", String.valueOf(false));
+        defaults.put("cf.metadata.default", "");
+
         //doc	Microsoft Word
         //html	HTML Format
         //odt	Open Document Format
@@ -589,6 +590,7 @@ protected void setDefaults() {
          * NTLM Windows Domain
          */
         defaults.put("webdav.ntlm.domain", StringUtils.EMPTY);
+        defaults.put("webdav.followRedirects", String.valueOf(true));
 
         /**
          * Maximum concurrent connections to the same host

source/ch/cyberduck/core/Status.java

@@ -200,6 +200,10 @@ public void setCurrent(long current) {
         this.current = current;
     }
 
+    public void addCurrent(long transferred) {
+        this.current += transferred;
+    }
+
     /**
      * A state variable to mark this path if the path is explicitly selected
      * for inclusion in the transfer prompt