Skip to content

Commit

Permalink
CXF-9009: Async operations fails (netty-conduit) (#1868)
Browse files Browse the repository at this point in the history
(cherry picked from commit 87f68a9)
  • Loading branch information
reta committed Jun 4, 2024
1 parent ea88d55 commit 8c3354d
Show file tree
Hide file tree
Showing 4 changed files with 211 additions and 18 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -69,8 +69,11 @@
import io.netty.handler.ssl.util.SimpleTrustManagerFactory;
import io.netty.handler.stream.ChunkedWriteHandler;
import io.netty.handler.timeout.ReadTimeoutHandler;
import io.netty.util.AttributeKey;

public class NettyHttpClientPipelineFactory extends ChannelInitializer<Channel> {
private static final String WHEN_READY_KEY = "WhenReady-Key";
private static final AttributeKey<ChannelFuture> WHEN_READY = AttributeKey.valueOf(WHEN_READY_KEY);

private static final Logger LOG =
LogUtils.getL7dLogger(NettyHttpClientPipelineFactory.class);
Expand All @@ -79,7 +82,6 @@ public class NettyHttpClientPipelineFactory extends ChannelInitializer<Channel>
private final int readTimeout;
private final int maxContentLength;
private final boolean enableHttp2;
private ChannelPromise readyFuture;

public NettyHttpClientPipelineFactory(TLSClientParameters clientParameters) {
this(clientParameters, 0);
Expand Down Expand Up @@ -115,8 +117,9 @@ protected void initChannel(Channel ch) throws Exception {
}

final NettyHttpClientHandler responseHandler = new NettyHttpClientHandler();
readyFuture = ch.newPromise();

final ChannelPromise readyFuture = ch.newPromise();
ch.attr(WHEN_READY).set(readyFuture);

if (enableHttp2) {
final Http2Connection connection = new DefaultHttp2Connection(false);
final Http2SettingsHandler settingsHandler = new Http2SettingsHandler(readyFuture);
Expand Down Expand Up @@ -188,8 +191,8 @@ protected void configurePipeline(ChannelHandlerContext ctx, String protocol) {
}
}

public ChannelFuture whenReady() {
return readyFuture;
ChannelFuture whenReady(Channel channel) {
return channel.attr(NettyHttpClientPipelineFactory.WHEN_READY).get();
}

private SslHandler configureClientSSLOnDemand(Channel channel) throws Exception {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -458,15 +458,17 @@ public void operationComplete(ChannelFuture future) throws Exception {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
if (future.isSuccess()) {
handler.whenReady().addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
if (future.isSuccess()) {
ChannelFuture channelFuture = future.channel().writeAndFlush(entity);
channelFuture.addListener(writeFailureListener);
handler.whenReady(future.channel())
.addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
if (future.isSuccess()) {
ChannelFuture channelFuture = future.channel().writeAndFlush(entity);
channelFuture.addListener(writeFailureListener);
}
}
}
});
);
}
}
});
Expand Down
18 changes: 12 additions & 6 deletions systests/transport-netty/pom.xml
Original file line number Diff line number Diff line change
Expand Up @@ -95,27 +95,32 @@
<dependency>
<groupId>org.apache.cxf</groupId>
<artifactId>cxf-core</artifactId>
<version>${project.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.cxf</groupId>
<artifactId>cxf-rt-databinding-jaxb</artifactId>
<version>${project.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.cxf</groupId>
<artifactId>cxf-rt-transports-http</artifactId>
<version>${project.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.cxf</groupId>
<artifactId>cxf-rt-frontend-jaxrs</artifactId>
<version>${project.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.cxf</groupId>
<artifactId>cxf-rt-rs-client</artifactId>
<version>${project.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.cxf</groupId>
<artifactId>cxf-rt-frontend-jaxws</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.jaxrs</groupId>
Expand All @@ -140,7 +145,7 @@
<dependency>
<groupId>org.apache.cxf</groupId>
<artifactId>cxf-rt-transports-http-netty-client</artifactId>
<version>${project.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
Expand Down Expand Up @@ -192,6 +197,7 @@
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-core</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>junit</groupId>
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,182 @@
/**
* 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.http2.netty.jaxws;

import java.io.IOException;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import javax.jws.WebService;
import javax.xml.ws.Response;
import javax.xml.ws.soap.SOAPFaultException;

import org.apache.cxf.endpoint.Client;
import org.apache.cxf.greeter_control.AbstractGreeterImpl;
import org.apache.cxf.greeter_control.Greeter;
import org.apache.cxf.greeter_control.types.GreetMeResponse;
import org.apache.cxf.jaxws.JaxWsProxyFactoryBean;
import org.apache.cxf.testutil.common.AbstractBusClientServerTestBase;
import org.apache.cxf.testutil.common.AbstractBusTestServerBase;
import org.apache.cxf.transport.http.HTTPConduit;

import io.netty.handler.timeout.ReadTimeoutException;

import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;

import static org.hamcrest.CoreMatchers.is;
import static org.hamcrest.CoreMatchers.not;
import static org.hamcrest.CoreMatchers.nullValue;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;

public class JAXWSAsyncClientTest extends AbstractBusClientServerTestBase {
static final String PORT = allocatePort(Server.class);

public static class Server extends AbstractBusTestServerBase {

protected void run() {
GreeterImpl implementor = new GreeterImpl();
String address = "http://localhost:" + PORT + "/SoapContext/GreeterPort";
javax.xml.ws.Endpoint.publish(address, implementor);
}

public static void main(String[] args) {
try {
Server s = new Server();
s.start();
} catch (Exception ex) {
ex.printStackTrace();
System.exit(-1);
} finally {
System.out.println("done!");
}
}

@WebService(serviceName = "BasicGreeterService",
portName = "GreeterPort",
endpointInterface = "org.apache.cxf.greeter_control.Greeter",
targetNamespace = "http://cxf.apache.org/greeter_control",
wsdlLocation = "testutils/greeter_control.wsdl")
public class GreeterImpl extends AbstractGreeterImpl {
@Override
public String greetMe(String arg) {
if ("timeout".equalsIgnoreCase(arg)) {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
// Do nothing
}
}

return super.greetMe(arg);
}
}
}


@BeforeClass
public static void startServers() throws Exception {
assertTrue("server did not launch correctly", launchServer(Server.class, true));
}

@AfterClass
public static void stopServers() throws Exception {
stopAllServers();
}

@Test
public void testAsyncClient() throws Exception {
// setup the feature by using JAXWS front-end API
JaxWsProxyFactoryBean factory = new JaxWsProxyFactoryBean();
factory.setAddress("http://localhost:" + PORT + "/SoapContext/GreeterPort");
factory.setServiceClass(Greeter.class);
Greeter proxy = factory.create(Greeter.class);

Response<GreetMeResponse> response = proxy.greetMeAsync("cxf");
int waitCount = 0;
while (!response.isDone() && waitCount < 15) {
Thread.sleep(1000);
waitCount++;
}

assertTrue("Response still not received.", response.isDone());
}

@Test
public void testAsyncClientConcurrently() throws Exception {
final ExecutorService executor = Executors.newFixedThreadPool(3);

// setup the feature by using JAXWS front-end API
JaxWsProxyFactoryBean factory = new JaxWsProxyFactoryBean();
factory.setAddress("http://localhost:" + PORT + "/SoapContext/GreeterPort");
factory.setServiceClass(Greeter.class);
Greeter proxy = factory.create(Greeter.class);

final Callable<Object> callable = () -> proxy.greetMeAsync("cxf", resp -> { }).get(5, TimeUnit.SECONDS);
final List<Future<Object>> futures = executor.invokeAll(List.of(callable, callable, callable));

for (final Future<?> response: futures) {
int waitCount = 0;
while (!response.isDone() && waitCount < 15) {
Thread.sleep(1000);
waitCount++;
}
}

assertTrue("Response still not received.", futures.stream().allMatch(Future::isDone));
for (final Future<?> response: futures) {
assertThat(response.get(), is(not(nullValue())));
}
executor.shutdown();

assertTrue(executor.awaitTermination(10, TimeUnit.SECONDS));
}

@Test
public void testTimeout() throws Exception {
// setup the feature by using JAXWS front-end API
JaxWsProxyFactoryBean factory = new JaxWsProxyFactoryBean();
factory.setAddress("http://localhost:" + PORT + "/SoapContext/GreeterPort");
factory.setServiceClass(Greeter.class);
Greeter proxy = factory.create(Greeter.class);

HTTPConduit cond = (HTTPConduit)((Client)proxy).getConduit();
cond.getClient().setReceiveTimeout(500);

try {
proxy.greetMeAsync("timeout").get();
fail("Should have faulted");
} catch (SOAPFaultException ex) {
fail("should not be a SOAPFaultException");
} catch (ExecutionException ex) {
//expected
assertTrue(ex.getCause().getClass().getName(),
ex.getCause() instanceof IOException
|| ex.getCause() instanceof ReadTimeoutException);
}
}
}

0 comments on commit 8c3354d

Please sign in to comment.