Make sure netty client isn't used as the default for sync calls unless specifically asked for (like the hc client). Also, if SSLSocketFactory specified, make sure we delegate down to URL version so it's used.
Project: http://git-wip-us.apache.org/repos/asf/cxf/repo Commit: http://git-wip-us.apache.org/repos/asf/cxf/commit/44b35f11 Tree: http://git-wip-us.apache.org/repos/asf/cxf/tree/44b35f11 Diff: http://git-wip-us.apache.org/repos/asf/cxf/diff/44b35f11 Branch: refs/heads/master Commit: 44b35f1193552619bf6ff843254a5db0b8104263 Parents: 63e985d Author: Daniel Kulp <[email protected]> Authored: Thu Jun 12 12:39:04 2014 -0400 Committer: Daniel Kulp <[email protected]> Committed: Thu Jun 12 13:18:08 2014 -0400 ---------------------------------------------------------------------- .../http/netty/client/NettyHttpConduit.java | 72 ++++++++++++++++---- .../netty/client/NettyHttpConduitFactory.java | 22 ++++++ 2 files changed, 80 insertions(+), 14 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/cxf/blob/44b35f11/rt/transports/http-netty/netty-client/src/main/java/org/apache/cxf/transport/http/netty/client/NettyHttpConduit.java ---------------------------------------------------------------------- diff --git a/rt/transports/http-netty/netty-client/src/main/java/org/apache/cxf/transport/http/netty/client/NettyHttpConduit.java b/rt/transports/http-netty/netty-client/src/main/java/org/apache/cxf/transport/http/netty/client/NettyHttpConduit.java index 311bf6b..583087f 100644 --- a/rt/transports/http-netty/netty-client/src/main/java/org/apache/cxf/transport/http/netty/client/NettyHttpConduit.java +++ b/rt/transports/http-netty/netty-client/src/main/java/org/apache/cxf/transport/http/netty/client/NettyHttpConduit.java @@ -92,6 +92,9 @@ public class NettyHttpConduit extends URLConnectionHTTPConduit implements BusLif // Using Netty API directly protected void setupConnection(Message message, URI uri, HTTPClientPolicy csPolicy) throws IOException { + + + // need to do some clean up work on the URI address String uriString = uri.toString(); if (uriString.startsWith("netty://")) { @@ -105,6 +108,45 @@ public class NettyHttpConduit extends URLConnectionHTTPConduit implements BusLif if (!"http".equals(s) && !"https".equals(s)) { throw new MalformedURLException("unknown protocol: " + s); } + + Object o = message.getContextualProperty(USE_ASYNC); + if (o == null) { + o = factory.getUseAsyncPolicy(); + } + if (o instanceof NettyHttpConduitFactory.UseAsyncPolicy) { + switch ((NettyHttpConduitFactory.UseAsyncPolicy)o) { + case ALWAYS: + o = true; + break; + case NEVER: + o = false; + break; + case ASYNC_ONLY: + default: + o = !message.getExchange().isSynchronous(); + break; + } + + } + // check tlsClientParameters from message header + TLSClientParameters clientParameters = message.get(TLSClientParameters.class); + if (clientParameters == null) { + clientParameters = tlsClientParameters; + } + if (uri.getScheme().equals("https") + && clientParameters != null + && clientParameters.getSSLSocketFactory() != null) { + //if they configured in an SSLSocketFactory, we cannot do anything + //with it as the NIO based transport cannot use socket created from + //the SSLSocketFactory. + o = false; + } + if (!MessageUtils.isTrue(o)) { + message.put(USE_ASYNC, Boolean.FALSE); + super.setupConnection(message, uri, csPolicy); + return; + } + message.put(USE_ASYNC, Boolean.TRUE); if (StringUtils.isEmpty(uri.getPath())) { //hc needs to have the path be "/" @@ -134,20 +176,22 @@ public class NettyHttpConduit extends URLConnectionHTTPConduit implements BusLif boolean isChunking, int chunkThreshold) throws IOException { - NettyHttpClientRequest entity = message.get(NettyHttpClientRequest.class); - NettyWrappedOutputStream out = new NettyWrappedOutputStream(message, - needToCacheRequest, - isChunking, - chunkThreshold, - getConduitName(), - entity.getUri()); - entity.createRequest(out.getOutBuffer()); - // TODO need to check how to set the Chunked feature - //request.getRequest().setChunked(true); - entity.getRequest().headers().set(Message.CONTENT_TYPE, (String)message.get(Message.CONTENT_TYPE)); - return out; - - + if (Boolean.TRUE.equals(message.get(USE_ASYNC))) { + + NettyHttpClientRequest entity = message.get(NettyHttpClientRequest.class); + NettyWrappedOutputStream out = new NettyWrappedOutputStream(message, + needToCacheRequest, + isChunking, + chunkThreshold, + getConduitName(), + entity.getUri()); + entity.createRequest(out.getOutBuffer()); + // TODO need to check how to set the Chunked feature + //request.getRequest().setChunked(true); + entity.getRequest().headers().set(Message.CONTENT_TYPE, (String)message.get(Message.CONTENT_TYPE)); + return out; + } + return super.createOutputStream(message, needToCacheRequest, isChunking, chunkThreshold); } public class NettyWrappedOutputStream extends WrappedOutputStream { http://git-wip-us.apache.org/repos/asf/cxf/blob/44b35f11/rt/transports/http-netty/netty-client/src/main/java/org/apache/cxf/transport/http/netty/client/NettyHttpConduitFactory.java ---------------------------------------------------------------------- diff --git a/rt/transports/http-netty/netty-client/src/main/java/org/apache/cxf/transport/http/netty/client/NettyHttpConduitFactory.java b/rt/transports/http-netty/netty-client/src/main/java/org/apache/cxf/transport/http/netty/client/NettyHttpConduitFactory.java index 1b7ac41..6a55ae0 100644 --- a/rt/transports/http-netty/netty-client/src/main/java/org/apache/cxf/transport/http/netty/client/NettyHttpConduitFactory.java +++ b/rt/transports/http-netty/netty-client/src/main/java/org/apache/cxf/transport/http/netty/client/NettyHttpConduitFactory.java @@ -24,6 +24,7 @@ import java.io.IOException; import org.apache.cxf.Bus; import org.apache.cxf.buslifecycle.BusLifeCycleListener; import org.apache.cxf.buslifecycle.BusLifeCycleManager; +import org.apache.cxf.common.util.SystemPropertyAction; import org.apache.cxf.service.model.EndpointInfo; import org.apache.cxf.transport.http.HTTPConduit; import org.apache.cxf.transport.http.HTTPConduitFactory; @@ -36,7 +37,28 @@ import io.netty.channel.nio.NioEventLoopGroup; public class NettyHttpConduitFactory implements HTTPConduitFactory { + //CXF specific + public static final String USE_POLICY = "org.apache.cxf.transport.http.netty.usePolicy"; + + public static enum UseAsyncPolicy { + ALWAYS, ASYNC_ONLY, NEVER + }; + + UseAsyncPolicy policy; public NettyHttpConduitFactory() { + Object st = SystemPropertyAction.getPropertyOrNull(USE_POLICY); + if (st instanceof UseAsyncPolicy) { + policy = (UseAsyncPolicy)st; + } else if (st instanceof String) { + policy = UseAsyncPolicy.valueOf((String)st); + } else { + //policy = UseAsyncPolicy.ALWAYS; + policy = UseAsyncPolicy.ASYNC_ONLY; + } + } + + public UseAsyncPolicy getUseAsyncPolicy() { + return policy; } @Override
