This is an automated email from the ASF dual-hosted git repository.
ffang pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/cxf.git
The following commit(s) were added to refs/heads/main by this push:
new 479c8098262 CXF-9258: Fix WS-RM retransmission of messages with
attachments (#3598)
479c8098262 is described below
commit 479c809826280804227edc744205fcdcfa9e74d2
Author: Freeman(Yue) Fang <[email protected]>
AuthorDate: Mon Oct 19 12:29:27 2026 -0400
CXF-9258: Fix WS-RM retransmission of messages with attachments (#3598)
* CXF-9258: Fix WS-RM retransmission of messages with attachments
AttachmentOutInterceptor replaces the SOAP Content-Type with the multipart
wrapper and marks the message as already processed, so a retransmitted
message was sent without its attachments, or with a nested multipart, both
when it was resent from memory and when it was recovered from the RMStore.
In-memory path:
- Save the bare SOAP Content-Type (SAVED_CONTENT_TYPE) before
AttachmentOutInterceptor replaces it
- On resend, clear ATTACHMENT_OUT_CHECKED and the AttachmentSerializer
content
and restore the saved Content-Type, so the multipart is rebuilt correctly
Persisted path (PersistenceUtils):
- encodeRMContent writes the root part with the SOAP Content-Type instead of
the multipart one, keeps the root part encoding, and treats an empty
attachment collection as no attachments
- decodeRMContent restores the root Content-Type from the stored message
- For inbound (destination) messages the root type is taken from
ATTACHMENT_PART_HEADERS, with the charset stripped
- The root type is never a multipart type: if it cannot be determined, fall
back to text/xml with a warning so a multipart is never nested
- The root part of a captured MTOM message is stored as an XOP part, and
decodeRMContent unwraps the SOAP type from an XOP root part and enables
MTOM, so a recovered MTOM message is retransmitted as an XOP package
Tests:
- PersistenceUtilsTest: round trips, SOAP 1.2 action and charset handling,
inbound path, missing root type, encoding, MTOM/XOP round trips
- RetransmissionAttachmentTest: SwA and MTOM retransmission, and
retransmission
of SwA and MTOM messages recovered from the client store after a restart
- ServerAttachmentRecoveryTest: an inbound message with an attachment that
could not be delivered is persisted by the destination and redelivered
with
its attachment after a server restart
- The recovery tests use their own in-memory database per test, as an
in-memory HSQLDB database lives as long as the JVM
* CXF-9258: Address review comments
- Parse the Content-Type parameters with jakarta.activation.MimeType
instead of regular expressions, so a charset in a quoted type value
is no longer removed with the closing quote, and only the first
Content-Type of the root part is used
- Take the root type of a multipart type, like one stored as the type
of the root part, from its start-info and type parameters, so a
SOAP 1.2 message keeps its type and action and an XOP package is
restored with MTOM, instead of falling back to text/xml
- Store a message as an XOP package if it is outbound with MTOM, not if
it has a saved Content-Type
- Encode the inbound messages without attachments with encodeRMContent
too, so the destination stores the normalised root type
- Remove a redundant removal of the AttachmentSerializer content on resend
---
.../org/apache/cxf/ws/rm/DestinationSequence.java | 10 +-
.../apache/cxf/ws/rm/RMCaptureOutInterceptor.java | 3 +
.../org/apache/cxf/ws/rm/RMMessageConstants.java | 3 +
.../cxf/ws/rm/persistence/PersistenceUtils.java | 162 +++++++++-
.../cxf/ws/rm/soap/RetransmissionQueueImpl.java | 8 +-
.../ws/rm/persistence/PersistenceUtilsTest.java | 339 +++++++++++++++++++++
.../ws/rm/RetransmissionAttachmentTest.java | 299 ++++++++++++++++++
.../ws/rm/ServerAttachmentRecoveryTest.java | 219 +++++++++++++
8 files changed, 1029 insertions(+), 14 deletions(-)
diff --git
a/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/DestinationSequence.java
b/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/DestinationSequence.java
index 983d3089293..946b2a237af 100644
--- a/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/DestinationSequence.java
+++ b/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/DestinationSequence.java
@@ -20,7 +20,6 @@
package org.apache.cxf.ws.rm;
import java.io.IOException;
-import java.io.InputStream;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
@@ -191,14 +190,7 @@ public class DestinationSequence extends AbstractSequence {
.get(RMMessageConstants.SAVED_CONTENT);
msg.setMessageNumber(st.getMessageNumber());
msg.setCreatedTime(rmps.getCreatedTime());
- // in case no attachments are available, cos can be saved
directly
- if (message.getAttachments() == null) {
- msg.setContent(cos);
-
msg.setContentType((String)message.get(Message.CONTENT_TYPE));
- } else {
- InputStream is = cos.getInputStream();
- PersistenceUtils.encodeRMContent(msg, message, is);
- }
+ PersistenceUtils.encodeRMContent(msg, message, cos);
store.persistIncoming(this, msg);
} catch (IOException e) {
throw new Fault(e);
diff --git
a/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMCaptureOutInterceptor.java
b/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMCaptureOutInterceptor.java
index b70a0d4d18e..62a40a0bc62 100644
--- a/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMCaptureOutInterceptor.java
+++ b/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMCaptureOutInterceptor.java
@@ -222,6 +222,9 @@ public class RMCaptureOutInterceptor extends
AbstractRMInterceptor<Message> {
}
private void captureMessage(Message message) {
message.put(RMMessageConstants.MESSAGE_CAPTURE, Boolean.TRUE);
+ if (message.get(Message.CONTENT_TYPE) != null) {
+ message.put(RMMessageConstants.SAVED_CONTENT_TYPE,
message.get(Message.CONTENT_TYPE));
+ }
message.getInterceptorChain().add(new CaptureStart());
message.getInterceptorChain().add(new CaptureEnd());
diff --git
a/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMMessageConstants.java
b/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMMessageConstants.java
index c3aa61284e1..b06f5d69a91 100644
--- a/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMMessageConstants.java
+++ b/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMMessageConstants.java
@@ -39,6 +39,9 @@ public final class RMMessageConstants {
/** Message content must be an instance of {@link CachedOutputStream}. */
public static final String SAVED_CONTENT = "org.apache.cxf.ws.rm.content";
+ /** Saved Content-Type before AttachmentOutInterceptor replaces it with
multipart. */
+ public static final String SAVED_CONTENT_TYPE =
"org.apache.cxf.ws.rm.content.type";
+
/** Variable holds reference to source streams of the attachments.
* It must be an instance of {@link Closeable}. */
public static final String ATTACHMENTS_CLOSEABLE =
"org.apache.cxf.ws.rm.attachment.closeable";
diff --git
a/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/persistence/PersistenceUtils.java
b/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/persistence/PersistenceUtils.java
index 3aef23dda4e..3661ba6ad77 100644
---
a/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/persistence/PersistenceUtils.java
+++
b/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/persistence/PersistenceUtils.java
@@ -23,20 +23,30 @@ import java.io.Closeable;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
+import java.util.List;
+import java.util.Map;
+import java.util.logging.Level;
+import java.util.logging.Logger;
import javax.xml.stream.XMLStreamReader;
+import jakarta.activation.MimeType;
+import jakarta.activation.MimeTypeParseException;
import jakarta.xml.bind.JAXBContext;
import jakarta.xml.bind.JAXBElement;
import jakarta.xml.bind.JAXBException;
import org.apache.cxf.attachment.AttachmentDeserializer;
import org.apache.cxf.attachment.AttachmentSerializer;
+import org.apache.cxf.attachment.AttachmentUtil;
+import org.apache.cxf.common.logging.LogUtils;
import org.apache.cxf.common.util.PackageUtils;
+import org.apache.cxf.helpers.CastUtils;
import org.apache.cxf.helpers.IOUtils;
import org.apache.cxf.helpers.LoadingByteArrayOutputStream;
import org.apache.cxf.io.CachedOutputStream;
import org.apache.cxf.message.Message;
import org.apache.cxf.message.MessageImpl;
+import org.apache.cxf.message.MessageUtils;
import org.apache.cxf.staxutils.StaxUtils;
import org.apache.cxf.ws.rm.RMMessageConstants;
import org.apache.cxf.ws.rm.v200702.SequenceAcknowledgement;
@@ -46,6 +56,12 @@ import org.apache.cxf.ws.rm.v200702.SequenceAcknowledgement;
*/
public final class PersistenceUtils {
+ private static final Logger LOG =
LogUtils.getLogger(PersistenceUtils.class);
+
+ private static final String DEFAULT_ROOT_CONTENT_TYPE = "text/xml";
+
+ private static final String XOP_CONTENT_TYPE = "application/xop+xml";
+
private static PersistenceUtils instance;
private JAXBContext context;
@@ -103,11 +119,125 @@ public final class PersistenceUtils {
return context;
}
+ /**
+ * Content type of the SOAP root message (without the multipart wrapper),
or null if unknown.
+ * Captured outbound messages provide it via SAVED_CONTENT_TYPE; for
inbound multipart
+ * messages it is taken from the headers of the root part. The result is
never a multipart type,
+ * as the multipart wrapper is recreated when the message is written, so a
multipart type would
+ * end up nested into another multipart message.
+ */
+ private static String getRootContentType(Message msg) {
+ String ct = (String)msg.get(RMMessageConstants.SAVED_CONTENT_TYPE);
+ if (ct == null) {
+ Map<String, List<String>> ih =
+ CastUtils.cast((Map<?,
?>)msg.get(AttachmentDeserializer.ATTACHMENT_PART_HEADERS));
+ List<String> types = ih != null ? ih.get("Content-Type") : null;
+ if (types != null && !types.isEmpty()) {
+ // the charset is added again by the AttachmentSerializer when
the message is written
+ ct = removeCharset(types.get(0));
+ }
+ }
+ if (ct == null) {
+ ct = (String) msg.get(Message.CONTENT_TYPE);
+ }
+ if (isMultipart(ct)) {
+ return getMultipartRootType(ct);
+ }
+ return ct;
+ }
+
+ /**
+ * Content type of the root part of a multipart message of the given type,
used when the multipart
+ * type is the only known type, like when it was stored as the type of the
root part. The start-info
+ * parameter written by the AttachmentSerializer holds the SOAP content
type with its parameters,
+ * like the SOAP 1.2 action, and the type parameter is the one of the root
part, application/xop+xml
+ * for an XOP package.
+ */
+ private static String getMultipartRootType(String multipartType) {
+ String rootType = null;
+ try {
+ MimeType mt = new MimeType(multipartType);
+ String type = mt.getParameter("type");
+ String startInfo = mt.getParameter("start-info");
+ if (XOP_CONTENT_TYPE.equalsIgnoreCase(type)) {
+ MimeType xop = new MimeType(XOP_CONTENT_TYPE);
+ if (startInfo != null) {
+ xop.setParameter("type", removeCharset(startInfo));
+ }
+ rootType = xop.toString();
+ } else {
+ String soapType = startInfo != null ? startInfo : type;
+ if (soapType != null && !isMultipart(soapType)) {
+ rootType = removeCharset(soapType);
+ }
+ }
+ } catch (MimeTypeParseException e) {
+ LOG.log(Level.FINE, "Invalid multipart Content-Type " +
multipartType, e);
+ }
+ if (rootType == null) {
+ LOG.log(Level.WARNING, "The Content-Type of the SOAP root message
is unknown, using "
+ + DEFAULT_ROOT_CONTENT_TYPE);
+ return DEFAULT_ROOT_CONTENT_TYPE;
+ }
+ return rootType;
+ }
+
+ /**
+ * Type of the SOAP message wrapped into an XOP root part of the given
content type, or null if
+ * the content type is not the one of an XOP root part.
+ */
+ private static String getXopWrappedType(String rootType) {
+ if (rootType == null) {
+ return null;
+ }
+ MimeType mt;
+ try {
+ mt = new MimeType(rootType);
+ } catch (MimeTypeParseException e) {
+ return null;
+ }
+ if (!XOP_CONTENT_TYPE.equalsIgnoreCase(mt.getBaseType())) {
+ return null;
+ }
+ String type = mt.getParameter("type");
+ return type != null ? removeCharset(type) : DEFAULT_ROOT_CONTENT_TYPE;
+ }
+
+ private static boolean isMultipart(String ct) {
+ return ct != null && ct.regionMatches(true, 0, "multipart/", 0,
"multipart/".length());
+ }
+
+ /** Removes the charset parameter, but not the one in the value of another
parameter */
+ private static String removeCharset(String ct) {
+ try {
+ MimeType mt = new MimeType(ct);
+ mt.removeParameter("charset");
+ return mt.toString();
+ } catch (MimeTypeParseException e) {
+ // only keep the type/subtype of an invalid content type
+ int index = ct.indexOf(';');
+ return (index == -1 ? ct : ct.substring(0, index)).trim();
+ }
+ }
+
+ /**
+ * Encodes the content of a message, which is stored as it is if the
message has no attachments.
+ */
+ public static void encodeRMContent(RMMessage rmmsg, Message msg,
CachedOutputStream msgContent)
+ throws IOException {
+ if (msg.getAttachments() == null || msg.getAttachments().isEmpty()) {
+ rmmsg.setContentType(getRootContentType(msg));
+ rmmsg.setContent(msgContent);
+ } else {
+ encodeRMContent(rmmsg, msg, msgContent.getInputStream());
+ }
+ }
+
public static void encodeRMContent(RMMessage rmmsg, Message msg,
InputStream msgContent)
throws IOException {
CachedOutputStream cos = new CachedOutputStream();
- if (msg.getAttachments() == null) {
- rmmsg.setContentType((String)msg.get(Message.CONTENT_TYPE));
+ if (msg.getAttachments() == null || msg.getAttachments().isEmpty()) {
+ rmmsg.setContentType(getRootContentType(msg));
IOUtils.copyAndCloseInput(msgContent, cos);
cos.flush();
rmmsg.setContent(cos);
@@ -115,10 +245,20 @@ public final class PersistenceUtils {
MessageImpl msgImpl1 = new MessageImpl();
msgImpl1.setContent(OutputStream.class, cos);
msgImpl1.setAttachments(msg.getAttachments());
- msgImpl1.put(Message.CONTENT_TYPE, msg.get(Message.CONTENT_TYPE));
+ // an outbound MTOM message is stored as an XOP package, so that
it is resent as one
+ boolean xop = MessageUtils.isOutbound(msg) &&
AttachmentUtil.isMtomEnabled(msg);
+ String rootType = getRootContentType(msg);
+ String xopWrappedType = xop ? getXopWrappedType(rootType) : null;
+ // the AttachmentSerializer wraps the SOAP type into the XOP root
part type
+ msgImpl1.put(Message.CONTENT_TYPE, xopWrappedType != null ?
xopWrappedType : rootType);
+ // the encoding of the root part, otherwise the
AttachmentSerializer declares UTF-8
+ Object encoding = msg.get(Message.ENCODING);
+ if (encoding != null) {
+ msgImpl1.put(Message.ENCODING, encoding);
+ }
msgImpl1.setContent(InputStream.class, msgContent);
AttachmentSerializer serializer = new
AttachmentSerializer(msgImpl1);
- serializer.setXop(false);
+ serializer.setXop(xop);
serializer.writeProlog();
// write soap root message into cached output stream
IOUtils.copyAndCloseInput(msgContent, cos);
@@ -143,6 +283,18 @@ public final class PersistenceUtils {
IOUtils.copy(msg.getContent(InputStream.class), cosSoap);
cosSoap.flush();
msg.put(RMMessageConstants.SAVED_CONTENT, cosSoap);
+
+ // restore the SOAP root content type, the multipart type is
recreated when the message is sent
+ String rootType = getRootContentType(msg);
+ String xopWrappedType = getXopWrappedType(rootType);
+ if (xopWrappedType != null) {
+ // the message was an XOP package, so the multipart is
recreated as one
+ rootType = xopWrappedType;
+ msg.put(Message.MTOM_ENABLED, Boolean.TRUE);
+ }
+ msg.put(RMMessageConstants.SAVED_CONTENT_TYPE, rootType);
+ msg.put(Message.CONTENT_TYPE, rootType);
+
// REVISIT -- At the moment references must be hold for
retransmission
// and the final cleanup of the CachedOutputStream.
msg.put(RMMessageConstants.ATTACHMENTS_CLOSEABLE, new Closeable() {
@@ -164,6 +316,8 @@ public final class PersistenceUtils {
});
} else {
msg.put(RMMessageConstants.SAVED_CONTENT, cos);
+ msg.put(Message.CONTENT_TYPE, contentType);
+ msg.put(RMMessageConstants.SAVED_CONTENT_TYPE, contentType);
}
}
diff --git
a/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/soap/RetransmissionQueueImpl.java
b/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/soap/RetransmissionQueueImpl.java
index 2eecdcc819c..9fb8dc35b38 100644
---
a/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/soap/RetransmissionQueueImpl.java
+++
b/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/soap/RetransmissionQueueImpl.java
@@ -59,6 +59,7 @@ import org.apache.cxf.endpoint.DeferredConduitSelector;
import org.apache.cxf.endpoint.Endpoint;
import org.apache.cxf.helpers.DOMUtils;
import org.apache.cxf.interceptor.AbstractOutDatabindingInterceptor;
+import org.apache.cxf.interceptor.AttachmentOutInterceptor;
import org.apache.cxf.interceptor.Fault;
import org.apache.cxf.interceptor.Interceptor;
import org.apache.cxf.io.CachedOutputStream;
@@ -810,8 +811,13 @@ public class RetransmissionQueueImpl implements
RetransmissionQueue {
}
message.setInterceptorChain(retransmitChain);
- // clear flag for SOAP out interceptor so envelope will be written
+ // clear flags for SOAP and attachment out interceptors so
envelope and attachments will be written
message.remove(SoapOutInterceptor.WROTE_ENVELOPE_START);
+ message.remove(AttachmentOutInterceptor.ATTACHMENT_OUT_CHECKED);
+ String savedContentType =
(String)message.get(RMMessageConstants.SAVED_CONTENT_TYPE);
+ if (savedContentType != null) {
+ message.put(org.apache.cxf.message.Message.CONTENT_TYPE,
savedContentType);
+ }
// discard all saved content
Set<Class<?>> formats = message.getContentFormats();
diff --git
a/rt/ws/rm/src/test/java/org/apache/cxf/ws/rm/persistence/PersistenceUtilsTest.java
b/rt/ws/rm/src/test/java/org/apache/cxf/ws/rm/persistence/PersistenceUtilsTest.java
index 3fe324733c8..c765c626bcb 100644
---
a/rt/ws/rm/src/test/java/org/apache/cxf/ws/rm/persistence/PersistenceUtilsTest.java
+++
b/rt/ws/rm/src/test/java/org/apache/cxf/ws/rm/persistence/PersistenceUtilsTest.java
@@ -22,16 +22,24 @@ package org.apache.cxf.ws.rm.persistence;
import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStream;
+import java.io.OutputStream;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeMap;
import jakarta.activation.DataHandler;
import jakarta.mail.util.ByteArrayDataSource;
+import org.apache.cxf.attachment.AttachmentDeserializer;
import org.apache.cxf.attachment.AttachmentImpl;
+import org.apache.cxf.attachment.AttachmentSerializer;
import org.apache.cxf.helpers.IOUtils;
import org.apache.cxf.io.CachedOutputStream;
import org.apache.cxf.message.Attachment;
+import org.apache.cxf.message.Exchange;
+import org.apache.cxf.message.ExchangeImpl;
import org.apache.cxf.message.Message;
import org.apache.cxf.message.MessageImpl;
import org.apache.cxf.ws.rm.RMMessageConstants;
@@ -41,7 +49,9 @@ import
org.apache.cxf.ws.rm.v200702.SequenceAcknowledgement.AcknowledgementRange
import org.junit.Test;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
/**
@@ -122,6 +132,42 @@ public class PersistenceUtilsTest {
assertStartsWith(cos.getInputStream(), SOAP_PART);
}
+ @Test
+ public void testEncodeDecodeRMContentRestoresRootContentType() throws
Exception {
+ String soap12 = "application/soap+xml; action=\"urn:test\"";
+ RMMessage rmmsg = new RMMessage();
+ Message messageImpl = new MessageImpl();
+ // AttachmentOutInterceptor has already replaced the content type with
the multipart one
+ messageImpl.put(Message.CONTENT_TYPE, MULTIPART_TYPE);
+ messageImpl.put(RMMessageConstants.SAVED_CONTENT_TYPE, soap12);
+ addAttachment(messageImpl);
+ PersistenceUtils.encodeRMContent(rmmsg, messageImpl, new
ByteArrayInputStream(SOAP_PART.getBytes()));
+ assertTrue(rmmsg.getContentType().startsWith("multipart/related"));
+
+ Message restored = new MessageImpl();
+ PersistenceUtils.decodeRMContent(rmmsg, restored);
+ assertEquals(1, restored.getAttachments().size());
+ assertEquals(soap12,
restored.get(RMMessageConstants.SAVED_CONTENT_TYPE));
+ assertEquals(soap12, restored.get(Message.CONTENT_TYPE));
+ }
+
+ @Test
+ public void testEncodeRMContentNoAttachmentsUsesSavedContentType() throws
Exception {
+ RMMessage rmmsg = new RMMessage();
+ Message messageImpl = new MessageImpl();
+ messageImpl.put(Message.CONTENT_TYPE, MULTIPART_TYPE);
+ messageImpl.put(RMMessageConstants.SAVED_CONTENT_TYPE, "text/xml");
+ messageImpl.setAttachments(new ArrayList<>());
+ PersistenceUtils.encodeRMContent(rmmsg, messageImpl, new
ByteArrayInputStream(SOAP_PART.getBytes()));
+ assertEquals("text/xml", rmmsg.getContentType());
+ assertStartsWith(rmmsg.getContent().getInputStream(), "<soap:");
+
+ Message restored = new MessageImpl();
+ PersistenceUtils.decodeRMContent(rmmsg, restored);
+ assertEquals("text/xml", restored.get(Message.CONTENT_TYPE));
+ assertEquals("text/xml",
restored.get(RMMessageConstants.SAVED_CONTENT_TYPE));
+ }
+
@Test
public void testDecodeRMContentWithAttachment() throws Exception {
InputStream is =
getClass().getResourceAsStream("SerializedRMMessage.txt");
@@ -138,6 +184,299 @@ public class PersistenceUtilsTest {
CachedOutputStream cos1 = (CachedOutputStream)messageImpl
.get(RMMessageConstants.SAVED_CONTENT);
assertStartsWith(cos1.getInputStream(), "<soap:Envelope");
+ assertNotNull("SAVED_CONTENT_TYPE should be set after decode",
+ messageImpl.get(RMMessageConstants.SAVED_CONTENT_TYPE));
+ String restored = (String) messageImpl.get(Message.CONTENT_TYPE);
+ assertFalse("Content-Type should not be multipart after decode",
+ restored.startsWith("multipart/"));
+ }
+
+ @Test
+ public void testEncodeDecodeInboundPathViaPartHeaders() throws Exception {
+ // Simulates the server (destination) side: inbound message has no
SAVED_CONTENT_TYPE
+ // but AttachmentDeserializer has set ATTACHMENT_PART_HEADERS with the
root Content-Type
+ RMMessage rmmsg = new RMMessage();
+ Message messageImpl = new MessageImpl();
+ messageImpl.put(Message.CONTENT_TYPE, MULTIPART_TYPE);
+ Map<String, List<String>> partHeaders = new
TreeMap<>(String.CASE_INSENSITIVE_ORDER);
+ partHeaders.put("Content-Type", List.of("text/xml; charset=UTF-8"));
+ messageImpl.put(AttachmentDeserializer.ATTACHMENT_PART_HEADERS,
partHeaders);
+ addAttachment(messageImpl);
+
+ PersistenceUtils.encodeRMContent(rmmsg, messageImpl, new
ByteArrayInputStream(SOAP_PART.getBytes()));
+ assertTrue("Persisted content should be multipart",
+ rmmsg.getContentType().startsWith("multipart/related"));
+
+ Message restored = new MessageImpl();
+ PersistenceUtils.decodeRMContent(rmmsg, restored);
+ assertEquals(1, restored.getAttachments().size());
+ String restoredCt = (String)
restored.get(RMMessageConstants.SAVED_CONTENT_TYPE);
+ assertNotNull("SAVED_CONTENT_TYPE should be set on decode",
restoredCt);
+ assertFalse("Restored type should not be multipart",
restoredCt.startsWith("multipart/"));
+ assertFalse("Charset should be stripped from restored type",
+ restoredCt.toLowerCase().contains("charset"));
+ assertEquals("text/xml", restoredCt);
+ }
+
+ @Test
+ public void testEncodeDecodeInboundSoap12WithActionAndCharset() throws
Exception {
+ RMMessage rmmsg = new RMMessage();
+ Message messageImpl = new MessageImpl();
+ messageImpl.put(Message.CONTENT_TYPE, MULTIPART_TYPE);
+ Map<String, List<String>> partHeaders = new
TreeMap<>(String.CASE_INSENSITIVE_ORDER);
+ partHeaders.put("Content-Type", List.of("application/soap+xml;
charset=UTF-8; action=\"urn:test\""));
+ messageImpl.put(AttachmentDeserializer.ATTACHMENT_PART_HEADERS,
partHeaders);
+ addAttachment(messageImpl);
+
+ PersistenceUtils.encodeRMContent(rmmsg, messageImpl, new
ByteArrayInputStream(SOAP_PART.getBytes()));
+
+ Message restored = new MessageImpl();
+ PersistenceUtils.decodeRMContent(rmmsg, restored);
+ String restoredCt = (String)
restored.get(RMMessageConstants.SAVED_CONTENT_TYPE);
+ assertEquals("application/soap+xml; action=\"urn:test\"", restoredCt);
+ }
+
+ @Test
+ public void testNoRootContentTypeDoesNotNestMultipart() throws Exception {
+ // neither SAVED_CONTENT_TYPE nor the part headers are available, only
a multipart type without
+ // the type of its root part
+ RMMessage rmmsg = new RMMessage();
+ Message messageImpl = new MessageImpl();
+ messageImpl.put(Message.CONTENT_TYPE, "multipart/related;
boundary=\"uuid:1\"");
+ addAttachment(messageImpl);
+ PersistenceUtils.encodeRMContent(rmmsg, messageImpl, new
ByteArrayInputStream(SOAP_PART.getBytes()));
+
+ Message restored = new MessageImpl();
+ PersistenceUtils.decodeRMContent(rmmsg, restored);
+ assertEquals(1, restored.getAttachments().size());
+ assertEquals("text/xml",
restored.get(RMMessageConstants.SAVED_CONTENT_TYPE));
+ assertEquals("text/xml", restored.get(Message.CONTENT_TYPE));
+ }
+
+ @Test
+ public void testRootPartEncodingIsPreserved() throws Exception {
+ RMMessage rmmsg = new RMMessage();
+ Message messageImpl = new MessageImpl();
+ messageImpl.put(Message.CONTENT_TYPE, MULTIPART_TYPE);
+ messageImpl.put(RMMessageConstants.SAVED_CONTENT_TYPE, "text/xml");
+ messageImpl.put(Message.ENCODING, "ISO-8859-1");
+ addAttachment(messageImpl);
+ PersistenceUtils.encodeRMContent(rmmsg, messageImpl, new
ByteArrayInputStream(SOAP_PART.getBytes()));
+
+ String stored = IOUtils.toString(rmmsg.getContent().getInputStream());
+ assertTrue(stored, stored.contains("charset=ISO-8859-1"));
+ assertFalse(stored, stored.contains("charset=UTF-8"));
+
+ Message restored = new MessageImpl();
+ PersistenceUtils.decodeRMContent(rmmsg, restored);
+ assertEquals("ISO-8859-1", restored.get(Message.ENCODING));
+ assertEquals("text/xml",
restored.get(RMMessageConstants.SAVED_CONTENT_TYPE));
+ }
+
+ @Test
+ public void testEncodeDecodeMtomRestoresXopPackage() throws Exception {
+ RMMessage rmmsg = encodeOutbound("text/xml", true);
+ assertTrue(rmmsg.getContentType(),
rmmsg.getContentType().contains("type=\"application/xop+xml\""));
+ String stored = IOUtils.toString(rmmsg.getContent().getInputStream());
+ assertTrue(stored, stored.contains("Content-Type: application/xop+xml;
charset=UTF-8; type=\"text/xml\""));
+
+ Message restored = new MessageImpl();
+ PersistenceUtils.decodeRMContent(rmmsg, restored);
+ assertEquals(1, restored.getAttachments().size());
+ assertEquals("text/xml",
restored.get(RMMessageConstants.SAVED_CONTENT_TYPE));
+ assertEquals("text/xml", restored.get(Message.CONTENT_TYPE));
+ assertEquals(Boolean.TRUE, restored.get(Message.MTOM_ENABLED));
+ }
+
+ @Test
+ public void testEncodeDecodeMtomSoap12WithAction() throws Exception {
+ String soap12 = "application/soap+xml; action=\"urn:test\"";
+ RMMessage rmmsg = encodeOutbound(soap12, true);
+
+ Message restored = new MessageImpl();
+ PersistenceUtils.decodeRMContent(rmmsg, restored);
+ assertEquals(soap12,
restored.get(RMMessageConstants.SAVED_CONTENT_TYPE));
+ assertEquals(Boolean.TRUE, restored.get(Message.MTOM_ENABLED));
+ }
+
+ @Test
+ public void testEncodeDecodeSwaDoesNotEnableMtom() throws Exception {
+ RMMessage rmmsg = encodeOutbound("text/xml", false);
+ assertFalse(rmmsg.getContentType(),
rmmsg.getContentType().contains("application/xop+xml"));
+
+ Message restored = new MessageImpl();
+ PersistenceUtils.decodeRMContent(rmmsg, restored);
+ assertEquals("text/xml",
restored.get(RMMessageConstants.SAVED_CONTENT_TYPE));
+ assertNull(restored.get(Message.MTOM_ENABLED));
+ }
+
+ @Test
+ public void testEncodeDecodeInboundXopRootPart() throws Exception {
+ RMMessage rmmsg = new RMMessage();
+ Message messageImpl = new MessageImpl();
+ messageImpl.put(Message.CONTENT_TYPE, MULTIPART_TYPE);
+ Map<String, List<String>> partHeaders = new
TreeMap<>(String.CASE_INSENSITIVE_ORDER);
+ partHeaders.put("Content-Type", List.of("application/xop+xml;
charset=UTF-8; type=\"text/xml\""));
+ messageImpl.put(AttachmentDeserializer.ATTACHMENT_PART_HEADERS,
partHeaders);
+ addAttachment(messageImpl);
+
+ PersistenceUtils.encodeRMContent(rmmsg, messageImpl, new
ByteArrayInputStream(SOAP_PART.getBytes()));
+
+ Message restored = new MessageImpl();
+ PersistenceUtils.decodeRMContent(rmmsg, restored);
+ assertEquals(1, restored.getAttachments().size());
+ assertEquals("text/xml", restored.get(Message.CONTENT_TYPE));
+ assertEquals(Boolean.TRUE, restored.get(Message.MTOM_ENABLED));
+ }
+
+ @Test
+ public void testMtomContentTypeIsNotOutbound() throws Exception {
+ // a stored message is only an XOP package if it was sent with MTOM,
not if it was received with it
+ RMMessage rmmsg = new RMMessage();
+ Message messageImpl = new MessageImpl();
+ messageImpl.put(Message.CONTENT_TYPE, MULTIPART_TYPE);
+ messageImpl.put(RMMessageConstants.SAVED_CONTENT_TYPE, "text/xml");
+ messageImpl.put(Message.MTOM_ENABLED, Boolean.TRUE);
+ addAttachment(messageImpl);
+ PersistenceUtils.encodeRMContent(rmmsg, messageImpl, new
ByteArrayInputStream(SOAP_PART.getBytes()));
+ assertFalse(rmmsg.getContentType(),
rmmsg.getContentType().contains("application/xop+xml"));
+ }
+
+ @Test
+ public void testEncodeDecodeOutboundMtomWithoutSavedContentType() throws
Exception {
+ // the SOAP type is taken from the start-info of the multipart type
written by AttachmentOutInterceptor
+ RMMessage rmmsg = new RMMessage();
+ Message messageImpl = new MessageImpl();
+ messageImpl.put(Message.CONTENT_TYPE, "multipart/related;
type=\"application/xop+xml\";"
+ + " boundary=\"uuid:1\"; start=\"<[email protected]>\";"
+ + " start-info=\"application/soap+xml; action=\\\"urn:test\\\"\"");
+ messageImpl.put(Message.MTOM_ENABLED, Boolean.TRUE);
+ setOutbound(messageImpl);
+ addAttachment(messageImpl);
+ PersistenceUtils.encodeRMContent(rmmsg, messageImpl, new
ByteArrayInputStream(SOAP_PART.getBytes()));
+ String stored = IOUtils.toString(rmmsg.getContent().getInputStream());
+ assertTrue(stored, stored.contains("Content-Type: application/xop+xml;
charset=UTF-8;"
+ + " type=\"application/soap+xml; action=\\\"urn:test\\\"\""));
+
+ Message restored = new MessageImpl();
+ PersistenceUtils.decodeRMContent(rmmsg, restored);
+ assertEquals("application/soap+xml; action=\"urn:test\"",
restored.get(Message.CONTENT_TYPE));
+ assertEquals(Boolean.TRUE, restored.get(Message.MTOM_ENABLED));
+ }
+
+ @Test
+ public void testDecodeInboundXopRootWithCharsetInType() throws Exception {
+ // the charset in the quoted type parameter is not a parameter of the
XOP root part type
+ Message restored = encodeDecodeInbound("application/xop+xml;
type=\"application/soap+xml; charset=utf-8\"");
+ assertEquals("application/soap+xml",
restored.get(Message.CONTENT_TYPE));
+ assertEquals(Boolean.TRUE, restored.get(Message.MTOM_ENABLED));
+ }
+
+ @Test
+ public void testDecodeInboundXopRootWithEscapedAction() throws Exception {
+ Message restored = encodeDecodeInbound(
+ "application/xop+xml; charset=UTF-8; type=\"application/soap+xml;
action=\\\"urn:test\\\"\"");
+ assertEquals("application/soap+xml; action=\"urn:test\"",
restored.get(Message.CONTENT_TYPE));
+ assertEquals(Boolean.TRUE, restored.get(Message.MTOM_ENABLED));
+ }
+
+ @Test
+ public void testInboundRootPartWithSeveralContentTypes() throws Exception {
+ Message restored = encodeDecodeInbound("text/xml; charset=UTF-8",
"application/soap+xml");
+ assertEquals("text/xml", restored.get(Message.CONTENT_TYPE));
+ }
+
+ /** Decodes a stored message whose root part has the multipart type of the
message as its type */
+ @Test
+ public void testDecodeSoap12MessageWithMultipartRootType() throws
Exception {
+ RMMessage rmmsg = encodeWithMultipartRootType("multipart/related;
type=\"application/soap+xml\";"
+ + " boundary=\"uuid:1\"; start=\"<[email protected]>\";"
+ + " start-info=\"application/soap+xml; action=\\\"urn:test\\\"\"");
+
+ Message restored = new MessageImpl();
+ PersistenceUtils.decodeRMContent(rmmsg, restored);
+ assertEquals(1, restored.getAttachments().size());
+ assertEquals("application/soap+xml; action=\"urn:test\"",
restored.get(Message.CONTENT_TYPE));
+ assertNull(restored.get(Message.MTOM_ENABLED));
+ }
+
+ @Test
+ public void testDecodeSoap12MtomMessageWithMultipartRootType() throws
Exception {
+ RMMessage rmmsg = encodeWithMultipartRootType("multipart/related;
type=\"application/xop+xml\";"
+ + " boundary=\"uuid:1\"; start=\"<[email protected]>\";"
+ + " start-info=\"application/soap+xml; action=\\\"urn:test\\\"\"");
+
+ Message restored = new MessageImpl();
+ PersistenceUtils.decodeRMContent(rmmsg, restored);
+ assertEquals("application/soap+xml; action=\"urn:test\"",
restored.get(Message.CONTENT_TYPE));
+ assertEquals(Boolean.TRUE, restored.get(Message.MTOM_ENABLED));
+ }
+
+ @Test
+ public void testDecodeSoap11MessageWithMultipartRootType() throws
Exception {
+ RMMessage rmmsg = encodeWithMultipartRootType("multipart/related;
type=\"text/xml\"; boundary=\"uuid:1\";"
+ + " start=\"<[email protected]>\";
start-info=\"text/xml\"");
+
+ Message restored = new MessageImpl();
+ PersistenceUtils.decodeRMContent(rmmsg, restored);
+ assertEquals("text/xml", restored.get(Message.CONTENT_TYPE));
+ assertNull(restored.get(Message.MTOM_ENABLED));
+ }
+
+ /** Encodes and decodes an inbound message whose root part has the given
Content-Type headers */
+ private static Message encodeDecodeInbound(String... rootTypes) throws
IOException {
+ RMMessage rmmsg = new RMMessage();
+ Message messageImpl = new MessageImpl();
+ messageImpl.put(Message.CONTENT_TYPE, MULTIPART_TYPE);
+ Map<String, List<String>> partHeaders = new
TreeMap<>(String.CASE_INSENSITIVE_ORDER);
+ partHeaders.put("Content-Type", List.of(rootTypes));
+ messageImpl.put(AttachmentDeserializer.ATTACHMENT_PART_HEADERS,
partHeaders);
+ addAttachment(messageImpl);
+ PersistenceUtils.encodeRMContent(rmmsg, messageImpl, new
ByteArrayInputStream(SOAP_PART.getBytes()));
+
+ Message restored = new MessageImpl();
+ PersistenceUtils.decodeRMContent(rmmsg, restored);
+ assertEquals(1, restored.getAttachments().size());
+ return restored;
+ }
+
+ /** Encodes a message with an attachment, with the given multipart type as
the type of its root part */
+ private static RMMessage encodeWithMultipartRootType(String multipartType)
throws IOException {
+ CachedOutputStream cos = new CachedOutputStream();
+ Message msg = new MessageImpl();
+ msg.setContent(OutputStream.class, cos);
+ addAttachment(msg);
+ msg.put(Message.CONTENT_TYPE, multipartType);
+ AttachmentSerializer serializer = new AttachmentSerializer(msg);
+ serializer.setXop(false);
+ serializer.writeProlog();
+ cos.write(SOAP_PART.getBytes(StandardCharsets.UTF_8));
+ cos.flush();
+ serializer.writeAttachments();
+ RMMessage rmmsg = new RMMessage();
+ rmmsg.setContentType((String)msg.get(Message.CONTENT_TYPE));
+ rmmsg.setContent(cos);
+ return rmmsg;
+ }
+
+ private static void setOutbound(Message msg) {
+ Exchange exchange = new ExchangeImpl();
+ exchange.setOutMessage(msg);
+ }
+
+ /** Encodes a captured outbound message with an attachment, written with
or without MTOM */
+ private static RMMessage encodeOutbound(String soapContentType, boolean
mtom) throws IOException {
+ RMMessage rmmsg = new RMMessage();
+ Message messageImpl = new MessageImpl();
+ messageImpl.put(Message.CONTENT_TYPE, MULTIPART_TYPE);
+ messageImpl.put(RMMessageConstants.SAVED_CONTENT_TYPE,
soapContentType);
+ if (mtom) {
+ messageImpl.put(Message.MTOM_ENABLED, Boolean.TRUE);
+ }
+ setOutbound(messageImpl);
+ addAttachment(messageImpl);
+ PersistenceUtils.encodeRMContent(rmmsg, messageImpl, new
ByteArrayInputStream(SOAP_PART.getBytes()));
+ return rmmsg;
}
private static void addAttachment(Message msg) throws IOException {
diff --git
a/systests/ws-rm/src/test/java/org/apache/cxf/systest/ws/rm/RetransmissionAttachmentTest.java
b/systests/ws-rm/src/test/java/org/apache/cxf/systest/ws/rm/RetransmissionAttachmentTest.java
new file mode 100644
index 00000000000..59f3136d68e
--- /dev/null
+++
b/systests/ws-rm/src/test/java/org/apache/cxf/systest/ws/rm/RetransmissionAttachmentTest.java
@@ -0,0 +1,299 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.cxf.systest.ws.rm;
+
+import java.io.InputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import jakarta.activation.DataHandler;
+import jakarta.mail.util.ByteArrayDataSource;
+import jakarta.xml.ws.BindingProvider;
+import jakarta.xml.ws.Endpoint;
+import jakarta.xml.ws.soap.SOAPBinding;
+import org.apache.cxf.Bus;
+import org.apache.cxf.BusFactory;
+import org.apache.cxf.attachment.AttachmentImpl;
+import org.apache.cxf.bus.spring.SpringBusFactory;
+import org.apache.cxf.frontend.ClientProxy;
+import org.apache.cxf.greeter_control.Greeter;
+import org.apache.cxf.greeter_control.GreeterService;
+import org.apache.cxf.jaxws.EndpointImpl;
+import org.apache.cxf.message.Attachment;
+import org.apache.cxf.message.Message;
+import org.apache.cxf.phase.AbstractPhaseInterceptor;
+import org.apache.cxf.phase.Phase;
+import org.apache.cxf.service.model.BindingOperationInfo;
+import org.apache.cxf.testutil.common.AbstractBusClientServerTestBase;
+import org.apache.cxf.testutil.common.AbstractBusTestServerBase;
+import org.apache.cxf.ws.rm.RMManager;
+import org.apache.cxf.ws.rm.RMUtils;
+import org.apache.cxf.ws.rm.SourceSequence;
+import org.apache.cxf.ws.rm.persistence.RMMessage;
+import org.apache.cxf.ws.rm.persistence.RMStore;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Tests the retransmission of messages with attachments, which are written by
the
+ * AttachmentOutInterceptor again for each retransmission.
+ */
+public class RetransmissionAttachmentTest extends
AbstractBusClientServerTestBase {
+ public static final String PORT =
allocatePort(RetransmissionAttachmentTest.class);
+
+ static final byte[] ATTACHMENT = "attachment
content".getBytes(StandardCharsets.UTF_8);
+ /** The Content-Type and the attachments of the greetMeOneWay requests
received by the server */
+ static final BlockingQueue<String> RECEIVED_CONTENT_TYPES = new
LinkedBlockingQueue<>();
+ static final BlockingQueue<byte[]> RECEIVED_ATTACHMENTS = new
LinkedBlockingQueue<>();
+
+ private static final String CONFIG =
"/org/apache/cxf/systest/ws/rm/simple.xml";
+ private static final String PERSISTENT_CONFIG =
"/org/apache/cxf/systest/ws/rm/persistent.xml";
+ private static final String DB_NAME = "rat-client";
+ /** Each test uses its own in-memory database, as an in-memory database
lives as long as the JVM */
+ private static final AtomicInteger DB_COUNT = new AtomicInteger();
+
+ private Bus bus;
+
+ public static class Server extends AbstractBusTestServerBase {
+ protected void run() {
+ Bus bus = new SpringBusFactory().createBus(CONFIG);
+ BusFactory.setDefaultBus(bus);
+ setBus(bus);
+ EndpointImpl ep = (EndpointImpl)Endpoint.create(new GreeterImpl());
+ ep.getInInterceptors().add(new RequestRecorder());
+ ep.publish("http://localhost:"; + PORT +
"/SoapContext/GreeterPort");
+ }
+ }
+
+ static class RequestRecorder extends AbstractPhaseInterceptor<Message> {
+ RequestRecorder() {
+ super(Phase.PRE_INVOKE);
+ }
+
+ public void handleMessage(Message message) {
+ if (!isGreetMeOneWay(message)) {
+ return;
+ }
+
RECEIVED_CONTENT_TYPES.add((String)message.get(Message.CONTENT_TYPE));
+ if (message.getAttachments() != null) {
+ for (Attachment a : message.getAttachments()) {
+ try (InputStream in = a.getDataHandler().getInputStream())
{
+ RECEIVED_ATTACHMENTS.add(in.readAllBytes());
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+ }
+ }
+
+ /** Adds a SwA attachment to the greetMeOneWay requests */
+ static class AttachmentAdder extends AbstractPhaseInterceptor<Message> {
+ AttachmentAdder() {
+ super(Phase.PRE_LOGICAL);
+ }
+
+ public void handleMessage(Message message) {
+ if (isGreetMeOneWay(message)) {
+ List<Attachment> attachments = new ArrayList<>();
+ attachments.add(new
AttachmentImpl("[email protected]",
+ new DataHandler(new ByteArrayDataSource(ATTACHMENT,
"text/plain"))));
+ message.setAttachments(attachments);
+ }
+ }
+ }
+
+ static boolean isGreetMeOneWay(Message message) {
+ BindingOperationInfo boi =
message.getExchange().getBindingOperationInfo();
+ return boi != null &&
"greetMeOneWay".equals(boi.getName().getLocalPart());
+ }
+
+ @BeforeClass
+ public static void startServers() throws Exception {
+ assertTrue("server did not launch correctly",
launchServer(Server.class, true));
+ }
+
+ @Before
+ public void clearReceived() {
+ RECEIVED_CONTENT_TYPES.clear();
+ RECEIVED_ATTACHMENTS.clear();
+ }
+
+ @After
+ public void shutdownBus() {
+ if (bus != null) {
+ bus.shutdown(true);
+ bus = null;
+ }
+ }
+
+ @Test
+ public void testAttachmentRetransmitted() throws Exception {
+ Greeter greeter = createGreeter();
+ ClientProxy.getClient(greeter).getOutInterceptors().add(new
AttachmentAdder());
+
+ greeter.greetMeOneWay("attachment");
+
+ String contentType = RECEIVED_CONTENT_TYPES.poll(10, TimeUnit.SECONDS);
+ assertNotNull("the retransmitted message was not received",
contentType);
+ assertTrue(contentType, contentType.startsWith("multipart/related;
type=\"text/xml\";"));
+ byte[] attachment = RECEIVED_ATTACHMENTS.poll(1, TimeUnit.SECONDS);
+ assertNotNull("no attachment was received", attachment);
+ assertArrayEquals(ATTACHMENT, attachment);
+ }
+
+ @Test
+ public void testMtomRetransmitted() throws Exception {
+ Greeter greeter = createGreeter();
+
((SOAPBinding)((BindingProvider)greeter).getBinding()).setMTOMEnabled(true);
+
+ greeter.greetMeOneWay("mtom");
+
+ String contentType = RECEIVED_CONTENT_TYPES.poll(10, TimeUnit.SECONDS);
+ assertNotNull("the retransmitted message was not received",
contentType);
+ assertTrue(contentType,
contentType.contains("start-info=\"text/xml\""));
+ }
+
+ /**
+ * The first transmission is lost and the client is restarted before the
message is retransmitted,
+ * so the message is retransmitted after having been recovered from the
persistent store.
+ */
+ @Test
+ public void testAttachmentRetransmittedAfterRecovery() throws Exception {
+ String persisted = retransmitAfterRecovery(false, "multipart/related;
type=\"text/xml\";");
+ // the content type of the root part must be the one of the SOAP
message, not the multipart one
+ assertTrue(persisted, persisted.contains("Content-Type: text/xml"));
+ assertFalse(persisted, persisted.contains("multipart/related"));
+ }
+
+ /**
+ * Same as testAttachmentRetransmittedAfterRecovery with MTOM, the
recovered message must be
+ * retransmitted as an XOP package again.
+ */
+ @Test
+ public void testMtomAttachmentRetransmittedAfterRecovery() throws
Exception {
+ String persisted = retransmitAfterRecovery(true, "multipart/related;
type=\"application/xop+xml\";");
+ assertTrue(persisted,
+ persisted.contains("Content-Type: application/xop+xml;
charset=UTF-8; type=\"text/xml\""));
+ assertFalse(persisted, persisted.contains("multipart/related"));
+ }
+
+ /**
+ * Sends a message with an attachment which is lost and persisted, then
restarts the client, which
+ * recovers and retransmits the message. Returns the persisted message.
+ */
+ private String retransmitAfterRecovery(boolean mtom, String
expectedContentTypePrefix) throws Exception {
+ String dbName = DB_NAME + "-" + DB_COUNT.getAndIncrement();
+
+ // the first client loses the message, which is persisted with its
attachment
+ Greeter greeter = createPersistentGreeter(dbName, 60000L, true);
+ ClientProxy.getClient(greeter).getOutInterceptors().add(new
AttachmentAdder());
+
((SOAPBinding)((BindingProvider)greeter).getBinding()).setMTOMEnabled(mtom);
+ greeter.greetMeOneWay("attachment");
+
+ RMStore store = bus.getExtension(RMManager.class).getStore();
+ assertNotNull(store);
+ String id =
RMUtils.getEndpointIdentifier(ClientProxy.getClient(greeter).getEndpoint());
+ Collection<SourceSequence> sequences = store.getSourceSequences(id);
+ assertEquals(1, sequences.size());
+ Collection<RMMessage> messages =
store.getMessages(sequences.iterator().next().getIdentifier(), true);
+ assertEquals(1, messages.size());
+ String persistedType = messages.iterator().next().getContentType();
+ assertTrue(persistedType,
persistedType.startsWith("multipart/related"));
+ String persisted;
+ try (InputStream in =
messages.iterator().next().getContent().getInputStream()) {
+ persisted = new String(in.readAllBytes(), StandardCharsets.UTF_8);
+ }
+ assertTrue("the message must not be received before the restart",
RECEIVED_CONTENT_TYPES.isEmpty());
+
+ bus.shutdown(true);
+ bus = null;
+
+ // the second client recovers the message from the store and
retransmits it
+ createPersistentGreeter(dbName, 2000L, false);
+
+ String contentType = RECEIVED_CONTENT_TYPES.poll(20, TimeUnit.SECONDS);
+ assertNotNull("the recovered message was not retransmitted",
contentType);
+ assertTrue(contentType,
contentType.startsWith(expectedContentTypePrefix));
+ byte[] attachment = RECEIVED_ATTACHMENTS.poll(1, TimeUnit.SECONDS);
+ assertNotNull("no attachment was received", attachment);
+ assertArrayEquals(ATTACHMENT, attachment);
+ assertTrue("exactly one attachment is expected",
RECEIVED_ATTACHMENTS.isEmpty());
+ return persisted;
+ }
+
+ /**
+ * Creates a client whose first transmission of each application message
is lost, so that
+ * the server only receives the retransmission.
+ */
+ private Greeter createGreeter() throws Exception {
+ bus = new SpringBusFactory().createBus(CONFIG);
+ BusFactory.setDefaultBus(bus);
+ BusFactory.setThreadDefaultBus(bus);
+
bus.getExtension(RMManager.class).getConfiguration().setBaseRetransmissionInterval(2000L);
+ MessageLossSimulator loser = new MessageLossSimulator();
+ loser.setMode(-1);
+ bus.getOutInterceptors().add(loser);
+
+ Greeter greeter = new GreeterService().getGreeterPort();
+ updateAddressPort(greeter, PORT);
+ return greeter;
+ }
+
+ /**
+ * Creates a client using the persistent store. The messages are recovered
from the store when the
+ * client is created, so the retransmission interval is set before.
+ */
+ private Greeter createPersistentGreeter(String dbName, long
retransmissionInterval, boolean lose)
+ throws Exception {
+ System.setProperty("db.name", dbName);
+ try {
+ bus = new SpringBusFactory().createBus(PERSISTENT_CONFIG);
+ } finally {
+ System.clearProperty("db.name");
+ }
+ BusFactory.setDefaultBus(bus);
+ BusFactory.setThreadDefaultBus(bus);
+
bus.getExtension(RMManager.class).getConfiguration().setBaseRetransmissionInterval(retransmissionInterval);
+ if (lose) {
+ MessageLossSimulator loser = new MessageLossSimulator();
+ loser.setMode(-1);
+ bus.getOutInterceptors().add(loser);
+ }
+ Greeter greeter = new GreeterService().getGreeterPort();
+ updateAddressPort(greeter, PORT);
+ return greeter;
+ }
+}
diff --git
a/systests/ws-rm/src/test/java/org/apache/cxf/systest/ws/rm/ServerAttachmentRecoveryTest.java
b/systests/ws-rm/src/test/java/org/apache/cxf/systest/ws/rm/ServerAttachmentRecoveryTest.java
new file mode 100644
index 00000000000..95b19808351
--- /dev/null
+++
b/systests/ws-rm/src/test/java/org/apache/cxf/systest/ws/rm/ServerAttachmentRecoveryTest.java
@@ -0,0 +1,219 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.cxf.systest.ws.rm;
+
+import java.io.InputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.Collection;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.cxf.Bus;
+import org.apache.cxf.BusFactory;
+import org.apache.cxf.bus.spring.SpringBusFactory;
+import org.apache.cxf.frontend.ClientProxy;
+import org.apache.cxf.greeter_control.Greeter;
+import org.apache.cxf.greeter_control.GreeterService;
+import org.apache.cxf.jaxws.EndpointImpl;
+import org.apache.cxf.message.Attachment;
+import org.apache.cxf.message.Message;
+import org.apache.cxf.phase.AbstractPhaseInterceptor;
+import org.apache.cxf.phase.Phase;
+import org.apache.cxf.testutil.common.AbstractBusClientServerTestBase;
+import org.apache.cxf.ws.rm.DestinationSequence;
+import org.apache.cxf.ws.rm.RMManager;
+import org.apache.cxf.ws.rm.RMUtils;
+import org.apache.cxf.ws.rm.manager.RetryPolicyType;
+import org.apache.cxf.ws.rm.persistence.RMMessage;
+import org.apache.cxf.ws.rm.persistence.RMStore;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+/**
+ * Tests that an inbound message with an attachment, which could not be
delivered and is persisted
+ * by the destination, is redelivered with its attachment and the SOAP content
type after the server
+ * has been restarted.
+ */
+public class ServerAttachmentRecoveryTest extends
AbstractBusClientServerTestBase {
+ public static final String PORT =
allocatePort(ServerAttachmentRecoveryTest.class);
+
+ private static final String SERVER_CONFIG =
"/org/apache/cxf/systest/ws/rm/sync-ack-persistent-server.xml";
+ private static final String CLIENT_CONFIG =
"/org/apache/cxf/systest/ws/rm/simple.xml";
+ private static final String DB_NAME = "sarec-server";
+ /** Each test uses its own in-memory database, as an in-memory database
lives as long as the JVM */
+ private static final AtomicInteger DB_COUNT = new AtomicInteger();
+ private static final String ADDRESS = "http://localhost:"; + PORT +
"/SoapContext/GreeterPort";
+
+ private static final BlockingQueue<String> DELIVERED_CONTENT_TYPES = new
LinkedBlockingQueue<>();
+ private static final BlockingQueue<byte[]> DELIVERED_ATTACHMENTS = new
LinkedBlockingQueue<>();
+
+ private String dbName;
+ private Bus serverBus;
+ private Bus clientBus;
+ private EndpointImpl endpoint;
+
+ /** Fails the delivery or records the delivered message, depending on the
state of the server. */
+ static class DeliveryRecorder extends AbstractPhaseInterceptor<Message> {
+ private final boolean fail;
+
+ DeliveryRecorder(boolean fail) {
+ super(Phase.PRE_INVOKE);
+ this.fail = fail;
+ }
+
+ public void handleMessage(Message message) {
+ if (!RetransmissionAttachmentTest.isGreetMeOneWay(message)) {
+ return;
+ }
+ if (fail) {
+ throw new RuntimeException("intentional delivery failure");
+ }
+
DELIVERED_CONTENT_TYPES.add((String)message.get(Message.CONTENT_TYPE));
+ if (message.getAttachments() != null) {
+ for (Attachment a : message.getAttachments()) {
+ try (InputStream in = a.getDataHandler().getInputStream())
{
+ DELIVERED_ATTACHMENTS.add(in.readAllBytes());
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+ }
+ }
+
+ @Before
+ public void clearDelivered() throws Exception {
+ dbName = DB_NAME + "-" + DB_COUNT.getAndIncrement();
+ DELIVERED_CONTENT_TYPES.clear();
+ DELIVERED_ATTACHMENTS.clear();
+ }
+
+ @After
+ public void shutdownBuses() {
+ stopServer();
+ if (clientBus != null) {
+ clientBus.shutdown(true);
+ clientBus = null;
+ }
+ }
+
+ @Test
+ public void testAttachmentRedeliveredAfterServerRestart() throws Exception
{
+ startServer(true);
+ Greeter greeter = createGreeter();
+ ClientProxy.getClient(greeter).getOutInterceptors().add(new
RetransmissionAttachmentTest.AttachmentAdder());
+ greeter.greetMeOneWay("attachment");
+
+ // the message is persisted by the destination although it could not
be delivered
+ RMManager manager = serverBus.getExtension(RMManager.class);
+ RMStore store = manager.getStore();
+ assertNotNull(store);
+ String id =
RMUtils.getEndpointIdentifier(endpoint.getServer().getEndpoint());
+ RMMessage persisted = awaitPersistedMessage(store, id);
+ assertTrue(persisted.getContentType(),
persisted.getContentType().startsWith("multipart/related"));
+ String content;
+ try (InputStream in = persisted.getContent().getInputStream()) {
+ content = new String(in.readAllBytes(), StandardCharsets.UTF_8);
+ }
+ // the content type of the root part is the one of the SOAP message
and not a nested multipart type
+ assertTrue(content, content.contains("Content-Type: text/xml"));
+ assertFalse(content, content.contains("multipart/related"));
+ assertTrue("the message must not be delivered before the restart",
DELIVERED_CONTENT_TYPES.isEmpty());
+
+ // restart the server, which recovers the message from the store and
redelivers it
+ stopServer();
+ DELIVERED_CONTENT_TYPES.clear();
+ DELIVERED_ATTACHMENTS.clear();
+ startServer(false);
+
+ String contentType = DELIVERED_CONTENT_TYPES.poll(30,
TimeUnit.SECONDS);
+ assertNotNull("the recovered message was not redelivered",
contentType);
+ assertFalse(contentType, contentType.startsWith("multipart/"));
+ assertTrue(contentType, contentType.startsWith("text/xml"));
+ byte[] attachment = DELIVERED_ATTACHMENTS.poll(1, TimeUnit.SECONDS);
+ assertNotNull("the attachment was lost", attachment);
+ assertArrayEquals(RetransmissionAttachmentTest.ATTACHMENT, attachment);
+ assertTrue("exactly one attachment is expected",
DELIVERED_ATTACHMENTS.isEmpty());
+ }
+
+ private RMMessage awaitPersistedMessage(RMStore store, String endpointId)
throws Exception {
+ for (int i = 0; i < 100; i++) {
+ Collection<DestinationSequence> sequences =
store.getDestinationSequences(endpointId);
+ if (sequences.size() == 1) {
+ Collection<RMMessage> messages =
store.getMessages(sequences.iterator().next().getIdentifier(), false);
+ if (messages.size() == 1) {
+ return messages.iterator().next();
+ }
+ }
+ Thread.sleep(100);
+ }
+ fail("the message was not persisted by the destination");
+ return null;
+ }
+
+ private void startServer(boolean fail) {
+ System.setProperty("db.name", dbName);
+ try {
+ serverBus = new SpringBusFactory().createBus(SERVER_CONFIG);
+ } finally {
+ System.clearProperty("db.name");
+ }
+ RetryPolicyType retryPolicy = new RetryPolicyType();
+ retryPolicy.setMaxRetries(-1);
+
serverBus.getExtension(RMManager.class).getDestinationPolicy().setRetryPolicy(retryPolicy);
+
serverBus.getExtension(RMManager.class).getConfiguration().setAcknowledgementInterval(0L);
+
+ endpoint = new EndpointImpl(serverBus, new GreeterImpl());
+ endpoint.getInInterceptors().add(new DeliveryRecorder(fail));
+ endpoint.publish(ADDRESS);
+ }
+
+ private void stopServer() {
+ if (endpoint != null) {
+ endpoint.stop();
+ endpoint = null;
+ }
+ if (serverBus != null) {
+ serverBus.shutdown(true);
+ serverBus = null;
+ }
+ }
+
+ private Greeter createGreeter() throws Exception {
+ clientBus = new SpringBusFactory().createBus(CLIENT_CONFIG);
+ BusFactory.setDefaultBus(clientBus);
+ BusFactory.setThreadDefaultBus(clientBus);
+ // the client must not retransmit, only the server redelivers
+
clientBus.getExtension(RMManager.class).getConfiguration().setBaseRetransmissionInterval(60000L);
+ Greeter greeter = new GreeterService().getGreeterPort();
+ updateAddressPort(greeter, PORT);
+ return greeter;
+ }
+}
