This is an automated email from the ASF dual-hosted git repository.
reta 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 e28d4a77a7 CXF-9057: Chunked Stream is closed regularly when Exception
is thrown (MTOM) (#2234)
e28d4a77a7 is described below
commit e28d4a77a7ee59d03219ff4767eae1d7bedeb207
Author: Andriy Redko <[email protected]>
AuthorDate: Sat Feb 15 10:42:30 2025 -0500
CXF-9057: Chunked Stream is closed regularly when Exception is thrown
(MTOM) (#2234)
---
.../cxf/phase/AbortedInvocationException.java | 48 +++++++++
.../apache/cxf/phase/PhaseInterceptorChain.java | 3 +
.../interceptor/Soap11FaultOutInterceptor.java | 8 +-
.../interceptor/Soap12FaultOutInterceptor.java | 8 +-
.../transport/http/AbstractHTTPDestination.java | 29 +++++-
...st.java => AbstractAttachmentChunkingTest.java} | 75 +++-----------
.../cxf/systest/jaxws/AttachmentChunkingTest.java | 101 ++-----------------
...ngTest.java => AttachmentMtomChunkingTest.java} | 108 +++++----------------
.../src/test/resources/attachments/cxf9057.wsdl | 1 +
9 files changed, 137 insertions(+), 244 deletions(-)
diff --git
a/core/src/main/java/org/apache/cxf/phase/AbortedInvocationException.java
b/core/src/main/java/org/apache/cxf/phase/AbortedInvocationException.java
new file mode 100644
index 0000000000..ee2553d1af
--- /dev/null
+++ b/core/src/main/java/org/apache/cxf/phase/AbortedInvocationException.java
@@ -0,0 +1,48 @@
+/**
+ * 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.phase;
+
+/**
+ * Represents transport-specific exceptions which are used to indicate that
+ * a given invocation was suspended
+ */
+public class AbortedInvocationException extends RuntimeException {
+
+ private static final long serialVersionUID = 6889545463301144757L;
+
+
+ public AbortedInvocationException(Throwable cause) {
+ super(cause);
+ }
+
+ public AbortedInvocationException() {
+ }
+
+
+ /**
+ * Returns a transport-specific runtime exception
+ * @return RuntimeException the transport-specific runtime exception,
+ * can be null for asynchronous transports
+ */
+ public RuntimeException getRuntimeException() {
+ Throwable ex = getCause();
+ return ex instanceof RuntimeException ? (RuntimeException)ex : null;
+ }
+}
diff --git a/core/src/main/java/org/apache/cxf/phase/PhaseInterceptorChain.java
b/core/src/main/java/org/apache/cxf/phase/PhaseInterceptorChain.java
index c8a52a8e95..316e58d26f 100644
--- a/core/src/main/java/org/apache/cxf/phase/PhaseInterceptorChain.java
+++ b/core/src/main/java/org/apache/cxf/phase/PhaseInterceptorChain.java
@@ -325,6 +325,9 @@ public class PhaseInterceptorChain implements
InterceptorChain {
}
pause();
throw ex;
+ } catch (AbortedInvocationException ex) {
+ abort();
+ throw ex;
} catch (RuntimeException ex) {
if (!faultOccurred) {
faultOccurred = true;
diff --git
a/rt/bindings/soap/src/main/java/org/apache/cxf/binding/soap/interceptor/Soap11FaultOutInterceptor.java
b/rt/bindings/soap/src/main/java/org/apache/cxf/binding/soap/interceptor/Soap11FaultOutInterceptor.java
index 5e612a5932..95ebee9f7c 100644
---
a/rt/bindings/soap/src/main/java/org/apache/cxf/binding/soap/interceptor/Soap11FaultOutInterceptor.java
+++
b/rt/bindings/soap/src/main/java/org/apache/cxf/binding/soap/interceptor/Soap11FaultOutInterceptor.java
@@ -37,6 +37,7 @@ import org.apache.cxf.common.util.StringUtils;
import org.apache.cxf.interceptor.Fault;
import org.apache.cxf.message.Message;
import org.apache.cxf.message.MessageUtils;
+import org.apache.cxf.phase.AbortedInvocationException;
import org.apache.cxf.phase.Phase;
import org.apache.cxf.staxutils.StaxUtils;
@@ -69,7 +70,12 @@ public class Soap11FaultOutInterceptor extends
AbstractSoapInterceptor {
// have been streaming some data already and may not be able to
inject a fault in the middle
// of the data transfer.
if (MessageUtils.getContextualBoolean(message,
Message.PARTIAL_ATTACHMENTS_MESSAGE, false)) {
- throw f;
+ // Signal that response has to be aborted midway
+ if (MessageUtils.getContextualBoolean(message,
Message.MTOM_ENABLED, false)) {
+ throw new AbortedInvocationException(f);
+ } else {
+ throw f;
+ }
}
XMLStreamWriter writer = message.getContent(XMLStreamWriter.class);
diff --git
a/rt/bindings/soap/src/main/java/org/apache/cxf/binding/soap/interceptor/Soap12FaultOutInterceptor.java
b/rt/bindings/soap/src/main/java/org/apache/cxf/binding/soap/interceptor/Soap12FaultOutInterceptor.java
index 35cd7498a6..e766282c02 100644
---
a/rt/bindings/soap/src/main/java/org/apache/cxf/binding/soap/interceptor/Soap12FaultOutInterceptor.java
+++
b/rt/bindings/soap/src/main/java/org/apache/cxf/binding/soap/interceptor/Soap12FaultOutInterceptor.java
@@ -38,6 +38,7 @@ import org.apache.cxf.common.util.StringUtils;
import org.apache.cxf.interceptor.Fault;
import org.apache.cxf.message.Message;
import org.apache.cxf.message.MessageUtils;
+import org.apache.cxf.phase.AbortedInvocationException;
import org.apache.cxf.phase.Phase;
import org.apache.cxf.staxutils.StaxUtils;
@@ -71,7 +72,12 @@ public class Soap12FaultOutInterceptor extends
AbstractSoapInterceptor {
// have been streaming some data already and may not be able to
inject a fault in the middle
// of the data transfer.
if (MessageUtils.getContextualBoolean(message,
Message.PARTIAL_ATTACHMENTS_MESSAGE, false)) {
- throw f;
+ // Signal that response has to be aborted midway
+ if (MessageUtils.getContextualBoolean(message,
Message.MTOM_ENABLED, false)) {
+ throw new AbortedInvocationException(f);
+ } else {
+ throw f;
+ }
}
XMLStreamWriter writer = message.getContent(XMLStreamWriter.class);
diff --git
a/rt/transports/http/src/main/java/org/apache/cxf/transport/http/AbstractHTTPDestination.java
b/rt/transports/http/src/main/java/org/apache/cxf/transport/http/AbstractHTTPDestination.java
index 566e05a1b3..3433fa91a4 100644
---
a/rt/transports/http/src/main/java/org/apache/cxf/transport/http/AbstractHTTPDestination.java
+++
b/rt/transports/http/src/main/java/org/apache/cxf/transport/http/AbstractHTTPDestination.java
@@ -63,6 +63,7 @@ import org.apache.cxf.message.ExchangeImpl;
import org.apache.cxf.message.Message;
import org.apache.cxf.message.MessageImpl;
import org.apache.cxf.message.MessageUtils;
+import org.apache.cxf.phase.AbortedInvocationException;
import org.apache.cxf.policy.PolicyDataEngine;
import org.apache.cxf.security.SecurityContext;
import org.apache.cxf.security.transport.TLSSessionInfo;
@@ -262,8 +263,17 @@ public abstract class AbstractHTTPDestination
copyKnownRequestAttributes(req, inMessage);
try {
- incomingObserver.onMessage(inMessage);
- invokeComplete(context, req, resp, inMessage);
+ try {
+ incomingObserver.onMessage(inMessage);
+ invokeComplete(context, req, resp, inMessage);
+ } catch (AbortedInvocationException ex) {
+ maybeResetAndCloseResponseOutputStream(resp);
+ if (ex.getRuntimeException() != null) {
+ throw ex.getRuntimeException();
+ } else {
+ throw ex;
+ }
+ }
} catch (SuspendedInvocationException ex) {
if (ex.getRuntimeException() != null) {
throw ex.getRuntimeException();
@@ -692,6 +702,21 @@ public abstract class AbstractHTTPDestination
return responseStream;
}
+ private void maybeResetAndCloseResponseOutputStream(HttpServletResponse
response) throws IOException {
+ try {
+ // The Servlet API does not provide means to abort the response,
the best
+ // we could do is reset buffers (only partial data is going to be
sent) and close
+ // the connection.
+ if (!response.isCommitted()) {
+ response.setHeader(HttpHeaderHelper.CONNECTION,
HttpHeaderHelper.CLOSE);
+ response.resetBuffer();
+ response.getOutputStream().close();
+ }
+ } catch (IllegalStateException ex) {
+ // response.getWriter() has already been called
+ }
+ }
+
private void closeResponseOutputStream(HttpServletResponse response)
throws IOException {
try {
response.getOutputStream().close();
diff --git
a/systests/jaxws/src/test/java/org/apache/cxf/systest/jaxws/AttachmentChunkingTest.java
b/systests/jaxws/src/test/java/org/apache/cxf/systest/jaxws/AbstractAttachmentChunkingTest.java
similarity index 59%
copy from
systests/jaxws/src/test/java/org/apache/cxf/systest/jaxws/AttachmentChunkingTest.java
copy to
systests/jaxws/src/test/java/org/apache/cxf/systest/jaxws/AbstractAttachmentChunkingTest.java
index fb1e00a607..2c01e081f0 100644
---
a/systests/jaxws/src/test/java/org/apache/cxf/systest/jaxws/AttachmentChunkingTest.java
+++
b/systests/jaxws/src/test/java/org/apache/cxf/systest/jaxws/AbstractAttachmentChunkingTest.java
@@ -25,45 +25,33 @@ import java.io.InputStream;
import java.io.OutputStream;
import java.util.Arrays;
import java.util.concurrent.ThreadLocalRandom;
-import java.util.logging.Logger;
import jakarta.activation.DataHandler;
import jakarta.activation.DataSource;
import jakarta.xml.ws.Binding;
import jakarta.xml.ws.BindingProvider;
-import jakarta.xml.ws.Endpoint;
import jakarta.xml.ws.soap.SOAPBinding;
-import jakarta.xml.ws.soap.SOAPFaultException;
import org.apache.cxf.Download;
import org.apache.cxf.DownloadFault_Exception;
import org.apache.cxf.DownloadNextResponseType;
-import org.apache.cxf.common.logging.LogUtils;
import org.apache.cxf.jaxws.JaxWsProxyFactoryBean;
import org.apache.cxf.testutil.common.AbstractBusClientServerTestBase;
-import org.apache.cxf.testutil.common.AbstractBusTestServerBase;
-import org.junit.BeforeClass;
import org.junit.Test;
-import static org.hamcrest.CoreMatchers.containsString;
import static org.hamcrest.CoreMatchers.equalTo;
import static org.hamcrest.MatcherAssert.assertThat;
-import static org.junit.Assert.assertThrows;
-import static org.junit.Assert.assertTrue;
-public class AttachmentChunkingTest extends AbstractBusClientServerTestBase {
- private static final String PORT = allocatePort(DownloadServer.class);
- private static final Logger LOG =
LogUtils.getLogger(AttachmentChunkingTest.class);
-
- private static final class DownloadImpl implements Download {
+abstract class AbstractAttachmentChunkingTest extends
AbstractBusClientServerTestBase {
+ protected static final class DownloadImpl implements Download {
@Override
- public DownloadNextResponseType downloadNext(Boolean simulate) {
+ public DownloadNextResponseType downloadNext(Integer
minNumberOfChunks, Boolean simulate) {
final DownloadNextResponseType responseType = new
DownloadNextResponseType();
responseType.setDataContent(new DataHandler(new DataSource() {
@Override
public InputStream getInputStream() {
if (simulate) {
- return simulate();
+ return simulate((minNumberOfChunks == null) ? 1 :
minNumberOfChunks);
} else {
return generate(100000);
}
@@ -88,49 +76,6 @@ public class AttachmentChunkingTest extends
AbstractBusClientServerTestBase {
return responseType;
}
}
-
- public static class DownloadServer extends AbstractBusTestServerBase {
- protected void run() {
- Object implementor = new DownloadImpl();
- String address = "http://localhost:"; + PORT +
"/SoapContext/SoapPort";
- Endpoint.publish(address, implementor);
- }
-
- public static void main(String[] args) {
- try {
- DownloadServer s = new DownloadServer();
- s.start();
- } catch (Exception ex) {
- ex.printStackTrace();
- System.exit(-1);
- } finally {
- LOG.info("done!");
- }
- }
- }
-
- @BeforeClass
- public static void startServers() throws Exception {
- assertTrue("server did not launch correctly",
launchServer(DownloadServer.class, true));
- }
-
- @Test
- public void testChunkingPartialFailure() {
- final JaxWsProxyFactoryBean factory = new JaxWsProxyFactoryBean();
- factory.setServiceClass(Download.class);
-
- final Download client = (Download) factory.create();
- final BindingProvider bindingProvider = (BindingProvider) client;
- final Binding binding = bindingProvider.getBinding();
-
- final String address =
String.format("http://localhost:%s/SoapContext/SoapPort/DownloadPort";, PORT);
-
bindingProvider.getRequestContext().put("jakarta.xml.ws.service.endpoint.address",
address);
- ((SOAPBinding) binding).setMTOMEnabled(true);
-
- // See please https://issues.apache.org/jira/browse/CXF-9057
- SOAPFaultException ex = assertThrows(SOAPFaultException.class, () ->
client.downloadNext(true));
- assertThat(ex.getMessage(), containsString("simulated error during
stream processing"));
- }
@Test
public void testChunking() throws IOException, DownloadFault_Exception {
@@ -141,22 +86,26 @@ public class AttachmentChunkingTest extends
AbstractBusClientServerTestBase {
final BindingProvider bindingProvider = (BindingProvider) client;
final Binding binding = bindingProvider.getBinding();
- final String address =
String.format("http://localhost:%s/SoapContext/SoapPort/DownloadPort";, PORT);
+ final String address =
String.format("http://localhost:%s/SoapContext/SoapPort/DownloadPort";,
getPort());
bindingProvider.getRequestContext().put("jakarta.xml.ws.service.endpoint.address",
address);
((SOAPBinding) binding).setMTOMEnabled(true);
- final DownloadNextResponseType response = client.downloadNext(false);
+ final DownloadNextResponseType response = client.downloadNext(1,
false);
assertThat(response.getDataContent().getInputStream().readAllBytes().length,
equalTo(100000));
}
+ protected abstract String getPort();
+
private static InputStream generate(int size) {
final byte[] buf = new byte[size];
Arrays.fill(buf, (byte) 'x');
return new ByteArrayInputStream(buf);
}
- private static InputStream simulate() {
+ private static InputStream simulate(final int minNumberOfChunks) {
return new InputStream() {
+ private int chunk;
+
@Override
public int read() {
return (byte) 'x';
@@ -164,7 +113,7 @@ public class AttachmentChunkingTest extends
AbstractBusClientServerTestBase {
@Override
public int read(byte[] b, int off, int len) {
- if (ThreadLocalRandom.current().nextBoolean()) {
+ if (chunk++ >= minNumberOfChunks &&
ThreadLocalRandom.current().nextBoolean()) {
throw new IllegalArgumentException("simulated error during
stream processing");
}
diff --git
a/systests/jaxws/src/test/java/org/apache/cxf/systest/jaxws/AttachmentChunkingTest.java
b/systests/jaxws/src/test/java/org/apache/cxf/systest/jaxws/AttachmentChunkingTest.java
index fb1e00a607..dcb3ea4c7c 100644
---
a/systests/jaxws/src/test/java/org/apache/cxf/systest/jaxws/AttachmentChunkingTest.java
+++
b/systests/jaxws/src/test/java/org/apache/cxf/systest/jaxws/AttachmentChunkingTest.java
@@ -19,76 +19,32 @@
package org.apache.cxf.systest.jaxws;
-import java.io.ByteArrayInputStream;
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
-import java.util.Arrays;
-import java.util.concurrent.ThreadLocalRandom;
+import java.util.List;
import java.util.logging.Logger;
-import jakarta.activation.DataHandler;
-import jakarta.activation.DataSource;
import jakarta.xml.ws.Binding;
import jakarta.xml.ws.BindingProvider;
import jakarta.xml.ws.Endpoint;
import jakarta.xml.ws.soap.SOAPBinding;
import jakarta.xml.ws.soap.SOAPFaultException;
import org.apache.cxf.Download;
-import org.apache.cxf.DownloadFault_Exception;
-import org.apache.cxf.DownloadNextResponseType;
import org.apache.cxf.common.logging.LogUtils;
+import org.apache.cxf.ext.logging.LoggingFeature;
import org.apache.cxf.jaxws.JaxWsProxyFactoryBean;
-import org.apache.cxf.testutil.common.AbstractBusClientServerTestBase;
import org.apache.cxf.testutil.common.AbstractBusTestServerBase;
import org.junit.BeforeClass;
import org.junit.Test;
import static org.hamcrest.CoreMatchers.containsString;
-import static org.hamcrest.CoreMatchers.equalTo;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.Assert.assertThrows;
import static org.junit.Assert.assertTrue;
-public class AttachmentChunkingTest extends AbstractBusClientServerTestBase {
+public class AttachmentChunkingTest extends AbstractAttachmentChunkingTest {
private static final String PORT = allocatePort(DownloadServer.class);
private static final Logger LOG =
LogUtils.getLogger(AttachmentChunkingTest.class);
- private static final class DownloadImpl implements Download {
- @Override
- public DownloadNextResponseType downloadNext(Boolean simulate) {
- final DownloadNextResponseType responseType = new
DownloadNextResponseType();
- responseType.setDataContent(new DataHandler(new DataSource() {
- @Override
- public InputStream getInputStream() {
- if (simulate) {
- return simulate();
- } else {
- return generate(100000);
- }
- }
-
- @Override
- public OutputStream getOutputStream() {
- return null;
- }
-
- @Override
- public String getContentType() {
- return "";
- }
-
- @Override
- public String getName() {
- return "";
- }
- }));
-
- return responseType;
- }
- }
-
public static class DownloadServer extends AbstractBusTestServerBase {
protected void run() {
Object implementor = new DownloadImpl();
@@ -117,63 +73,24 @@ public class AttachmentChunkingTest extends
AbstractBusClientServerTestBase {
@Test
public void testChunkingPartialFailure() {
final JaxWsProxyFactoryBean factory = new JaxWsProxyFactoryBean();
+ factory.setFeatures(List.of(new LoggingFeature()));
factory.setServiceClass(Download.class);
final Download client = (Download) factory.create();
final BindingProvider bindingProvider = (BindingProvider) client;
final Binding binding = bindingProvider.getBinding();
- final String address =
String.format("http://localhost:%s/SoapContext/SoapPort/DownloadPort";, PORT);
+ final String address =
String.format("http://localhost:%s/SoapContext/SoapPort/DownloadPort";,
getPort());
bindingProvider.getRequestContext().put("jakarta.xml.ws.service.endpoint.address",
address);
((SOAPBinding) binding).setMTOMEnabled(true);
// See please https://issues.apache.org/jira/browse/CXF-9057
- SOAPFaultException ex = assertThrows(SOAPFaultException.class, () ->
client.downloadNext(true));
+ SOAPFaultException ex = assertThrows(SOAPFaultException.class, () ->
client.downloadNext(1, true));
assertThat(ex.getMessage(), containsString("simulated error during
stream processing"));
}
-
- @Test
- public void testChunking() throws IOException, DownloadFault_Exception {
- final JaxWsProxyFactoryBean factory = new JaxWsProxyFactoryBean();
- factory.setServiceClass(Download.class);
-
- final Download client = (Download) factory.create();
- final BindingProvider bindingProvider = (BindingProvider) client;
- final Binding binding = bindingProvider.getBinding();
-
- final String address =
String.format("http://localhost:%s/SoapContext/SoapPort/DownloadPort";, PORT);
-
bindingProvider.getRequestContext().put("jakarta.xml.ws.service.endpoint.address",
address);
- ((SOAPBinding) binding).setMTOMEnabled(true);
-
- final DownloadNextResponseType response = client.downloadNext(false);
-
assertThat(response.getDataContent().getInputStream().readAllBytes().length,
equalTo(100000));
- }
-
- private static InputStream generate(int size) {
- final byte[] buf = new byte[size];
- Arrays.fill(buf, (byte) 'x');
- return new ByteArrayInputStream(buf);
- }
-
- private static InputStream simulate() {
- return new InputStream() {
- @Override
- public int read() {
- return (byte) 'x';
- }
- @Override
- public int read(byte[] b, int off, int len) {
- if (ThreadLocalRandom.current().nextBoolean()) {
- throw new IllegalArgumentException("simulated error during
stream processing");
- }
-
- for (int i = off; i < off + len; i++) {
- b[i] = (byte) 'x';
- }
-
- return len;
- }
- };
+ @Override
+ protected String getPort() {
+ return PORT;
}
}
diff --git
a/systests/jaxws/src/test/java/org/apache/cxf/systest/jaxws/AttachmentChunkingTest.java
b/systests/jaxws/src/test/java/org/apache/cxf/systest/jaxws/AttachmentMtomChunkingTest.java
similarity index 54%
copy from
systests/jaxws/src/test/java/org/apache/cxf/systest/jaxws/AttachmentChunkingTest.java
copy to
systests/jaxws/src/test/java/org/apache/cxf/systest/jaxws/AttachmentMtomChunkingTest.java
index fb1e00a607..c4da85e900 100644
---
a/systests/jaxws/src/test/java/org/apache/cxf/systest/jaxws/AttachmentChunkingTest.java
+++
b/systests/jaxws/src/test/java/org/apache/cxf/systest/jaxws/AttachmentMtomChunkingTest.java
@@ -19,81 +19,40 @@
package org.apache.cxf.systest.jaxws;
-import java.io.ByteArrayInputStream;
import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
-import java.util.Arrays;
-import java.util.concurrent.ThreadLocalRandom;
+import java.util.List;
import java.util.logging.Logger;
-import jakarta.activation.DataHandler;
-import jakarta.activation.DataSource;
import jakarta.xml.ws.Binding;
import jakarta.xml.ws.BindingProvider;
import jakarta.xml.ws.Endpoint;
+import jakarta.xml.ws.WebServiceException;
import jakarta.xml.ws.soap.SOAPBinding;
-import jakarta.xml.ws.soap.SOAPFaultException;
import org.apache.cxf.Download;
import org.apache.cxf.DownloadFault_Exception;
import org.apache.cxf.DownloadNextResponseType;
import org.apache.cxf.common.logging.LogUtils;
+import org.apache.cxf.ext.logging.LoggingFeature;
+import org.apache.cxf.helpers.IOUtils;
import org.apache.cxf.jaxws.JaxWsProxyFactoryBean;
-import org.apache.cxf.testutil.common.AbstractBusClientServerTestBase;
import org.apache.cxf.testutil.common.AbstractBusTestServerBase;
import org.junit.BeforeClass;
import org.junit.Test;
-import static org.hamcrest.CoreMatchers.containsString;
-import static org.hamcrest.CoreMatchers.equalTo;
-import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.Assert.assertThrows;
import static org.junit.Assert.assertTrue;
-public class AttachmentChunkingTest extends AbstractBusClientServerTestBase {
+public class AttachmentMtomChunkingTest extends AbstractAttachmentChunkingTest
{
private static final String PORT = allocatePort(DownloadServer.class);
- private static final Logger LOG =
LogUtils.getLogger(AttachmentChunkingTest.class);
-
- private static final class DownloadImpl implements Download {
- @Override
- public DownloadNextResponseType downloadNext(Boolean simulate) {
- final DownloadNextResponseType responseType = new
DownloadNextResponseType();
- responseType.setDataContent(new DataHandler(new DataSource() {
- @Override
- public InputStream getInputStream() {
- if (simulate) {
- return simulate();
- } else {
- return generate(100000);
- }
- }
-
- @Override
- public OutputStream getOutputStream() {
- return null;
- }
-
- @Override
- public String getContentType() {
- return "";
- }
-
- @Override
- public String getName() {
- return "";
- }
- }));
-
- return responseType;
- }
- }
+ private static final Logger LOG =
LogUtils.getLogger(AttachmentMtomChunkingTest.class);
public static class DownloadServer extends AbstractBusTestServerBase {
protected void run() {
Object implementor = new DownloadImpl();
String address = "http://localhost:"; + PORT +
"/SoapContext/SoapPort";
- Endpoint.publish(address, implementor);
+ final Endpoint endpoint = Endpoint.publish(address, implementor,
new LoggingFeature());
+ ((SOAPBinding)endpoint.getBinding()).setMTOMEnabled(true);
}
public static void main(String[] args) {
@@ -113,67 +72,46 @@ public class AttachmentChunkingTest extends
AbstractBusClientServerTestBase {
public static void startServers() throws Exception {
assertTrue("server did not launch correctly",
launchServer(DownloadServer.class, true));
}
-
+
@Test
- public void testChunkingPartialFailure() {
+ public void testChunkingPartialEarlyFailure() throws IOException,
DownloadFault_Exception {
final JaxWsProxyFactoryBean factory = new JaxWsProxyFactoryBean();
+ factory.setFeatures(List.of(new LoggingFeature()));
factory.setServiceClass(Download.class);
final Download client = (Download) factory.create();
final BindingProvider bindingProvider = (BindingProvider) client;
final Binding binding = bindingProvider.getBinding();
- final String address =
String.format("http://localhost:%s/SoapContext/SoapPort/DownloadPort";, PORT);
+ final String address =
String.format("http://localhost:%s/SoapContext/SoapPort/DownloadPort";,
getPort());
bindingProvider.getRequestContext().put("jakarta.xml.ws.service.endpoint.address",
address);
((SOAPBinding) binding).setMTOMEnabled(true);
// See please https://issues.apache.org/jira/browse/CXF-9057
- SOAPFaultException ex = assertThrows(SOAPFaultException.class, () ->
client.downloadNext(true));
- assertThat(ex.getMessage(), containsString("simulated error during
stream processing"));
+ assertThrows(WebServiceException.class, () -> client.downloadNext(1,
true));
}
-
+
@Test
- public void testChunking() throws IOException, DownloadFault_Exception {
+ public void testChunkingPartialLateFailure() throws IOException,
DownloadFault_Exception {
final JaxWsProxyFactoryBean factory = new JaxWsProxyFactoryBean();
+ factory.setFeatures(List.of(new LoggingFeature()));
factory.setServiceClass(Download.class);
final Download client = (Download) factory.create();
final BindingProvider bindingProvider = (BindingProvider) client;
final Binding binding = bindingProvider.getBinding();
- final String address =
String.format("http://localhost:%s/SoapContext/SoapPort/DownloadPort";, PORT);
+ final String address =
String.format("http://localhost:%s/SoapContext/SoapPort/DownloadPort";,
getPort());
bindingProvider.getRequestContext().put("jakarta.xml.ws.service.endpoint.address",
address);
((SOAPBinding) binding).setMTOMEnabled(true);
- final DownloadNextResponseType response = client.downloadNext(false);
-
assertThat(response.getDataContent().getInputStream().readAllBytes().length,
equalTo(100000));
- }
-
- private static InputStream generate(int size) {
- final byte[] buf = new byte[size];
- Arrays.fill(buf, (byte) 'x');
- return new ByteArrayInputStream(buf);
+ // See please https://issues.apache.org/jira/browse/CXF-9057
+ final DownloadNextResponseType response = client.downloadNext(10,
true);
+ assertThrows(IOException.class, () ->
IOUtils.readBytesFromStream(response.getDataContent().getInputStream()));
}
-
- private static InputStream simulate() {
- return new InputStream() {
- @Override
- public int read() {
- return (byte) 'x';
- }
- @Override
- public int read(byte[] b, int off, int len) {
- if (ThreadLocalRandom.current().nextBoolean()) {
- throw new IllegalArgumentException("simulated error during
stream processing");
- }
-
- for (int i = off; i < off + len; i++) {
- b[i] = (byte) 'x';
- }
-
- return len;
- }
- };
+ @Override
+ protected String getPort() {
+ return PORT;
}
}
diff --git a/systests/jaxws/src/test/resources/attachments/cxf9057.wsdl
b/systests/jaxws/src/test/resources/attachments/cxf9057.wsdl
index 721fa08bb4..9a83f31a96 100644
--- a/systests/jaxws/src/test/resources/attachments/cxf9057.wsdl
+++ b/systests/jaxws/src/test/resources/attachments/cxf9057.wsdl
@@ -21,6 +21,7 @@
</xs:complexType>
<xs:complexType name="downloadNext">
<xs:sequence>
+ <xs:element minOccurs="0" name="minNumberOfChunks"
type="xs:int"/>
<xs:element minOccurs="0" name="simulate"
type="xs:boolean"/>
</xs:sequence>
</xs:complexType>
