From 9f48391a2b723e6c9788dafba622c26eff748399 Mon Sep 17 00:00:00 2001 From: gail Date: Tue, 3 May 2011 14:42:08 -0700 Subject: [PATCH] Handle larger files in multi-part request. --- .../netty/NettyAsyncHttpProvider.java | 33 +++- .../ning/http/multipart/FilePartSource.java | 5 + .../ning/http/multipart/MultipartBody.java | 165 ++++++++++++++++++ .../java/com/ning/http/multipart/Part.java | 37 +++- 4 files changed, 234 insertions(+), 6 deletions(-) create mode 100644 src/main/java/com/ning/http/multipart/MultipartBody.java diff --git a/src/main/java/com/ning/http/client/providers/netty/NettyAsyncHttpProvider.java b/src/main/java/com/ning/http/client/providers/netty/NettyAsyncHttpProvider.java index cd079eec05..4ecaf0c76c 100644 --- a/src/main/java/com/ning/http/client/providers/netty/NettyAsyncHttpProvider.java +++ b/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 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); diff --git a/src/main/java/com/ning/http/multipart/FilePartSource.java b/src/main/java/com/ning/http/multipart/FilePartSource.java index 848d44a2f3..bc4c55fb4c 100644 --- a/src/main/java/com/ning/http/multipart/FilePartSource.java +++ b/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; + } + } diff --git a/src/main/java/com/ning/http/multipart/MultipartBody.java b/src/main/java/com/ning/http/multipart/MultipartBody.java new file mode 100644 index 0000000000..3e70e61243 --- /dev/null +++ b/src/main/java/com/ning/http/multipart/MultipartBody.java @@ -0,0 +1,165 @@ + +package com.ning.http.multipart; + +import com.ning.http.client.RandomAccessBody; + +import java.io.ByteArrayOutputStream; +import java.io.File; +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 parts, String boundary, String contentLength) { + _boundary = MultipartEncodingUtil.getAsciiBytes(boundary.substring("multipart/form-data; boundary=".length())); + _contentLength = Long.parseLong(contentLength); + _parts = parts; + + _files = new ArrayList(); + + _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; + } + + long availableLength = count; + + int tempPart = _startPart; + long totalLength = 0; + boolean full = false; + + while(!full && tempPart < _parts.size()) { + Part currentPart = (Part) _parts.get(tempPart); + + currentPart.setPartBoundary(_boundary); + + long length = currentPart.length(); + + if((length + totalLength) < availableLength ) { + totalLength += length; + tempPart++; + + if(currentPart.getClass().equals(StringPart.class)) { + + ByteArrayOutputStream outputStream = + new ByteArrayOutputStream(); + + Part.sendPart(outputStream, currentPart, _boundary); + + overallLength += writeToTarget(target, outputStream); + } + else if(currentPart.getClass().equals(FilePart.class)) { + + FilePart filePart = (FilePart)currentPart; + + ByteArrayOutputStream overhead = + new ByteArrayOutputStream(); + + filePart.setPartBoundary(_boundary); + + filePart.sendStart(overhead); + filePart.sendDispositionHeader(overhead); + filePart.sendContentTypeHeader(overhead); + filePart.sendTransferEncodingHeader(overhead); + filePart.sendEndOfHeader(overhead); + + overallLength += writeToTarget(target, overhead); + + 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."); + } + + ByteArrayOutputStream endOverhead = + new ByteArrayOutputStream(); + + filePart.sendEnd(endOverhead); + + overallLength += this.writeToTarget(target, endOverhead); + } + } + else { + full = true; + } + } + + ByteArrayOutputStream endWriter = + new ByteArrayOutputStream(); + + Part.sendMessageEnd(endWriter, _boundary); + + overallLength += writeToTarget(target, endWriter); + + _startPart = tempPart; + + return overallLength; + } + + 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 _parts; + private List _files; + private int _startPart; + +} diff --git a/src/main/java/com/ning/http/multipart/Part.java b/src/main/java/com/ning/http/multipart/Part.java index bb6aa65155..d817e5dc38 100644 --- a/src/main/java/com/ning/http/multipart/Part.java +++ b/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.