Move send/write implementations into parts

Author: Stephane Landelle

api/src/main/java/org/asynchttpclient/RandomAccessBody.java

@@ -1,15 +1,15 @@
 /*
-* Copyright (c) 2010-2012 Sonatype, Inc. All rights reserved.
-*
-* This program is licensed to you under the Apache License Version 2.0,
-* and you may not use this file except in compliance with the Apache License Version 2.0.
-* You may obtain a copy of the Apache License Version 2.0 at http://www.apache.org/licenses/LICENSE-2.0.
-*
-* Unless required by applicable law or agreed to in writing,
-* software distributed under the Apache License Version 2.0 is distributed on an
-* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-* See the Apache License Version 2.0 for the specific language governing permissions and limitations there under.
-*/
+ * Copyright (c) 2010-2012 Sonatype, Inc. All rights reserved.
+ *
+ * This program is licensed to you under the Apache License Version 2.0,
+ * and you may not use this file except in compliance with the Apache License Version 2.0.
+ * You may obtain a copy of the Apache License Version 2.0 at http://www.apache.org/licenses/LICENSE-2.0.
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the Apache License Version 2.0 is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the Apache License Version 2.0 for the specific language governing permissions and limitations there under.
+ */
 
 package org.asynchttpclient;
 
@@ -19,19 +19,20 @@
 /**
  * A request body which supports random access to its contents.
  */
-public interface RandomAccessBody
-        extends Body {
+public interface RandomAccessBody extends Body {
 
     /**
      * Transfers the specified chunk of bytes from this body to the specified channel.
-     *
-     * @param position The zero-based byte index from which to start the transfer, must not be negative.
-     * @param count    The maximum number of bytes to transfer, must not be negative.
-     * @param target   The destination channel to transfer the body chunk to, must not be {@code null}.
+     * 
+     * @param position
+     *            The zero-based byte index from which to start the transfer, must not be negative.
+     * @param count
+     *            The maximum number of bytes to transfer, must not be negative.
+     * @param target
+     *            The destination channel to transfer the body chunk to, must not be {@code null}.
      * @return The non-negative number of bytes actually transferred.
-     * @throws IOException If the body chunk could not be transferred.
+     * @throws IOException
+     *             If the body chunk could not be transferred.
      */
-    long transferTo(long position, long count, WritableByteChannel target)
-            throws IOException;
-
+    long transferTo(long position, long count, WritableByteChannel target) throws IOException;
 }

api/src/main/java/org/asynchttpclient/multipart/AbstractFilePart.java

@@ -15,6 +15,7 @@
  */
 package org.asynchttpclient.multipart;
 
+import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 
 import org.asynchttpclient.util.StandardCharsets;
@@ -85,4 +86,24 @@ public void setStalledTime(long ms) {
     public long getStalledTime() {
         return stalledTime;
     }
+
+    protected byte[] generateFileStart(byte[] boundary) throws IOException {
+        ByteArrayOutputStream out = new ByteArrayOutputStream();
+        OutputStreamPartVisitor visitor = new OutputStreamPartVisitor(out);
+        visitStart(visitor, boundary);
+        visitDispositionHeader(visitor);
+        visitContentTypeHeader(visitor);
+        visitTransferEncodingHeader(visitor);
+        visitContentIdHeader(visitor);
+        visitEndOfHeader(visitor);
+
+        return out.toByteArray();
+    }
+
+    protected byte[] generateFileEnd() throws IOException {
+        ByteArrayOutputStream out = new ByteArrayOutputStream();
+        OutputStreamPartVisitor visitor = new OutputStreamPartVisitor(out);
+        visitEnd(visitor);
+        return out.toByteArray();
+    }
 }

api/src/main/java/org/asynchttpclient/multipart/ByteArrayPart.java

@@ -17,6 +17,7 @@
 
 import java.io.IOException;
 import java.io.OutputStream;
+import java.nio.channels.WritableByteChannel;
 
 public class ByteArrayPart extends AbstractFilePart {
 
@@ -66,4 +67,21 @@ protected long getDataLength() {
     public byte[] getBytes() {
         return bytes;
     }
+
+    @Override
+    public long write(WritableByteChannel target, byte[] boundary) throws IOException {
+        FilePartStallHandler handler = new FilePartStallHandler(getStalledTime(), this);
+
+        try {
+            handler.start();
+
+            long length = MultipartUtils.writeBytesToChannel(target, generateFileStart(boundary));
+            length += MultipartUtils.writeBytesToChannel(target, bytes);
+            length += MultipartUtils.writeBytesToChannel(target, generateFileEnd());
+
+            return length;
+        } finally {
+            handler.completed();
+        }
+    }
 }

api/src/main/java/org/asynchttpclient/multipart/FilePart.java

@@ -20,9 +20,17 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
+import java.io.RandomAccessFile;
+import java.nio.channels.FileChannel;
+import java.nio.channels.WritableByteChannel;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class FilePart extends AbstractFilePart {
 
+    private static final Logger LOGGER = LoggerFactory.getLogger(FilePart.class);
+
     private final File file;
     private final String fileName;
 
@@ -93,4 +101,70 @@ protected long getDataLength() {
     public File getFile() {
         return file;
     }
+
+    @Override
+    public long write(WritableByteChannel target, byte[] boundary) throws IOException {
+        FilePartStallHandler handler = new FilePartStallHandler(getStalledTime(), this);
+
+        handler.start();
+
+        int length = 0;
+
+        length += MultipartUtils.writeBytesToChannel(target, generateFileStart(boundary));
+
+        RandomAccessFile raf = new RandomAccessFile(file, "r");
+        FileChannel fc = raf.getChannel();
+
+        long l = file.length();
+        int fileLength = 0;
+        long nWrite = 0;
+        // FIXME why sync?
+        try {
+            synchronized (fc) {
+                while (fileLength != l) {
+                    if (handler.isFailed()) {
+                        LOGGER.debug("Stalled error");
+                        throw new FileUploadStalledException();
+                    }
+                    try {
+                        nWrite = fc.transferTo(fileLength, l, target);
+
+                        if (nWrite == 0) {
+                            LOGGER.info("Waiting for writing...");
+                            try {
+                                fc.wait(50);
+                            } catch (InterruptedException e) {
+                                LOGGER.trace(e.getMessage(), e);
+                            }
+                        } else {
+                            handler.writeHappened();
+                        }
+                    } catch (IOException ex) {
+                        String message = ex.getMessage();
+
+                        // http://bugs.sun.com/view_bug.do?bug_id=5103988
+                        if (message != null && message.equalsIgnoreCase("Resource temporarily unavailable")) {
+                            try {
+                                fc.wait(1000);
+                            } catch (InterruptedException e) {
+                                LOGGER.trace(e.getMessage(), e);
+                            }
+                            LOGGER.warn("Experiencing NIO issue http://bugs.sun.com/view_bug.do?bug_id=5103988. Retrying");
+                            continue;
+                        } else {
+                            throw ex;
+                        }
+                    }
+                    fileLength += nWrite;
+                }
+            }
+        } finally {
+            handler.completed();
+            raf.close();
+        }
+
+        length += MultipartUtils.writeBytesToChannel(target, generateFileEnd());
+
+        return length;
+    }
 }