Author: veithen
Date: Mon Aug 22 18:44:37 2011
New Revision: 1160369
URL: http://svn.apache.org/viewvc?rev=1160369&view=rev
Log:
AXIOM-377 (step 4): Implemented DataHandlerExt#readOnce() to enable streaming
of the MIME part content.
Modified:
webservices/commons/trunk/modules/axiom/modules/axiom-api/src/main/java/org/apache/axiom/attachments/PartDataHandler.java
webservices/commons/trunk/modules/axiom/modules/axiom-api/src/main/java/org/apache/axiom/attachments/PartDataSource.java
webservices/commons/trunk/modules/axiom/modules/axiom-api/src/main/java/org/apache/axiom/attachments/PartImpl.java
webservices/commons/trunk/modules/axiom/modules/axiom-api/src/main/java/org/apache/axiom/attachments/lifecycle/DataHandlerExt.java
webservices/commons/trunk/modules/axiom/modules/axiom-api/src/main/java/org/apache/axiom/attachments/lifecycle/impl/DataHandlerExtImpl.java
webservices/commons/trunk/modules/axiom/modules/axiom-api/src/test/java/org/apache/axiom/attachments/AttachmentsTest.java
webservices/commons/trunk/modules/axiom/modules/axiom-testutils/src/main/java/org/apache/axiom/testutils/activation/RandomDataSource.java
Modified:
webservices/commons/trunk/modules/axiom/modules/axiom-api/src/main/java/org/apache/axiom/attachments/PartDataHandler.java
URL:
http://svn.apache.org/viewvc/webservices/commons/trunk/modules/axiom/modules/axiom-api/src/main/java/org/apache/axiom/attachments/PartDataHandler.java?rev=1160369&r1=1160368&r2=1160369&view=diff
==============================================================================
---
webservices/commons/trunk/modules/axiom/modules/axiom-api/src/main/java/org/apache/axiom/attachments/PartDataHandler.java
(original)
+++
webservices/commons/trunk/modules/axiom/modules/axiom-api/src/main/java/org/apache/axiom/attachments/PartDataHandler.java
Mon Aug 22 18:44:37 2011
@@ -19,6 +19,7 @@
package org.apache.axiom.attachments;
import java.io.IOException;
+import java.io.InputStream;
import javax.activation.DataHandler;
import javax.activation.DataSource;
@@ -41,6 +42,10 @@ class PartDataHandler extends DataHandle
return dataSource == null ? super.getDataSource() : dataSource;
}
+ public InputStream readOnce() throws IOException {
+ return part.getInputStream(false);
+ }
+
public void purgeDataSource() throws IOException {
part.releaseContent();
}
Modified:
webservices/commons/trunk/modules/axiom/modules/axiom-api/src/main/java/org/apache/axiom/attachments/PartDataSource.java
URL:
http://svn.apache.org/viewvc/webservices/commons/trunk/modules/axiom/modules/axiom-api/src/main/java/org/apache/axiom/attachments/PartDataSource.java?rev=1160369&r1=1160368&r2=1160369&view=diff
==============================================================================
---
webservices/commons/trunk/modules/axiom/modules/axiom-api/src/main/java/org/apache/axiom/attachments/PartDataSource.java
(original)
+++
webservices/commons/trunk/modules/axiom/modules/axiom-api/src/main/java/org/apache/axiom/attachments/PartDataSource.java
Mon Aug 22 18:44:37 2011
@@ -37,7 +37,7 @@ class PartDataSource implements DataSour
}
public InputStream getInputStream() throws IOException {
- return part.getInputStream();
+ return part.getInputStream(true);
}
public String getName() {
Modified:
webservices/commons/trunk/modules/axiom/modules/axiom-api/src/main/java/org/apache/axiom/attachments/PartImpl.java
URL:
http://svn.apache.org/viewvc/webservices/commons/trunk/modules/axiom/modules/axiom-api/src/main/java/org/apache/axiom/attachments/PartImpl.java?rev=1160369&r1=1160368&r2=1160369&view=diff
==============================================================================
---
webservices/commons/trunk/modules/axiom/modules/axiom-api/src/main/java/org/apache/axiom/attachments/PartImpl.java
(original)
+++
webservices/commons/trunk/modules/axiom/modules/axiom-api/src/main/java/org/apache/axiom/attachments/PartImpl.java
Mon Aug 22 18:44:37 2011
@@ -42,6 +42,28 @@ import java.util.Hashtable;
* Actual implementation of the {@link Part} interface.
*/
final class PartImpl implements Part {
+ /**
+ * The part has not been read yet. In this case the parser is in state
+ * {@link EntityState#T_BODY}.
+ */
+ private static final int STATE_UNREAD = 0;
+
+ /**
+ * The part has been read into a memory or file based buffer.
+ */
+ private static final int STATE_BUFFERED = 1;
+
+ /**
+ * The part content is being streamed, i.a. the application code consumes
the part content
+ * without buffering.
+ */
+ private static final int STATE_STREAMING = 2;
+
+ /**
+ * The part content has been discarded and can no longer be read. This
state is reached either
+ * when the content has been streamed or when it is discarded explicitly
after being buffered.
+ */
+ private static final int STATE_DISCARDED = 3;
private static Log log = LogFactory.getLog(PartImpl.class);
@@ -52,14 +74,16 @@ final class PartImpl implements Part {
// Value is a javax.mail.Header object
private Hashtable headers;
+ private int state = STATE_UNREAD;
+
/**
- * The MIME parser from which the content of this part is read. This is
only set if the content
- * has not been read yet. In this case the parser is in state {@link
EntityState#T_BODY}.
+ * The MIME parser from which the content of this part is read. This is
only set if the state is
+ * {@link #STATE_UNREAD} or {@link #STATE_STREAMING}.
*/
private MimeTokenStream parser;
/**
- * The content of this part. This is only set if the content of the part
is buffered.
+ * The content of this part. This is only set if the state is {@link
#STATE_BUFFERED}.
*/
private ContentStore content;
@@ -155,14 +179,15 @@ final class PartImpl implements Part {
}
private ContentStore getContent() {
- if (content == null) {
- if (parser == null) {
- throw new IllegalStateException("The content of the MIME part
has already been consumed");
- } else {
+ switch (state) {
+ case STATE_UNREAD:
fetch();
- }
+ // Fall through
+ case STATE_BUFFERED:
+ return content;
+ default:
+ throw new IllegalStateException("The content of the MIME part
has already been consumed");
}
- return content;
}
private static void checkParserState(EntityState state, EntityState
expected) throws IllegalStateException {
@@ -178,36 +203,53 @@ final class PartImpl implements Part {
* in the stream.
*/
void fetch() {
- if (content == null && parser != null) {
- checkParserState(parser.getState(), EntityState.T_BODY);
-
- // The PartFactory will determine which Part implementation is
most appropriate.
- content =
ContentStoreFactory.createContentStore(message.getLifecycleManager(),
- parser.getDecodedInputStream(),
- isSOAPPart,
- message.getThreshold(),
- message.getAttachmentRepoDir(),
- message.getContentLengthIfKnown());
// content-length for the whole message
- try {
- checkParserState(parser.next(), EntityState.T_END_BODYPART);
- EntityState state = parser.next();
- if (state == EntityState.T_EPILOGUE) {
- while (parser.next() != EntityState.T_END_MULTIPART) {
- // Just loop
- }
- } else if (state != EntityState.T_START_BODYPART && state !=
EntityState.T_END_MULTIPART) {
- throw new IllegalStateException("Internal error:
unexpected parser state " + state);
+ switch (state) {
+ case STATE_UNREAD:
+ checkParserState(parser.getState(), EntityState.T_BODY);
+
+ // The PartFactory will determine which Part implementation is
most appropriate.
+ content =
ContentStoreFactory.createContentStore(message.getLifecycleManager(),
+ parser.getDecodedInputStream(),
+ isSOAPPart,
+ message.getThreshold(),
+ message.getAttachmentRepoDir(),
+
message.getContentLengthIfKnown()); // content-length for the whole message
+ moveToNextPart();
+ state = STATE_BUFFERED;
+ break;
+ case STATE_STREAMING:
+ moveToNextPart();
+ state = STATE_DISCARDED;
+ }
+ }
+
+ private void moveToNextPart() {
+ try {
+ checkParserState(parser.next(), EntityState.T_END_BODYPART);
+ EntityState state = parser.next();
+ if (state == EntityState.T_EPILOGUE) {
+ while (parser.next() != EntityState.T_END_MULTIPART) {
+ // Just loop
}
- } catch (IOException ex) {
- throw new OMException(ex);
- } catch (MimeException ex) {
- throw new OMException(ex);
+ } else if (state != EntityState.T_START_BODYPART && state !=
EntityState.T_END_MULTIPART) {
+ throw new IllegalStateException("Internal error: unexpected
parser state " + state);
}
+ } catch (IOException ex) {
+ throw new OMException(ex);
+ } catch (MimeException ex) {
+ throw new OMException(ex);
}
+ parser = null;
}
- InputStream getInputStream() throws IOException {
- return getContent().getInputStream();
+ InputStream getInputStream(boolean preserve) throws IOException {
+ if (!preserve && state == STATE_UNREAD) {
+ checkParserState(parser.getState(), EntityState.T_BODY);
+ state = STATE_STREAMING;
+ return parser.getDecodedInputStream();
+ } else {
+ return getContent().getInputStream();
+ }
}
DataSource getDataSource() {
@@ -219,17 +261,20 @@ final class PartImpl implements Part {
}
void releaseContent() throws IOException {
- if (content != null) {
- content.destroy();
- } else if (parser != null) {
- try {
- EntityState state;
- do {
- state = parser.next();
- } while (state != EntityState.T_START_BODYPART && state !=
EntityState.T_END_MULTIPART);
- } catch (MimeException ex) {
- throw new OMException(ex);
- }
+ switch (state) {
+ case STATE_UNREAD:
+ try {
+ EntityState state;
+ do {
+ state = parser.next();
+ } while (state != EntityState.T_START_BODYPART && state !=
EntityState.T_END_MULTIPART);
+ } catch (MimeException ex) {
+ throw new OMException(ex);
+ }
+ state = STATE_DISCARDED;
+ break;
+ case STATE_BUFFERED:
+ content.destroy();
}
}
}
Modified:
webservices/commons/trunk/modules/axiom/modules/axiom-api/src/main/java/org/apache/axiom/attachments/lifecycle/DataHandlerExt.java
URL:
http://svn.apache.org/viewvc/webservices/commons/trunk/modules/axiom/modules/axiom-api/src/main/java/org/apache/axiom/attachments/lifecycle/DataHandlerExt.java?rev=1160369&r1=1160368&r2=1160369&view=diff
==============================================================================
---
webservices/commons/trunk/modules/axiom/modules/axiom-api/src/main/java/org/apache/axiom/attachments/lifecycle/DataHandlerExt.java
(original)
+++
webservices/commons/trunk/modules/axiom/modules/axiom-api/src/main/java/org/apache/axiom/attachments/lifecycle/DataHandlerExt.java
Mon Aug 22 18:44:37 2011
@@ -20,8 +20,22 @@
package org.apache.axiom.attachments.lifecycle;
import java.io.IOException;
+import java.io.InputStream;
+
+import javax.activation.DataHandler;
public interface DataHandlerExt {
+ /**
+ * Get an {@link InputStream} that consumes the content of this data
handler. This method is
+ * similar to {@link DataHandler#getInputStream()} except that it can be
invoked only once. If
+ * the content has not been buffered yet, then the implementation may
choose to enable streaming
+ * of the content.
+ *
+ * @return the stream representing the content; never <code>null</code>
+ * @throws IOException
+ * if an error occurs
+ */
+ InputStream readOnce() throws IOException;
/**
* This method will give users an option to trigger a purge
Modified:
webservices/commons/trunk/modules/axiom/modules/axiom-api/src/main/java/org/apache/axiom/attachments/lifecycle/impl/DataHandlerExtImpl.java
URL:
http://svn.apache.org/viewvc/webservices/commons/trunk/modules/axiom/modules/axiom-api/src/main/java/org/apache/axiom/attachments/lifecycle/impl/DataHandlerExtImpl.java?rev=1160369&r1=1160368&r2=1160369&view=diff
==============================================================================
---
webservices/commons/trunk/modules/axiom/modules/axiom-api/src/main/java/org/apache/axiom/attachments/lifecycle/impl/DataHandlerExtImpl.java
(original)
+++
webservices/commons/trunk/modules/axiom/modules/axiom-api/src/main/java/org/apache/axiom/attachments/lifecycle/impl/DataHandlerExtImpl.java
Mon Aug 22 18:44:37 2011
@@ -21,6 +21,7 @@ package org.apache.axiom.attachments.lif
import java.io.File;
import java.io.IOException;
+import java.io.InputStream;
import java.util.Observable;
import java.util.Observer;
@@ -47,7 +48,13 @@ public class DataHandlerExtImpl extends
this.manager = manager;
}
- public void deleteWhenReadOnce() throws IOException {
+ public InputStream readOnce() throws IOException {
+ // Instances of DataHandlerExtImpl are only created by legacy code,
so it's OK to
+ // throw an UnsupportedOperationException here.
+ throw new UnsupportedOperationException();
+ }
+
+ public void deleteWhenReadOnce() throws IOException {
deleteOnreadOnce = true;
FileAccessor fa =manager.getFileAccessor(getName());
if(fa==null){
Modified:
webservices/commons/trunk/modules/axiom/modules/axiom-api/src/test/java/org/apache/axiom/attachments/AttachmentsTest.java
URL:
http://svn.apache.org/viewvc/webservices/commons/trunk/modules/axiom/modules/axiom-api/src/test/java/org/apache/axiom/attachments/AttachmentsTest.java?rev=1160369&r1=1160368&r2=1160369&view=diff
==============================================================================
---
webservices/commons/trunk/modules/axiom/modules/axiom-api/src/test/java/org/apache/axiom/attachments/AttachmentsTest.java
(original)
+++
webservices/commons/trunk/modules/axiom/modules/axiom-api/src/test/java/org/apache/axiom/attachments/AttachmentsTest.java
Mon Aug 22 18:44:37 2011
@@ -25,6 +25,8 @@ import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
+import java.io.PipedInputStream;
+import java.io.PipedOutputStream;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;
@@ -32,6 +34,7 @@ import java.util.Random;
import java.util.Set;
import javax.activation.DataHandler;
+import javax.activation.DataSource;
import javax.mail.Session;
import javax.mail.internet.MimeBodyPart;
import javax.mail.internet.MimeMessage;
@@ -42,6 +45,7 @@ import org.apache.axiom.attachments.life
import org.apache.axiom.om.AbstractTestCase;
import org.apache.axiom.om.OMException;
import org.apache.axiom.om.TestConstants;
+import org.apache.axiom.testutils.activation.RandomDataSource;
import org.apache.axiom.testutils.io.IOTestUtils;
import org.apache.axiom.util.UIDGenerator;
import org.apache.commons.io.IOUtils;
@@ -516,4 +520,67 @@ public class AttachmentsTest extends Abs
in.close();
}
+
+ /**
+ * Tests that a call to {@link DataHandlerExt#readOnce()} on a {@link
DataHandler} returned by
+ * the {@link Attachments} object streams the content of the MIME part.
+ *
+ * @throws Exception
+ */
+ public void testDataHandlerStreaming() throws Exception {
+ // Note: We are only interested in the MimeMultipart, but we need to
create a
+ // MimeMessage to be able to calculate the correct content type
+ MimeMessage message = new MimeMessage((Session)null);
+ final MimeMultipart mp = new MimeMultipart("related");
+
+ // Prepare the "SOAP" part
+ MimeBodyPart bp1 = new MimeBodyPart();
+ // Obviously this is not SOAP, but this is irrelevant for this test
+ bp1.setText("<root/>", "utf-8", "xml");
+ bp1.addHeader("Content-Transfer-Encoding", "binary");
+ bp1.addHeader("Content-ID", "[email protected]");
+ mp.addBodyPart(bp1);
+
+ // Create an attachment that is larger than the maximum heap
+ DataSource dataSource = new
RandomDataSource((int)Math.min(Runtime.getRuntime().maxMemory(),
Integer.MAX_VALUE));
+ MimeBodyPart bp2 = new MimeBodyPart();
+ bp2.setDataHandler(new DataHandler(dataSource));
+ bp2.addHeader("Content-Transfer-Encoding", "binary");
+ bp2.addHeader("Content-ID", "[email protected]");
+ mp.addBodyPart(bp2);
+
+ message.setContent(mp);
+ // Compute the correct content type
+ message.saveChanges();
+
+ // We use a pipe (with a producer running in a separate thread)
because obviously we can't
+ // store the multipart in memory.
+ final PipedOutputStream pipeOut = new PipedOutputStream();
+ PipedInputStream pipeIn = new PipedInputStream(pipeOut);
+
+ Thread producerThread = new Thread(new Runnable() {
+ public void run() {
+ try {
+ try {
+ mp.writeTo(pipeOut);
+ } finally {
+ pipeOut.close();
+ }
+ } catch (Exception ex) {
+ ex.printStackTrace();
+ }
+ }
+ });
+ producerThread.start();
+
+ try {
+ // We configure Attachments to buffer MIME parts in memory. If the
part content is not
+ // streamed, then this will result in an OOM error.
+ Attachments attachments = new Attachments(pipeIn,
message.getContentType());
+ DataHandlerExt dh =
(DataHandlerExt)attachments.getDataHandler("[email protected]");
+ IOTestUtils.compareStreams(dataSource.getInputStream(),
dh.readOnce());
+ } finally {
+ pipeIn.close();
+ }
+ }
}
Modified:
webservices/commons/trunk/modules/axiom/modules/axiom-testutils/src/main/java/org/apache/axiom/testutils/activation/RandomDataSource.java
URL:
http://svn.apache.org/viewvc/webservices/commons/trunk/modules/axiom/modules/axiom-testutils/src/main/java/org/apache/axiom/testutils/activation/RandomDataSource.java?rev=1160369&r1=1160368&r2=1160369&view=diff
==============================================================================
---
webservices/commons/trunk/modules/axiom/modules/axiom-testutils/src/main/java/org/apache/axiom/testutils/activation/RandomDataSource.java
(original)
+++
webservices/commons/trunk/modules/axiom/modules/axiom-testutils/src/main/java/org/apache/axiom/testutils/activation/RandomDataSource.java
Mon Aug 22 18:44:37 2011
@@ -52,7 +52,7 @@ public class RandomDataSource implements
}
public String getContentType() {
- return null;
+ return "application/octet-stream";
}
public InputStream getInputStream() throws IOException {
