Author: dkulp
Date: Mon Mar 26 02:53:14 2012
New Revision: 1305204
URL: http://svn.apache.org/viewvc?rev=1305204&view=rev
Log:
[CXF-4154] If the URL protocol changes, grab a new conduit.
Modified:
cxf/trunk/api/src/main/java/org/apache/cxf/endpoint/AbstractConduitSelector.java
cxf/trunk/api/src/main/java/org/apache/cxf/endpoint/DeferredConduitSelector.java
cxf/trunk/api/src/main/java/org/apache/cxf/endpoint/UpfrontConduitSelector.java
cxf/trunk/rt/features/clustering/src/main/java/org/apache/cxf/clustering/FailoverFeature.java
cxf/trunk/rt/features/clustering/src/main/java/org/apache/cxf/clustering/FailoverTargetSelector.java
cxf/trunk/rt/features/clustering/src/main/java/org/apache/cxf/clustering/LoadDistributorFeature.java
cxf/trunk/rt/features/clustering/src/main/java/org/apache/cxf/clustering/LoadDistributorTargetSelector.java
cxf/trunk/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/client/AbstractClient.java
cxf/trunk/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/failover/FailoverTest.java
Modified:
cxf/trunk/api/src/main/java/org/apache/cxf/endpoint/AbstractConduitSelector.java
URL:
http://svn.apache.org/viewvc/cxf/trunk/api/src/main/java/org/apache/cxf/endpoint/AbstractConduitSelector.java?rev=1305204&r1=1305203&r2=1305204&view=diff
==============================================================================
---
cxf/trunk/api/src/main/java/org/apache/cxf/endpoint/AbstractConduitSelector.java
(original)
+++
cxf/trunk/api/src/main/java/org/apache/cxf/endpoint/AbstractConduitSelector.java
Mon Mar 26 02:53:14 2012
@@ -21,6 +21,8 @@ package org.apache.cxf.endpoint;
import java.io.Closeable;
import java.io.IOException;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
import java.util.logging.Logger;
import org.apache.cxf.BusException;
@@ -40,28 +42,40 @@ import org.apache.cxf.ws.addressing.Endp
/**
* Abstract base class holding logic common to any ConduitSelector
- * that retreives a Conduit from the ConduitInitiator.
+ * that retrieves a Conduit from the ConduitInitiator.
*/
public abstract class AbstractConduitSelector implements ConduitSelector,
Closeable {
+ public static final String CONDUIT_COMPARE_FULL_URL
+ = "org.apache.cxf.ConduitSelector.compareFullUrl";
protected static final String KEEP_CONDUIT_ALIVE = "KeepConduitAlive";
- protected Conduit selectedConduit;
+
+ //collection of conduits that were created so we can close them all at the
end
+ protected List<Conduit> conduits = new CopyOnWriteArrayList<Conduit>();
+
+ //protected Conduit selectedConduit;
protected Endpoint endpoint;
+
+ public AbstractConduitSelector() {
+ }
+
/**
* Constructor, allowing a specific conduit to override normal selection.
*
* @param c specific conduit
*/
public AbstractConduitSelector(Conduit c) {
- selectedConduit = c;
+ if (c != null) {
+ conduits.add(c);
+ }
}
public void close() {
- if (selectedConduit != null) {
- selectedConduit.close();
- selectedConduit = null;
+ for (Conduit c : conduits) {
+ c.close();
}
+ conduits.clear();
}
/**
* Mechanics to actually get the Conduit from the ConduitInitiator
@@ -69,8 +83,9 @@ public abstract class AbstractConduitSel
*
* @param message the current Message
*/
- protected synchronized Conduit getSelectedConduit(Message message) {
- if (selectedConduit == null) {
+ protected Conduit getSelectedConduit(Message message) {
+ Conduit c = findCompatibleConduit(message);
+ if (c == null) {
Exchange exchange = message.getExchange();
EndpointInfo ei = endpoint.getEndpointInfo();
String transportID = ei.getTransportId();
@@ -82,21 +97,22 @@ public abstract class AbstractConduitSel
conduitInitiatorMgr.getConduitInitiator(transportID);
if (conduitInitiator != null) {
String add =
(String)message.get(Message.ENDPOINT_ADDRESS);
+ String basePath =
(String)message.get(Message.BASE_PATH);
if (StringUtils.isEmpty(add)
|| add.equals(ei.getAddress())) {
- replaceEndpointAddressPropertyIfNeeded(message,
add);
- selectedConduit = conduitInitiator.getConduit(ei);
+ c = conduitInitiator.getConduit(ei);
+ replaceEndpointAddressPropertyIfNeeded(message,
add, c);
} else {
EndpointReferenceType epr = new
EndpointReferenceType();
AttributedURIType ad = new AttributedURIType();
- ad.setValue(add);
+ ad.setValue(StringUtils.isEmpty(basePath) ? add :
basePath);
epr.setAddress(ad);
- selectedConduit = conduitInitiator.getConduit(ei,
epr);
+ c = conduitInitiator.getConduit(ei, epr);
}
MessageObserver observer =
exchange.get(MessageObserver.class);
if (observer != null) {
- selectedConduit.setMessageObserver(observer);
+ c.setMessageObserver(observer);
} else {
getLogger().warning("MessageObserver not found");
}
@@ -112,30 +128,23 @@ public abstract class AbstractConduitSel
} catch (IOException ex) {
throw new Fault(ex);
}
+ conduits.add(c);
+ }
+ if (c.getTarget() != null && c.getTarget().getAddress() != null) {
+ replaceEndpointAddressPropertyIfNeeded(message,
c.getTarget().getAddress().getValue(), c);
}
- return selectedConduit;
+ message.put(Conduit.class, c);
+ return c;
}
// Some conduits may replace the endpoint address after it has already
been prepared
// but before the invocation has been done (ex,
org.apache.cxf.clustering.LoadDistributorTargetSelector)
// which may affect JAX-RS clients where actual endpoint address property
may include additional path
// segments.
- protected void replaceEndpointAddressPropertyIfNeeded(Message message,
String endpointAddress) {
- String requestURI = (String)message.get(Message.REQUEST_URI);
- if (requestURI != null && !requestURI.startsWith(endpointAddress)) {
- String basePath = (String)message.get(Message.BASE_PATH);
- if (basePath != null && requestURI.startsWith(basePath)) {
- String pathInfo = requestURI.substring(basePath.length());
- final String slash = "/";
- boolean startsWithSlash = pathInfo.startsWith(slash);
- if (endpointAddress.endsWith(slash)) {
- endpointAddress = endpointAddress + (startsWithSlash ?
pathInfo.substring(1) : pathInfo);
- } else {
- endpointAddress = endpointAddress + (startsWithSlash ?
pathInfo : (slash + pathInfo));
- }
- message.put(Message.ENDPOINT_ADDRESS, endpointAddress);
- }
- }
+ protected boolean replaceEndpointAddressPropertyIfNeeded(Message message,
+ String
endpointAddress,
+ Conduit cond) {
+ return false;
}
/**
@@ -167,7 +176,12 @@ public abstract class AbstractConduitSel
}
try {
if (exchange.getInMessage() != null) {
-
getSelectedConduit(exchange.getInMessage()).close(exchange.getInMessage());
+ Conduit c = exchange.getOutMessage().get(Conduit.class);
+ if (c == null) {
+
getSelectedConduit(exchange.getInMessage()).close(exchange.getInMessage());
+ } else {
+ c.close(exchange.getInMessage());
+ }
}
} catch (IOException e) {
//IGNORE
@@ -177,4 +191,60 @@ public abstract class AbstractConduitSel
* @return the logger to use
*/
protected abstract Logger getLogger();
+
+ /**
+ * If address protocol was changed, conduit should be re-initialised
+ *
+ * @param message the current Message
+ */
+ protected Conduit findCompatibleConduit(Message message) {
+ Conduit c = message.get(Conduit.class);
+ if (c == null
+ && message.getExchange() != null
+ && message.getExchange().getOutMessage() != null
+ && message.getExchange().getOutMessage() != message) {
+ c = message.getExchange().getOutMessage().get(Conduit.class);
+ }
+ if (c != null) {
+ return c;
+ }
+ boolean full = MessageUtils.getContextualBoolean(message,
+
CONDUIT_COMPARE_FULL_URL,
+ false);
+ for (Conduit c2 : conduits) {
+ if (c2.getTarget() == null
+ || c2.getTarget().getAddress() == null
+ || c2.getTarget().getAddress().getValue() == null) {
+ continue;
+ }
+ String conduitAddress = c2.getTarget().getAddress().getValue();
+
+ EndpointInfo ei = endpoint.getEndpointInfo();
+ String actualAddress = ei.getAddress();
+
+ String messageAddress =
(String)message.get(Message.ENDPOINT_ADDRESS);
+ if (messageAddress != null) {
+ actualAddress = messageAddress;
+ }
+
+ if (!full) {
+ int idx = conduitAddress.indexOf(':');
+ conduitAddress = idx == -1 ? "" : conduitAddress.substring(0,
idx);
+ idx = actualAddress.indexOf(':');
+ actualAddress = idx == -1 ? "" : actualAddress.substring(0,
idx);
+ }
+
+ if (conduitAddress.equalsIgnoreCase(actualAddress)) {
+ return c2;
+ }
+ }
+ for (Conduit c2 : conduits) {
+ if (c2.getTarget() == null
+ || c2.getTarget().getAddress() == null
+ || c2.getTarget().getAddress().getValue() == null) {
+ return c2;
+ }
+ }
+ return null;
+ }
}
Modified:
cxf/trunk/api/src/main/java/org/apache/cxf/endpoint/DeferredConduitSelector.java
URL:
http://svn.apache.org/viewvc/cxf/trunk/api/src/main/java/org/apache/cxf/endpoint/DeferredConduitSelector.java?rev=1305204&r1=1305203&r2=1305204&view=diff
==============================================================================
---
cxf/trunk/api/src/main/java/org/apache/cxf/endpoint/DeferredConduitSelector.java
(original)
+++
cxf/trunk/api/src/main/java/org/apache/cxf/endpoint/DeferredConduitSelector.java
Mon Mar 26 02:53:14 2012
@@ -38,7 +38,7 @@ public class DeferredConduitSelector ext
* Normal constructor.
*/
public DeferredConduitSelector() {
- this(null);
+ super();
}
/**
@@ -66,7 +66,12 @@ public class DeferredConduitSelector ext
* @return the Conduit to use for mediation of the message
*/
public Conduit selectConduit(Message message) {
- return getSelectedConduit(message);
+ Conduit c = message.get(Conduit.class);
+ if (c == null) {
+ c = getSelectedConduit(message);
+ message.put(Conduit.class, c);
+ }
+ return c;
}
/**
Modified:
cxf/trunk/api/src/main/java/org/apache/cxf/endpoint/UpfrontConduitSelector.java
URL:
http://svn.apache.org/viewvc/cxf/trunk/api/src/main/java/org/apache/cxf/endpoint/UpfrontConduitSelector.java?rev=1305204&r1=1305203&r2=1305204&view=diff
==============================================================================
---
cxf/trunk/api/src/main/java/org/apache/cxf/endpoint/UpfrontConduitSelector.java
(original)
+++
cxf/trunk/api/src/main/java/org/apache/cxf/endpoint/UpfrontConduitSelector.java
Mon Mar 26 02:53:14 2012
@@ -40,7 +40,7 @@ public class UpfrontConduitSelector exte
* Normal constructor.
*/
public UpfrontConduitSelector() {
- this(null);
+ super();
}
/**
@@ -58,7 +58,11 @@ public class UpfrontConduitSelector exte
* @param message the current Message
*/
public void prepare(Message message) {
- getSelectedConduit(message);
+ Conduit c = message.get(Conduit.class);
+ if (c == null) {
+ getSelectedConduit(message);
+ message.put(Conduit.class, c);
+ }
}
/**
@@ -68,7 +72,12 @@ public class UpfrontConduitSelector exte
* @return the Conduit to use for mediation of the message
*/
public Conduit selectConduit(Message message) {
- return getSelectedConduit(message);
+ Conduit c = message.get(Conduit.class);
+ if (c == null) {
+ c = getSelectedConduit(message);
+ message.put(Conduit.class, c);
+ }
+ return c;
}
/**
Modified:
cxf/trunk/rt/features/clustering/src/main/java/org/apache/cxf/clustering/FailoverFeature.java
URL:
http://svn.apache.org/viewvc/cxf/trunk/rt/features/clustering/src/main/java/org/apache/cxf/clustering/FailoverFeature.java?rev=1305204&r1=1305203&r2=1305204&view=diff
==============================================================================
---
cxf/trunk/rt/features/clustering/src/main/java/org/apache/cxf/clustering/FailoverFeature.java
(original)
+++
cxf/trunk/rt/features/clustering/src/main/java/org/apache/cxf/clustering/FailoverFeature.java
Mon Mar 26 02:53:14 2012
@@ -63,7 +63,7 @@ public class FailoverFeature extends Abs
return selector;
}
- protected FailoverTargetSelector getTargetSelector() {
+ public FailoverTargetSelector getTargetSelector() {
if (this.targetSelector == null) {
this.targetSelector = new FailoverTargetSelector();
}
Modified:
cxf/trunk/rt/features/clustering/src/main/java/org/apache/cxf/clustering/FailoverTargetSelector.java
URL:
http://svn.apache.org/viewvc/cxf/trunk/rt/features/clustering/src/main/java/org/apache/cxf/clustering/FailoverTargetSelector.java?rev=1305204&r1=1305203&r2=1305204&view=diff
==============================================================================
---
cxf/trunk/rt/features/clustering/src/main/java/org/apache/cxf/clustering/FailoverTargetSelector.java
(original)
+++
cxf/trunk/rt/features/clustering/src/main/java/org/apache/cxf/clustering/FailoverTargetSelector.java
Mon Mar 26 02:53:14 2012
@@ -48,14 +48,15 @@ public class FailoverTargetSelector exte
private static final Logger LOG =
LogUtils.getL7dLogger(FailoverTargetSelector.class);
- protected Map<InvocationKey, InvocationContext> inProgress;
+ protected Map<InvocationKey, InvocationContext> inProgress
+ = new ConcurrentHashMap<InvocationKey, InvocationContext>();;
protected FailoverStrategy failoverStrategy;
/**
* Normal constructor.
*/
public FailoverTargetSelector() {
- this(null);
+ super();
}
/**
@@ -65,7 +66,6 @@ public class FailoverTargetSelector exte
*/
public FailoverTargetSelector(Conduit c) {
super(c);
- inProgress = new ConcurrentHashMap<InvocationKey, InvocationContext>();
}
/**
@@ -99,6 +99,10 @@ public class FailoverTargetSelector exte
* @return the Conduit to use for mediation of the message
*/
public Conduit selectConduit(Message message) {
+ Conduit c = message.get(Conduit.class);
+ if (c != null) {
+ return c;
+ }
return getSelectedConduit(message);
}
@@ -115,11 +119,15 @@ public class FailoverTargetSelector exte
}
boolean failover = false;
if (requiresFailover(exchange)) {
+ Conduit old =
(Conduit)exchange.getOutMessage().remove(Conduit.class.getName());
+
Endpoint failoverTarget = getFailoverTarget(exchange, invocation);
if (failoverTarget != null) {
setEndpoint(failoverTarget);
- selectedConduit.close();
- selectedConduit = null;
+ if (old != null) {
+ old.close();
+ conduits.remove(old);
+ }
Exception prevExchangeFault =
(Exception)exchange.remove(Exception.class.getName());
Message outMessage = exchange.getOutMessage();
@@ -155,7 +163,7 @@ public class FailoverTargetSelector exte
}
}
if (!failover) {
- getLogger().info("FAILOVER_NOT_REQUIRED");
+ getLogger().fine("FAILOVER_NOT_REQUIRED");
synchronized (this) {
inProgress.remove(key);
}
@@ -281,14 +289,52 @@ public class FailoverTargetSelector exte
* @param context the request context
*/
protected void overrideAddressProperty(Map<String, Object> context) {
+ overrideAddressProperty(context,
getEndpoint().getEndpointInfo().getAddress());
+ }
+
+ protected void overrideAddressProperty(Map<String, Object> context,
+ String address) {
Map<String, Object> requestContext =
CastUtils.cast((Map<?, ?>)context.get(Client.REQUEST_CONTEXT));
if (requestContext != null) {
- requestContext.put(Message.ENDPOINT_ADDRESS,
- getEndpoint().getEndpointInfo().getAddress());
- requestContext.put("javax.xml.ws.service.endpoint.address",
- getEndpoint().getEndpointInfo().getAddress());
+ requestContext.put(Message.ENDPOINT_ADDRESS, address);
+ requestContext.put("javax.xml.ws.service.endpoint.address",
address);
+ }
+ }
+
+ // Some conduits may replace the endpoint address after it has already
been prepared
+ // but before the invocation has been done (ex,
org.apache.cxf.clustering.LoadDistributorTargetSelector)
+ // which may affect JAX-RS clients where actual endpoint address property
may include additional path
+ // segments.
+ protected boolean replaceEndpointAddressPropertyIfNeeded(Message message,
+ String
endpointAddress,
+ Conduit cond) {
+ String requestURI = (String)message.get(Message.REQUEST_URI);
+ if (requestURI != null && endpointAddress != null &&
!requestURI.startsWith(endpointAddress)) {
+ String basePath = (String)message.get(Message.BASE_PATH);
+ if (basePath != null && requestURI.startsWith(basePath)) {
+ String pathInfo = requestURI.substring(basePath.length());
+ message.put(Message.BASE_PATH, endpointAddress);
+ final String slash = "/";
+ boolean startsWithSlash = pathInfo.startsWith(slash);
+ if (endpointAddress.endsWith(slash)) {
+ endpointAddress = endpointAddress + (startsWithSlash ?
pathInfo.substring(1) : pathInfo);
+ } else {
+ endpointAddress = endpointAddress + (startsWithSlash ?
pathInfo : (slash + pathInfo));
+ }
+ message.put(Message.ENDPOINT_ADDRESS, endpointAddress);
+
+ Exchange exchange = message.getExchange();
+ InvocationKey key = new InvocationKey(exchange);
+ InvocationContext invocation = inProgress.get(key);
+ if (invocation != null) {
+ overrideAddressProperty(invocation.getContext(),
+
cond.getTarget().getAddress().getValue());
+ }
+ return true;
+ }
}
+ return false;
}
/**
Modified:
cxf/trunk/rt/features/clustering/src/main/java/org/apache/cxf/clustering/LoadDistributorFeature.java
URL:
http://svn.apache.org/viewvc/cxf/trunk/rt/features/clustering/src/main/java/org/apache/cxf/clustering/LoadDistributorFeature.java?rev=1305204&r1=1305203&r2=1305204&view=diff
==============================================================================
---
cxf/trunk/rt/features/clustering/src/main/java/org/apache/cxf/clustering/LoadDistributorFeature.java
(original)
+++
cxf/trunk/rt/features/clustering/src/main/java/org/apache/cxf/clustering/LoadDistributorFeature.java
Mon Mar 26 02:53:14 2012
@@ -30,7 +30,7 @@ import org.apache.cxf.common.injection.N
public class LoadDistributorFeature extends FailoverFeature {
@Override
- protected FailoverTargetSelector getTargetSelector() {
+ public FailoverTargetSelector getTargetSelector() {
return new LoadDistributorTargetSelector();
}
}
Modified:
cxf/trunk/rt/features/clustering/src/main/java/org/apache/cxf/clustering/LoadDistributorTargetSelector.java
URL:
http://svn.apache.org/viewvc/cxf/trunk/rt/features/clustering/src/main/java/org/apache/cxf/clustering/LoadDistributorTargetSelector.java?rev=1305204&r1=1305203&r2=1305204&view=diff
==============================================================================
---
cxf/trunk/rt/features/clustering/src/main/java/org/apache/cxf/clustering/LoadDistributorTargetSelector.java
(original)
+++
cxf/trunk/rt/features/clustering/src/main/java/org/apache/cxf/clustering/LoadDistributorTargetSelector.java
Mon Mar 26 02:53:14 2012
@@ -67,7 +67,7 @@ public class LoadDistributorTargetSelect
* Normal constructor.
*/
public LoadDistributorTargetSelector() {
- this(null);
+ super();
}
/**
@@ -98,7 +98,11 @@ public class LoadDistributorTargetSelect
* @param message
* @return the Conduit to use for mediation of the message
*/
- public Conduit selectConduit(Message message) {
+ public synchronized Conduit selectConduit(Message message) {
+ Conduit c = message.get(Conduit.class);
+ if (c != null) {
+ return c;
+ }
Exchange exchange = message.getExchange();
InvocationKey key = new InvocationKey(exchange);
InvocationContext invocation = inProgress.get(key);
@@ -106,11 +110,8 @@ public class LoadDistributorTargetSelect
Endpoint target = getDistributionTarget(exchange, invocation);
if (target != null) {
setEndpoint(target);
- if (selectedConduit != null) {
- selectedConduit.close();
- selectedConduit = null;
- }
message.put(Message.ENDPOINT_ADDRESS,
target.getEndpointInfo().getAddress());
+ message.put(CONDUIT_COMPARE_FULL_URL, Boolean.TRUE);
overrideAddressProperty(invocation.getContext());
invocation.getContext().put(IS_DISTRIBUTED, null);
}
Modified:
cxf/trunk/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/client/AbstractClient.java
URL:
http://svn.apache.org/viewvc/cxf/trunk/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/client/AbstractClient.java?rev=1305204&r1=1305203&r2=1305204&view=diff
==============================================================================
---
cxf/trunk/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/client/AbstractClient.java
(original)
+++
cxf/trunk/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/client/AbstractClient.java
Mon Mar 26 02:53:14 2012
@@ -24,6 +24,7 @@ import java.lang.annotation.Annotation;
import java.lang.reflect.Type;
import java.net.HttpURLConnection;
import java.net.URI;
+import java.net.URISyntaxException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Arrays;
@@ -466,6 +467,14 @@ public abstract class AbstractClient imp
// operating on InputStream don't have to close streams pro-actively
exchange.put(KEEP_CONDUIT_ALIVE, true);
getConfiguration().getConduitSelector().complete(exchange);
+ try {
+ String s = (String)exchange.getOutMessage().get(Message.BASE_PATH);
+ if (s != null) {
+ state.setBaseURI(new URI(s));
+ }
+ } catch (URISyntaxException e) {
+ //ignore
+ }
}
protected Object[] preProcessResult(Message message) throws Exception {
Modified:
cxf/trunk/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/failover/FailoverTest.java
URL:
http://svn.apache.org/viewvc/cxf/trunk/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/failover/FailoverTest.java?rev=1305204&r1=1305203&r2=1305204&view=diff
==============================================================================
---
cxf/trunk/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/failover/FailoverTest.java
(original)
+++
cxf/trunk/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/failover/FailoverTest.java
Mon Mar 26 02:53:14 2012
@@ -248,6 +248,7 @@ public class FailoverTest extends Abstra
}
for (int i = 0; i < 20; i++) {
+ feature.getTargetSelector().close();
if (!singleProxy) {
bookStore = getBookStore(inactiveReplica, feature);
}
@@ -273,7 +274,7 @@ public class FailoverTest extends Abstra
if (expectRandom) {
assertTrue(currEndpoint.equals(activeReplica1) ||
currEndpoint.equals(activeReplica2));
} else {
- assertTrue(currEndpoint.equals(activeReplica1));
+ assertEquals(activeReplica1 , currEndpoint);
}
if (expectServerException) {
assertNotNull(ex);
@@ -301,6 +302,7 @@ public class FailoverTest extends Abstra
boolean randomized = false;
String prevEndpoint = null;
for (int i = 0; i < 20; i++) {
+ feature.getTargetSelector().close();
WebClient bookStore = getWebClient(inactiveReplica, feature);
verifyStrategy(bookStore, expectRandom
? RandomStrategy.class
