CXF-9057: Chunked Stream is closed regularly when Exception is thrown (MTOM)

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;
+    }
+}

core/src/main/java/org/apache/cxf/phase/PhaseInterceptorChain.java

@@ -325,6 +325,9 @@ public synchronized boolean doIntercept(Message message) {
                     }
                     pause();
                     throw ex;
+                } catch (AbortedInvocationException ex) {
+                    abort();
+                    throw ex;
                 } catch (RuntimeException ex) {
                     if (!faultOccurred) {
                         faultOccurred = true;

rt/bindings/soap/src/main/java/org/apache/cxf/binding/soap/interceptor/Soap11FaultOutInterceptor.java

@@ -37,6 +37,7 @@
 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,7 @@ public void handleMessage(SoapMessage message) throws Fault {
             // 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;
+                throw new AbortedInvocationException(f);
             }
 
             XMLStreamWriter writer = message.getContent(XMLStreamWriter.class);

rt/bindings/soap/src/main/java/org/apache/cxf/binding/soap/interceptor/Soap12FaultOutInterceptor.java

@@ -38,6 +38,7 @@
 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,7 @@ public void handleMessage(SoapMessage message) throws Fault {
             // 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;
+                throw new AbortedInvocationException(f);
             }
 
             XMLStreamWriter writer = message.getContent(XMLStreamWriter.class);

rt/transports/http/src/main/java/org/apache/cxf/transport/http/AbstractHTTPDestination.java

@@ -63,6 +63,7 @@
 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 void invoke(final ServletConfig config,
         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,17 @@ protected OutputStream flushHeaders(Message outMessage, boolean getStream) throw
         return responseStream;
     }
 
+    private void maybeResetAndCloseResponseOutputStream(HttpServletResponse response) throws IOException {
+        try {
+            if (!response.isCommitted()) {
+                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();

systests/jaxws/src/test/java/org/apache/cxf/systest/jaxws/AbstractAttachmentChunkingTest.java

@@ -0,0 +1,126 @@
+/**
+ * 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.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 jakarta.activation.DataHandler;
+import jakarta.activation.DataSource;
+import jakarta.xml.ws.Binding;
+import jakarta.xml.ws.BindingProvider;
+import jakarta.xml.ws.soap.SOAPBinding;
+import org.apache.cxf.Download;
+import org.apache.cxf.DownloadFault_Exception;
+import org.apache.cxf.DownloadNextResponseType;
+import org.apache.cxf.jaxws.JaxWsProxyFactoryBean;
+import org.apache.cxf.testutil.common.AbstractBusClientServerTestBase;
+
+import org.junit.Test;
+
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.MatcherAssert.assertThat;
+
+abstract class AbstractAttachmentChunkingTest extends AbstractBusClientServerTestBase {
+    protected 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;
+        }
+    }
+
+    @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", 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));
+    }
+
+    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() {
+        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;
+            }
+        };
+    }
+}