Ahc 78

src/main/java/com/ning/http/client/providers/netty/NettyAsyncHttpProvider.java

@@ -44,6 +44,7 @@
 import com.ning.http.client.ntlm.NTLMEngine;
 import com.ning.http.client.ntlm.NTLMEngineException;
 import com.ning.http.client.providers.netty.spnego.SpnegoEngine;
+import com.ning.http.multipart.MultipartBody;
 import com.ning.http.multipart.MultipartRequestEntity;
 import com.ning.http.util.AsyncHttpProviderUtils;
 import com.ning.http.util.AuthenticatorUtils;
@@ -399,7 +400,34 @@ protected final <T> void writeRequest(final Channel channel,
 
             if (future.getAndSetWriteBody(true)) {
                 if (!future.getNettyRequest().getMethod().equals(HttpMethod.CONNECT)) {
-                    if (future.getRequest().getFile() != null) {
+                	if(future.getRequest().getParts() != null) {
+               		 String boundary = future.getNettyRequest().getHeader(
+               			"Content-Type");
+
+               		 String length = future.getNettyRequest().getHeader(
+               			"Content-Length");
+
+               		 final MultipartBody multipartBody = new MultipartBody(
+               			future.getRequest().getParts(), boundary, length);
+
+               		 ChannelFuture writeFuture = channel.write(
+               			new BodyFileRegion(multipartBody));
+
+                        final Body b = multipartBody;
+
+                        writeFuture.addListener(new ProgressListener(
+                       		false, future.getAsyncHandler(), future) {
+                            public void operationComplete(ChannelFuture cf) {
+                                try {
+                                    b.close();
+                                } catch (IOException e) {
+                                    log.warn("Failed to close request body: {}", e.getMessage(), e);
+                                }
+                                super.operationComplete(cf);
+                            }
+                        });
+               	}
+               	else if (future.getRequest().getFile() != null) {
                         final File file = future.getRequest().getFile();
                         long fileLength = 0;
                         final RandomAccessFile raf = new RandomAccessFile(file, "r");
@@ -682,9 +710,6 @@ private static HttpRequest construct(AsyncHttpClientConfig config,
                     nettyRequest.setHeader(HttpHeaders.Names.CONTENT_TYPE, mre.getContentType());
                     nettyRequest.setHeader(HttpHeaders.Names.CONTENT_LENGTH, String.valueOf(mre.getContentLength()));
 
-                    ChannelBuffer b = ChannelBuffers.dynamicBuffer(lenght);
-                    mre.writeRequest(new ChannelBufferOutputStream(b));
-                    nettyRequest.setContent(b);
                 } else if (request.getEntityWriter() != null) {
                     int lenght = computeAndSetContentLength(request, nettyRequest);
 

src/main/java/com/ning/http/multipart/FilePartSource.java

@@ -107,5 +107,10 @@ public InputStream createInputStream() throws IOException {
             return new ByteArrayInputStream(new byte[] {});
         }
     }
+
+    public File getFile() {
+    	return file;
+    }
+
 
 }

src/main/java/com/ning/http/multipart/MultipartBody.java

@@ -0,0 +1,240 @@
+
+package com.ning.http.multipart;
+
+import com.ning.http.client.RandomAccessBody;
+
+import java.io.ByteArrayOutputStream;
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.io.RandomAccessFile;
+import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
+import java.nio.channels.WritableByteChannel;
+import java.util.ArrayList;
+import java.util.List;
+
+public class MultipartBody implements RandomAccessBody {
+
+	public MultipartBody(List<com.ning.http.client.Part> parts, String boundary, String contentLength) {
+		_boundary = MultipartEncodingUtil.getAsciiBytes(boundary.substring("multipart/form-data; boundary=".length()));
+		_contentLength = Long.parseLong(contentLength);
+		_parts = parts;
+
+		_files = new ArrayList<RandomAccessFile>();
+
+		_startPart = 0;
+	}
+
+	public void close() throws IOException {
+		for(RandomAccessFile file : _files) {
+			file.close();
+		}
+	}
+
+	public long getContentLength() {
+		return _contentLength;
+	}
+
+	public long read(ByteBuffer buffer) throws IOException {
+		// TODO Not implemented
+		return 0;
+	}
+
+	public long transferTo(long position, long count, WritableByteChannel target)
+			throws IOException {
+
+		long overallLength = 0;
+
+		if(_startPart == _parts.size()) {
+			return overallLength;
+		}
+
+		int tempPart = _startPart;
+
+		for(com.ning.http.client.Part part : _parts) {
+			if(part instanceof Part) {
+				overallLength += handleMultiPart(target, (Part)part);
+			}
+			else {
+				overallLength += handleClientPart(target, part);
+			}
+
+			tempPart++;
+		}
+		ByteArrayOutputStream endWriter = 
+			new ByteArrayOutputStream();
+
+		Part.sendMessageEnd(endWriter, _boundary);
+
+		overallLength += writeToTarget(target, endWriter);
+
+		_startPart = tempPart;
+
+		return overallLength;
+	}
+
+	private long handleClientPart(
+		WritableByteChannel target,	com.ning.http.client.Part part) throws IOException {
+
+		if(part.getClass().equals(com.ning.http.client.StringPart.class)) {
+			com.ning.http.client.StringPart stringPart = (com.ning.http.client.StringPart)part;
+
+			StringPart currentPart = new StringPart(stringPart.getName(), stringPart.getValue());
+
+			return handleStringPart(target,currentPart);
+		}
+		else if(part.getClass().equals(com.ning.http.client.FilePart.class)) {
+			com.ning.http.client.FilePart currentPart = (com.ning.http.client.FilePart)part;
+
+			FilePart filePart =  new FilePart(currentPart.getName(), currentPart.getFile());
+
+			return handleFilePart(target, filePart);
+		}
+		else if(part.getClass().equals(com.ning.http.client.ByteArrayPart.class)) {
+			com.ning.http.client.ByteArrayPart bytePart = (com.ning.http.client.ByteArrayPart)part;
+
+		    ByteArrayPartSource source = new ByteArrayPartSource(bytePart.getFileName(), bytePart.getData());
+
+			FilePart filePart = new FilePart(bytePart.getName(), source, bytePart.getMimeType(), bytePart.getCharSet());
+
+			return handleByteArrayPart(target, filePart, bytePart.getData());
+		}
+
+		return 0;
+	}
+
+	private long handleByteArrayPart(WritableByteChannel target,
+			FilePart filePart, byte[] data) throws IOException {
+
+		int length = 0;
+
+		//length += handleFileHeaders(target, filePart);
+
+		ByteArrayOutputStream output = new ByteArrayOutputStream();
+
+		Part.sendPart(output, filePart, _boundary);
+
+		length += writeToTarget(target, output);
+
+		//length += handleFileEnd(target, filePart);
+
+		return length;
+
+	}
+
+	private long handleFileEnd(WritableByteChannel target, FilePart filePart) 
+		throws IOException {
+
+		ByteArrayOutputStream endOverhead =
+			new ByteArrayOutputStream();
+
+		filePart.sendEnd(endOverhead);
+
+		return this.writeToTarget(target, endOverhead);
+	}
+
+	private long handleFileHeaders(WritableByteChannel target,
+			FilePart filePart) throws IOException {
+		filePart.setPartBoundary(_boundary);
+
+		ByteArrayOutputStream overhead = new ByteArrayOutputStream();
+
+		filePart.setPartBoundary(_boundary);
+
+		filePart.sendStart(overhead);
+		filePart.sendDispositionHeader(overhead);
+		filePart.sendContentTypeHeader(overhead);
+		filePart.sendTransferEncodingHeader(overhead);
+		filePart.sendEndOfHeader(overhead);
+
+		return writeToTarget(target, overhead);
+	}
+
+	private long handleFilePart(WritableByteChannel target, FilePart filePart) 
+		throws IOException, FileNotFoundException {
+
+		int length = 0;
+
+		length += handleFileHeaders(target, filePart);
+
+		FilePartSource source = (FilePartSource)filePart.getSource();
+
+		File file = source.getFile();
+
+		RandomAccessFile raf = new RandomAccessFile(file, "r");
+			_files.add(raf);
+
+		FileChannel fc = raf.getChannel();
+
+
+		long fileLength = fc.transferTo(0, file.length(), target);
+
+		if(fileLength != file.length()) {
+			System.out.println("Did not complete file.");
+		}
+
+		length += handleFileEnd(target, filePart);
+
+		return length;
+	}
+
+	private long handleStringPart(WritableByteChannel target, StringPart currentPart) 
+		throws IOException {
+
+		currentPart.setPartBoundary(_boundary);
+
+		ByteArrayOutputStream outputStream = 
+			new ByteArrayOutputStream();
+
+		Part.sendPart(outputStream, currentPart, _boundary);
+
+		return writeToTarget(target, outputStream);
+	}
+
+	private long handleMultiPart(WritableByteChannel target, Part currentPart) 
+		throws IOException, FileNotFoundException {
+
+		currentPart.setPartBoundary(_boundary);
+
+		if(currentPart.getClass().equals(StringPart.class)) {
+			return handleStringPart(target, (StringPart)currentPart);
+		}
+		else if(currentPart.getClass().equals(FilePart.class)) {
+			FilePart filePart = (FilePart)currentPart;
+
+			return handleFilePart(target, filePart);
+		}
+		return 0;
+	}
+
+	private long writeToTarget(
+		WritableByteChannel target,	ByteArrayOutputStream byteWriter)
+			throws IOException {
+
+		int written = 0;
+		synchronized(byteWriter) {
+			while((target.isOpen()) && (written < byteWriter.size())) {
+				ByteBuffer message = ByteBuffer.wrap(byteWriter.toByteArray());
+				written = target.write(message);
+				if(written != byteWriter.size()) {
+					System.out.println("Waiting...");
+					try {
+						byteWriter.wait(1000);
+					} catch (InterruptedException e) {
+						// TODO Auto-generated catch block
+						e.printStackTrace();
+					}
+				}
+			}
+		}
+		return written;
+	}
+
+	private byte[] _boundary;
+	private long _contentLength;
+	private List<com.ning.http.client.Part> _parts;
+	private List<RandomAccessFile> _files;
+	private int _startPart;
+
+}

src/main/java/com/ning/http/multipart/Part.java

@@ -378,8 +378,41 @@ public static void sendParts(OutputStream out, Part[] parts, byte[] partBoundary
         out.write(EXTRA_BYTES);
         out.write(CRLF_BYTES);
     }
-
-    /**
+
+    public static void sendMessageEnd(OutputStream out, byte[] partBoundary) 
+		throws IOException {
+
+		if (partBoundary == null || partBoundary.length == 0) {
+	        throw new IllegalArgumentException("partBoundary may not be empty");
+	    }
+
+		out.write(EXTRA_BYTES);
+	    out.write(partBoundary);
+	    out.write(EXTRA_BYTES);
+	    out.write(CRLF_BYTES);
+	}
+
+	/**
+	 * Write all parts and the last boundary to the specified output stream.
+	 *
+	 * @param out          The stream to write to.
+	 * @param part         The part to write.
+	 * @throws IOException If an I/O error occurs while writing the parts.
+	 * @since N/A
+	 */
+	public static void sendPart(OutputStream out, Part part, byte[] partBoundary)
+	        throws IOException {
+
+	    if (part == null) {
+	        throw new IllegalArgumentException("Parts may not be null");
+	    }
+
+	    part.setPartBoundary(partBoundary);
+	    part.send(out);
+	}
+
+
+	/**
      * Return the total sum of all parts and that of the last boundary
      *
      * @param parts The parts.

src/test/java/com/ning/http/client/async/AuthTimeoutTest.java

@@ -38,6 +38,7 @@
 import javax.servlet.http.HttpServletResponse;
 import java.io.IOException;
 import java.io.OutputStream;
+import java.util.Arrays;
 import java.util.HashSet;
 import java.util.Set;
 import java.util.concurrent.Future;
@@ -88,6 +89,7 @@ public void setUpServer(String auth)
         knownRoles.add(admin);
 
         ConstraintSecurityHandler security = new ConstraintSecurityHandler();
+
         security.setConstraintMappings(new ConstraintMapping[]{mapping}, knownRoles);
         security.setAuthenticator(new BasicAuthenticator());
         security.setLoginService(loginService);

src/test/java/com/ning/http/client/async/BasicAuthTest.java

@@ -53,6 +53,7 @@
 import java.io.FileInputStream;
 import java.io.IOException;
 import java.net.URL;
+import java.util.Arrays;
 import java.util.HashSet;
 import java.util.Set;
 import java.util.concurrent.ExecutionException;

src/test/java/com/ning/http/client/async/DigestAuthTest.java

@@ -37,6 +37,7 @@
 import javax.servlet.http.HttpServletRequest;
 import javax.servlet.http.HttpServletResponse;
 import java.io.IOException;
+import java.util.Arrays;
 import java.util.HashSet;
 import java.util.Set;
 import java.util.concurrent.ExecutionException;