Skip to content

Commit

Permalink
CXF-9057: Chunked Stream is closed regularly when Exception is thrown…
Browse files Browse the repository at this point in the history
… (MTOM)
  • Loading branch information
reta committed Jan 31, 2025
1 parent 283c861 commit 54e1a67
Show file tree
Hide file tree
Showing 8 changed files with 322 additions and 95 deletions.
Original file line number Diff line number Diff line change
@@ -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;
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand Down Expand Up @@ -69,7 +70,12 @@ 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;
// 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);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand Down Expand Up @@ -71,7 +72,12 @@ 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;
// 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);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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();
Expand Down Expand Up @@ -692,6 +702,21 @@ protected OutputStream flushHeaders(Message outMessage, boolean getStream) throw
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();
Expand Down
Original file line number Diff line number Diff line change
@@ -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;
}
};
}
}
Loading

0 comments on commit 54e1a67

Please sign in to comment.