You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@synapse.apache.org by ve...@apache.org on 2008/03/20 00:16:02 UTC
svn commit: r639070 - in /synapse/trunk/java/modules/core/src:
main/java/org/apache/synapse/mediators/transform/
main/java/org/apache/synapse/util/ test/java/org/apache/synapse/util/
Author: veithen
Date: Wed Mar 19 16:15:53 2008
New Revision: 639070
URL: http://svn.apache.org/viewvc?rev=639070&view=rev
Log:
SYNAPSE-211: Modified XSLTMediator to switch from in-memory buffer to temporary file on the fly without restarting the transformation.
Modified:
synapse/trunk/java/modules/core/src/main/java/org/apache/synapse/mediators/transform/XSLTMediator.java
synapse/trunk/java/modules/core/src/main/java/org/apache/synapse/util/TemporaryData.java
synapse/trunk/java/modules/core/src/main/java/org/apache/synapse/util/TextFileDataSource.java
synapse/trunk/java/modules/core/src/test/java/org/apache/synapse/util/TemporaryDataTest.java
Modified: synapse/trunk/java/modules/core/src/main/java/org/apache/synapse/mediators/transform/XSLTMediator.java
URL: http://svn.apache.org/viewvc/synapse/trunk/java/modules/core/src/main/java/org/apache/synapse/mediators/transform/XSLTMediator.java?rev=639070&r1=639069&r2=639070&view=diff
==============================================================================
--- synapse/trunk/java/modules/core/src/main/java/org/apache/synapse/mediators/transform/XSLTMediator.java (original)
+++ synapse/trunk/java/modules/core/src/main/java/org/apache/synapse/mediators/transform/XSLTMediator.java Wed Mar 19 16:15:53 2008
@@ -27,7 +27,6 @@
import org.apache.axiom.om.impl.dom.DOOMAbstractFactory;
import org.apache.axiom.om.impl.dom.jaxp.DocumentBuilderFactoryImpl;
import org.apache.axiom.om.impl.llom.OMSourcedElementImpl;
-import org.apache.axiom.om.impl.llom.OMTextImpl;
import org.apache.axiom.om.util.ElementHelper;
import org.apache.axiom.om.util.StAXUtils;
import org.apache.axiom.soap.SOAP11Constants;
@@ -35,6 +34,7 @@
import org.apache.axiom.soap.SOAPEnvelope;
import org.apache.axiom.soap.impl.builder.StAXSOAPModelBuilder;
import org.apache.axis2.AxisFault;
+import org.apache.commons.io.IOUtils;
import org.apache.synapse.MessageContext;
import org.apache.synapse.SynapseException;
import org.apache.synapse.config.Entry;
@@ -42,14 +42,13 @@
import org.apache.synapse.mediators.AbstractMediator;
import org.apache.synapse.mediators.MediatorProperty;
import org.apache.synapse.transport.base.BaseConstants;
-import org.apache.synapse.util.FixedByteArrayOutputStream;
import org.apache.synapse.util.xpath.SynapseXPath;
+import org.apache.synapse.util.TemporaryData;
import org.apache.synapse.util.TextFileDataSource;
import org.jaxen.JaxenException;
import org.w3c.dom.Element;
import org.w3c.dom.Node;
-import javax.activation.FileDataSource;
import javax.xml.parsers.ParserConfigurationException;
import javax.xml.stream.XMLOutputFactory;
import javax.xml.stream.XMLStreamException;
@@ -214,11 +213,10 @@
boolean reCreate = false;
OMNode sourceNode = getTransformSource(synCtx);
- OutputStream osForTarget = null;
+ TemporaryData tempSourceData = null;
InputStream isForSource = null;
- ByteArrayOutputStream baosForTarget = new FixedByteArrayOutputStream(BYTE_ARRAY_SIZE);
- File tempTargetFile = null;
- File tempSourceFile = null;
+ TemporaryData tempTargetData = null;
+ OutputStream osForTarget = null;
boolean isSoapEnvelope = (sourceNode == synCtx.getEnvelope());
boolean isSoapBody = (sourceNode == synCtx.getEnvelope().getBody());
@@ -255,53 +253,31 @@
}
try {
- // create a byte array output stream and serialize the source node into it
- ByteArrayOutputStream baosForSource = new FixedByteArrayOutputStream(BYTE_ARRAY_SIZE);
- XMLStreamWriter xsWriterForSource = XMLOutputFactory.newInstance().
- createXMLStreamWriter(baosForSource);
-
- sourceNode.serialize(xsWriterForSource);
- isForSource = new ByteArrayInputStream(baosForSource.toByteArray());
- transformSrc = new StreamSource(isForSource);
- transformTgt = new StreamResult(baosForTarget);
-
- } catch (XMLStreamException e) {
- handleException("Error creating a StreamResult for the transformation", e, synCtx);
-
- } catch (SynapseException x) {
-
- if (traceOrDebugOn) {
- traceOrDebug(traceOn, "Error creating a StreamResult using a byte array" +
- " - attempting using temporary files for serialization");
- }
-
- OutputStream osForSource = null;
-
+ // Serialize the source node into a buffer or temporary file
+ tempSourceData = new TemporaryData(1024, BYTE_ARRAY_SIZE/1024, "xs_", ".xml");
+ OutputStream osForSource = tempSourceData.getOutputStream();
try {
- // create a output stream and serialize the source node into it
- tempSourceFile = File.createTempFile("xs_", ".xml");
- tempTargetFile = File.createTempFile("xt_", ".xml");
-
- osForSource = new FileOutputStream(tempSourceFile);
- osForTarget = new FileOutputStream(tempTargetFile);
-
- XMLStreamWriter xsWriterForSource =
- XMLOutputFactory.newInstance().createXMLStreamWriter(osForSource);
-
+ XMLStreamWriter xsWriterForSource = XMLOutputFactory.newInstance().
+ createXMLStreamWriter(osForSource);
+
sourceNode.serialize(xsWriterForSource);
- transformSrc = new StreamSource(tempSourceFile);
- transformTgt = new StreamResult(osForTarget);
-
- } catch (XMLStreamException e) {
- handleException("Error creating a StreamResult for the transformation", e, synCtx);
- } catch (IOException e) {
- handleException("Error using a temporary file/s for the transformation", e, synCtx);
- } finally {
- try {
- osForSource.close();
- } catch (IOException ignore) {}
}
- }
+ finally {
+ osForSource.close();
+ }
+ isForSource = tempSourceData.getInputStream();
+ transformSrc = new StreamSource(isForSource);
+
+ } catch (XMLStreamException e) {
+ handleException("Error creating a StreamSource for the transformation", e, synCtx);
+
+ } catch (IOException e) {
+ handleException("I/O error while creating a StreamSource for the transformation", e, synCtx);
+ }
+
+ tempTargetData = new TemporaryData(1024, BYTE_ARRAY_SIZE/1024, "xt_", ".xml");
+ osForTarget = tempTargetData.getOutputStream();
+ transformTgt = new StreamResult(osForTarget);
}
if (transformTgt == null) {
@@ -354,53 +330,38 @@
}
}
- try {
- transformer.setErrorListener(new ErrorListener() {
- public void warning(TransformerException e) throws TransformerException {
- if (traceOrDebugOn) {
- traceOrDebug(traceOn, "Warning encountered during transformation : " + e.getMessage());
- }
- log.warn("Transformation warning encountered", e);
- }
- public void error(TransformerException e) throws TransformerException {
- setTransformerException(e);
- }
- public void fatalError(TransformerException e) throws TransformerException {
- setTransformerException(e);
+ transformer.setErrorListener(new ErrorListener() {
+ public void warning(TransformerException e) throws TransformerException {
+ if (traceOrDebugOn) {
+ traceOrDebug(traceOn, "Warning encountered during transformation : " + e.getMessage());
}
- });
- transformer.transform(transformSrc, transformTgt);
-
- if (transformerException != null) {
- throw transformerException;
+ log.warn("Transformation warning encountered", e);
}
-
- } catch (TransformerException x) {
- // did we exceed the in-memory BYTE_ARRAY_SIZE? if so, use a file for output
- try {
- tempTargetFile = File.createTempFile("xt_", ".xml");
- osForTarget = new FileOutputStream(tempTargetFile);
- transformTgt = new StreamResult(osForTarget);
-
- // retry transformation again
- isForSource.reset();
- transformer.reset();
- transformer.transform(transformSrc, transformTgt);
-
- } catch (IOException e) {
- handleException("Error using a temporary file/s for the transformation", e, synCtx);
+ public void error(TransformerException e) throws TransformerException {
+ setTransformerException(e);
+ }
+ public void fatalError(TransformerException e) throws TransformerException {
+ setTransformerException(e);
}
+ });
+ transformer.transform(transformSrc, transformTgt);
+
+ if (transformerException != null) {
+ throw transformerException;
}
if (traceOrDebugOn) {
traceOrDebug(traceOn, "Transformation completed - processing result");
}
- if (tempSourceFile != null) {
- boolean deleted = tempSourceFile.delete();
- if (!deleted) {
- tempSourceFile.deleteOnExit();
- }
+ if (isForSource != null) {
+ IOUtils.closeQuietly(isForSource);
+ isForSource = null;
+ }
+
+ if (tempSourceData != null) {
+ tempSourceData.release();
+ tempSourceData = null;
}
// get the result OMElement
@@ -426,50 +387,24 @@
} else {
- // if we used a temporary file for the output of the transformation, read from it
- if (tempTargetFile != null) {
- try {
- XMLStreamReader reader = StAXUtils.createXMLStreamReader(
- new FileInputStream(tempTargetFile));
- if (isSoapEnvelope) {
- result = new StAXSOAPModelBuilder(reader).getSOAPEnvelope();
- } else {
- result = new StAXOMBuilder(reader).getDocumentElement();
- }
-
- } catch (XMLStreamException e) {
- handleException(
- "Error building result element from XSLT transformation", e, synCtx);
-
- } catch (Exception e) {
- result = handleNonXMLResult(tempTargetFile, traceOrDebugOn, traceOn);
-
- } finally {
- boolean deleted = tempTargetFile.delete();
- if (!deleted) {
- tempTargetFile.deleteOnExit();
- }
- }
+ try {
+ XMLStreamReader reader = StAXUtils.createXMLStreamReader(
+ tempTargetData.getInputStream());
+ if (isSoapEnvelope) {
+ result = new StAXSOAPModelBuilder(reader).getSOAPEnvelope();
+ } else {
+ result = new StAXOMBuilder(reader).getDocumentElement();
+ }
- } else {
- // read the Fixed byte array stream
- try {
- XMLStreamReader reader = StAXUtils.createXMLStreamReader(
- new ByteArrayInputStream(baosForTarget.toByteArray()));
- if (isSoapEnvelope) {
- result = new StAXSOAPModelBuilder(reader).getSOAPEnvelope();
- } else {
- result = new StAXOMBuilder(reader).getDocumentElement();
- }
+ } catch (XMLStreamException e) {
+ handleException(
+ "Error building result element from XSLT transformation", e, synCtx);
- } catch (XMLStreamException e) {
- handleException(
- "Error building result element from XSLT transformation", e, synCtx);
+ } catch (Exception e) {
+ result = handleNonXMLResult(tempTargetData, traceOrDebugOn, traceOn);
- } catch (Exception e) {
- result = handleNonXMLResult(baosForTarget.toString(), traceOrDebugOn, traceOn);
- }
}
+
}
if (result == null) {
@@ -603,12 +538,12 @@
/**
* If the transformation results in a non-XML payload, use standard wrapper elements
* to wrap the text payload so that other mediators could still process the result
- * @param file the text payload file
+ * @param tempData the text payload
* @param traceOrDebugOn is tracing on debug logging on?
* @param traceOn is tracing on?
* @return an OMElement wrapping the text payload
*/
- private OMElement handleNonXMLResult(File file, boolean traceOrDebugOn, boolean traceOn) {
+ private OMElement handleNonXMLResult(TemporaryData tempData, boolean traceOrDebugOn, boolean traceOn) {
OMFactory fac = OMAbstractFactory.getOMFactory();
OMElement wrapper = null;
@@ -617,41 +552,12 @@
traceOrDebug(traceOn, "Processing non SOAP/XML (text) transformation result");
}
if (traceOn && trace.isTraceEnabled()) {
- trace.trace("Wrapping text transformation result from : " + file);
+ trace.trace("Wrapping text transformation result");
}
- if (file != null) {
- TextFileDataSource txtFileDS = new TextFileDataSource(new FileDataSource(file));
+ if (tempData != null) {
+ TextFileDataSource txtFileDS = new TextFileDataSource(tempData);
wrapper = new OMSourcedElementImpl(BaseConstants.DEFAULT_TEXT_WRAPPER, fac, txtFileDS);
- }
-
- return wrapper;
- }
-
- /**
- * If the transformation results in a non-XML payload, use standard wrapper elements
- * to wrap the text payload so that other mediators could still process the result
- * @param textPayload the text payload to wrap
- * @param traceOrDebugOn is tracing on debug logging on?
- * @param traceOn is tracing on?
- * @return an OMElement wrapping the text payload
- */
- private OMElement handleNonXMLResult(String textPayload, boolean traceOrDebugOn, boolean traceOn) {
-
- OMFactory fac = OMAbstractFactory.getOMFactory();
- OMElement wrapper = null;
-
- if (traceOrDebugOn) {
- traceOrDebug(traceOn, "Processing non SOAP/XML (text) transformation result");
- }
- if (traceOn && trace.isTraceEnabled()) {
- trace.trace("Wrapping text transformation result : " + textPayload);
- }
-
- if (textPayload != null) {
- OMTextImpl textData = (OMTextImpl) fac.createOMText(textPayload);
- wrapper = fac.createOMElement(BaseConstants.DEFAULT_TEXT_WRAPPER, null);
- wrapper.addChild(textData);
}
return wrapper;
Modified: synapse/trunk/java/modules/core/src/main/java/org/apache/synapse/util/TemporaryData.java
URL: http://svn.apache.org/viewvc/synapse/trunk/java/modules/core/src/main/java/org/apache/synapse/util/TemporaryData.java?rev=639070&r1=639069&r2=639070&view=diff
==============================================================================
--- synapse/trunk/java/modules/core/src/main/java/org/apache/synapse/util/TemporaryData.java (original)
+++ synapse/trunk/java/modules/core/src/main/java/org/apache/synapse/util/TemporaryData.java Wed Mar 19 16:15:53 2008
@@ -7,6 +7,9 @@
import java.io.InputStream;
import java.io.OutputStream;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
/**
* Class representing some temporary data in the form of a byte stream.
* <p>
@@ -20,6 +23,8 @@
* call {@link #release()} to discard the temporary data.
*/
public class TemporaryData {
+ private static final Log log = LogFactory.getLog(TemporaryData.class);
+
class OutputStreamImpl extends OutputStream {
private FileOutputStream fileOutputStream;
@@ -29,6 +34,10 @@
} else if (len > (chunks.length-chunkIndex)*chunkSize - chunkOffset) {
// The buffer will overflow. Switch to a temporary file.
temporaryFile = File.createTempFile(tempPrefix, tempSuffix);
+ if (log.isDebugEnabled()) {
+ log.debug("Using temporary file " + temporaryFile);
+ }
+ temporaryFile.deleteOnExit();
fileOutputStream = new FileOutputStream(temporaryFile);
// Write the buffer to the temporary file.
for (int i=0; i<chunkIndex; i++) {
@@ -93,6 +102,8 @@
class InputStreamImpl extends InputStream {
private int currentChunkIndex;
private int currentChunkOffset;
+ private int markChunkIndex;
+ private int markChunkOffset;
public int available() throws IOException {
return (chunkIndex-currentChunkIndex)*chunkSize + chunkOffset - currentChunkOffset;
@@ -140,6 +151,20 @@
return read(b) == -1 ? -1 : (int)b[0] & 0xFF;
}
+ public boolean markSupported() {
+ return true;
+ }
+
+ public void mark(int readlimit) {
+ markChunkIndex = currentChunkIndex;
+ markChunkOffset = currentChunkOffset;
+ }
+
+ public void reset() throws IOException {
+ currentChunkIndex = markChunkIndex;
+ currentChunkOffset = markChunkOffset;
+ }
+
public long skip(long n) throws IOException {
int available = available();
int c = n < available ? (int)n : available;
@@ -214,6 +239,13 @@
public void release() {
if (temporaryFile != null) {
+ temporaryFile.delete();
+ }
+ }
+
+ protected void finalize() throws Throwable {
+ if (temporaryFile != null) {
+ log.warn("Cleaning up unreleased temporary file " + temporaryFile);
temporaryFile.delete();
}
}
Modified: synapse/trunk/java/modules/core/src/main/java/org/apache/synapse/util/TextFileDataSource.java
URL: http://svn.apache.org/viewvc/synapse/trunk/java/modules/core/src/main/java/org/apache/synapse/util/TextFileDataSource.java?rev=639070&r1=639069&r2=639070&view=diff
==============================================================================
--- synapse/trunk/java/modules/core/src/main/java/org/apache/synapse/util/TextFileDataSource.java (original)
+++ synapse/trunk/java/modules/core/src/main/java/org/apache/synapse/util/TextFileDataSource.java Wed Mar 19 16:15:53 2008
@@ -49,6 +49,15 @@
"Unable to get an InputStream for DataSource : " + ds.getName(), e);
}
}
+
+ public TextFileDataSource(TemporaryData temporaryData) {
+ try {
+ this.is = temporaryData.getInputStream();
+ } catch (IOException e) {
+ throw new SynapseException(
+ "Unable to get an InputStream for temporary data", e);
+ }
+ }
public void serialize(OutputStream out, OMOutputFormat format) throws XMLStreamException {
try {
Modified: synapse/trunk/java/modules/core/src/test/java/org/apache/synapse/util/TemporaryDataTest.java
URL: http://svn.apache.org/viewvc/synapse/trunk/java/modules/core/src/test/java/org/apache/synapse/util/TemporaryDataTest.java?rev=639070&r1=639069&r2=639070&view=diff
==============================================================================
--- synapse/trunk/java/modules/core/src/test/java/org/apache/synapse/util/TemporaryDataTest.java (original)
+++ synapse/trunk/java/modules/core/src/test/java/org/apache/synapse/util/TemporaryDataTest.java Wed Mar 19 16:15:53 2008
@@ -19,12 +19,15 @@
package org.apache.synapse.util;
+import java.io.DataInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.Arrays;
import java.util.Random;
+import org.apache.commons.io.IOUtils;
+
import junit.framework.TestCase;
public class TemporaryDataTest extends TestCase {
@@ -76,5 +79,27 @@
public void testRandomReadWriteWithTemporaryFile() throws IOException {
doTestRandomReadWrite(100000);
+ }
+
+ public void testMarkReset() throws IOException {
+ byte[] sourceData1 = new byte[2000];
+ byte[] sourceData2 = new byte[2000];
+ random.nextBytes(sourceData1);
+ random.nextBytes(sourceData2);
+ TemporaryData tmp = new TemporaryData(16, 512, "test", ".dat");
+ OutputStream out = tmp.getOutputStream();
+ out.write(sourceData1);
+ out.write(sourceData2);
+ out.close();
+ DataInputStream in = new DataInputStream(tmp.getInputStream());
+ byte[] data1 = new byte[sourceData1.length];
+ byte[] data2 = new byte[sourceData2.length];
+ in.readFully(data1);
+ in.mark(sourceData2.length);
+ in.readFully(data2);
+ in.reset();
+ in.readFully(data2);
+ assertTrue(Arrays.equals(sourceData1, data1));
+ assertTrue(Arrays.equals(sourceData2, data2));
}
}