You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@synapse.apache.org by ru...@apache.org on 2008/05/26 20:41:00 UTC
svn commit: r660268 - in
/synapse/branches/1.2/modules/core/src/main/java/org/apache/synapse/endpoints:
EndpointContext.java SALoadbalanceEndpoint.java
algorithms/AlgorithmContext.java dispatch/DispatcherContext.java
dispatch/HttpSessionDispatcher.java
Author: ruwan
Date: Mon May 26 11:40:57 2008
New Revision: 660268
URL: http://svn.apache.org/viewvc?rev=660268&view=rev
Log:
* Fixing the issue with the transportsession, session affinity load balancing when operating on a cluster environment (SYNAPSE-325)
* Added logs to the transport session execution
Modified:
synapse/branches/1.2/modules/core/src/main/java/org/apache/synapse/endpoints/EndpointContext.java
synapse/branches/1.2/modules/core/src/main/java/org/apache/synapse/endpoints/SALoadbalanceEndpoint.java
synapse/branches/1.2/modules/core/src/main/java/org/apache/synapse/endpoints/algorithms/AlgorithmContext.java
synapse/branches/1.2/modules/core/src/main/java/org/apache/synapse/endpoints/dispatch/DispatcherContext.java
synapse/branches/1.2/modules/core/src/main/java/org/apache/synapse/endpoints/dispatch/HttpSessionDispatcher.java
Modified: synapse/branches/1.2/modules/core/src/main/java/org/apache/synapse/endpoints/EndpointContext.java
URL: http://svn.apache.org/viewvc/synapse/branches/1.2/modules/core/src/main/java/org/apache/synapse/endpoints/EndpointContext.java?rev=660268&r1=660267&r2=660268&view=diff
==============================================================================
--- synapse/branches/1.2/modules/core/src/main/java/org/apache/synapse/endpoints/EndpointContext.java (original)
+++ synapse/branches/1.2/modules/core/src/main/java/org/apache/synapse/endpoints/EndpointContext.java Mon May 26 11:40:57 2008
@@ -26,10 +26,10 @@
import org.apache.synapse.SynapseException;
/**
- * Keeps the states of the endpoint.This hides where those states are kept .For a cluster environment
- * ,all states are kept in the axis2 configuration context in order to replicate those states so that
- * other synapse instance in the same cluster can see those changes . This class can be evolved to
- * keep any run time states related to the endpoint .For a non-clustered environment ,
+ * Keeps the states of the endpoint.This hides where those states are kept .For a cluster
+ * environment,all states are kept in the axis2 configuration context in order to replicate those
+ * states so that other synapse instance in the same cluster can see those changes . This class can
+ * be evolved to keep any run time states related to the endpoint .For a non-clustered environment,
* all data are kept locally.
* <p/>
* This class provide the abstraction need to separate the dynamic data from the static data and
@@ -41,7 +41,7 @@
private static final Log log = LogFactory.getLog(EndpointContext.class);
/* The static constant only for construct key prefix for each property in endpoint context
- *as it is need when those property state going to replicate in a cluster env. */
+ * as it is need when those property state going to replicate in a cluster env. */
private static final String ACTIVE = "active";
private static final String RECOVER_ON = "recover_on";
private static final String UNDERSCORE_STRING = "_";
@@ -53,13 +53,14 @@
/* Time to recover a failed endpoint.*/
private long recoverOn = Long.MAX_VALUE;
- /*The axis configuration context- this will hold the all callers states
+ /* The axis configuration context- this will hold the all callers states
* when doing throttling in a clustered environment. */
private ConfigurationContext configCtx;
- /*The key for 'active' attribute and this is used when this attribute value being replicated */
+ /* The key for 'active' attribute and this is used when this attribute value being replicated */
private String activePropertyKey;
- /*The key for 'recoverOn' attribute and this is used when this attribute value being replicated */
+ /* The key for 'recoverOn' attribute and this is used when this attribute value being
+ * replicated */
private String recoverOnPropertyKey;
/* Is this env. support clustering*/
@@ -232,13 +233,17 @@
try {
if (log.isDebugEnabled()) {
- log.debug("Going to replicate the property with key : " + key +
+ log.debug("Start replicating the property with key : " + key +
" value : " + value);
}
configCtx.setProperty(key, value);
Replicator.replicate(configCtx, new String[]{key});
+ if (log.isDebugEnabled()) {
+ log.debug("Completed replication of the property with key : " + key);
+ }
+
} catch (ClusteringFault clusteringFault) {
handleException("Error during the replicating states ", clusteringFault);
}
@@ -251,7 +256,6 @@
* @param msg The error message
*/
protected void handleException(String msg) {
-
log.error(msg);
throw new SynapseException(msg);
}
@@ -263,8 +267,7 @@
* @param e The exception
*/
protected void handleException(String msg, Exception e) {
-
log.error(msg, e);
throw new SynapseException(msg, e);
}
-}
+}
\ No newline at end of file
Modified: synapse/branches/1.2/modules/core/src/main/java/org/apache/synapse/endpoints/SALoadbalanceEndpoint.java
URL: http://svn.apache.org/viewvc/synapse/branches/1.2/modules/core/src/main/java/org/apache/synapse/endpoints/SALoadbalanceEndpoint.java?rev=660268&r1=660267&r2=660268&view=diff
==============================================================================
--- synapse/branches/1.2/modules/core/src/main/java/org/apache/synapse/endpoints/SALoadbalanceEndpoint.java (original)
+++ synapse/branches/1.2/modules/core/src/main/java/org/apache/synapse/endpoints/SALoadbalanceEndpoint.java Mon May 26 11:40:57 2008
@@ -205,7 +205,7 @@
if (dispatcher.isServerInitiatedSession()) {
if (log.isDebugEnabled()) {
- log.debug("Creating a new server initiated session for the current message");
+ log.debug("Adding a new server initiated session for the current message");
}
// add this endpoint to the endpoint sequence of operation context.
@@ -213,17 +213,14 @@
OperationContext opCtx = axis2MsgCtx.getAxis2MessageContext().getOperationContext();
if (isClusteringEnable) {
- // If running on a cluster keeps endpoint names, because it is heavy task to
+ // If running on a cluster keep endpoint names, because it is heavy to
// replicate endpoint itself
Object o = opCtx.getPropertyNonReplicable(ENDPOINT_NAME_LIST);
List<String> epNameList;
if (o instanceof List) {
-
epNameList = (List<String>) o;
- // todo : check the following statement : ruwan
epNameList.add(endpointName);
-
} else {
// this is the first endpoint in the heirachy. so create the queue and
// insert this as the first element.
@@ -235,7 +232,14 @@
// if the next endpoint is not a session affinity one, endpoint sequence ends
// here. but we have to add the next endpoint to the list.
if (!(endpoint instanceof SALoadbalanceEndpoint)) {
- String name = endpoint.getName();
+
+ String name;
+ if (endpoint instanceof IndirectEndpoint) {
+ name = ((IndirectEndpoint) endpoint).getKey();
+ } else {
+ name = endpoint.getName();
+ }
+
if (name == null) {
log.warn(WARN_MESSAGE);
name = SynapseConstants.ANONYMOUS_ENDPOINT;
@@ -326,8 +330,7 @@
Endpoint endpoint = null;
if (isClusteringEnable) {
- // if this is a clustering env.
- // Only keeps endpoint names , because , it is heavy task to
+ // if this is a clustering env. only keep endpoint names, because, it is heavy to
// replicate endpoint itself
String epNameObj = (String) endpointList.remove(0);
for (Endpoint ep : endpoints) {
Modified: synapse/branches/1.2/modules/core/src/main/java/org/apache/synapse/endpoints/algorithms/AlgorithmContext.java
URL: http://svn.apache.org/viewvc/synapse/branches/1.2/modules/core/src/main/java/org/apache/synapse/endpoints/algorithms/AlgorithmContext.java?rev=660268&r1=660267&r2=660268&view=diff
==============================================================================
--- synapse/branches/1.2/modules/core/src/main/java/org/apache/synapse/endpoints/algorithms/AlgorithmContext.java (original)
+++ synapse/branches/1.2/modules/core/src/main/java/org/apache/synapse/endpoints/algorithms/AlgorithmContext.java Mon May 26 11:40:57 2008
@@ -45,17 +45,18 @@
private static final String UNDERSCORE_STRING = "_";
private static final String CURRENT_EPR = "currentEPR";
- /*The axis configuration context- this will hold the all callers states
- *when doing throttling in a clustered environment. */
+ /* The axis configuration context- this will hold the all callers states
+ * when doing throttling in a clustered environment. */
private ConfigurationContext configCtx;
/* Is this env. support clustering*/
private boolean isClusteringEnable = false;
- /*The key for 'currentEPR' attribute and this is used when this attribute value being replicated */
+ /* The key for 'currentEPR' attribute and this is used when this attribute value being
+ * replicated */
private String currentEPRPropertyKey;
- /*The pointer to current epr - The position of the current EPR */
+ /* The pointer to current epr - The position of the current EPR */
private int currentEPR = 0;
/**
@@ -102,10 +103,17 @@
if (isClusteringEnable) { // if this is a clustering env.
if (currentEPRPropertyKey != null) {
+ if (log.isDebugEnabled()) {
+ log.debug("Setting the current EPR " + currentEPR
+ + " with the key " + currentEPRPropertyKey);
+ }
// Sets the property and replicates the current state so that all instances
setAndReplicateState(currentEPRPropertyKey, currentEPR);
}
} else {
+ if (log.isDebugEnabled()) {
+ log.debug("Setting the current EPR " + currentEPR);
+ }
this.currentEPR = currentEPR;
}
}
@@ -155,7 +163,6 @@
buffer.append(UNDERSCORE_STRING);
buffer.append(CURRENT_EPR);
currentEPRPropertyKey = buffer.toString();
-
}
@@ -165,7 +172,6 @@
* @param msg The error message
*/
protected void handleException(String msg) {
-
log.error(msg);
throw new SynapseException(msg);
}
@@ -177,7 +183,6 @@
* @param e The exception
*/
protected void handleException(String msg, Exception e) {
-
log.error(msg, e);
throw new SynapseException(msg, e);
}
@@ -196,17 +201,21 @@
try {
if (log.isDebugEnabled()) {
- log.debug("Going to replicate the property with key : " + key
+ log.debug("Start replicating the property with key : " + key
+ " value : " + value);
}
configCtx.setProperty(key, value);
Replicator.replicate(configCtx, new String[]{key});
+ if (log.isDebugEnabled()) {
+ log.debug("Completed replication of the property with key : " + key);
+ }
+
} catch (ClusteringFault clusteringFault) {
handleException("Error during the replicating states ", clusteringFault);
}
}
}
-}
+}
\ No newline at end of file
Modified: synapse/branches/1.2/modules/core/src/main/java/org/apache/synapse/endpoints/dispatch/DispatcherContext.java
URL: http://svn.apache.org/viewvc/synapse/branches/1.2/modules/core/src/main/java/org/apache/synapse/endpoints/dispatch/DispatcherContext.java?rev=660268&r1=660267&r2=660268&view=diff
==============================================================================
--- synapse/branches/1.2/modules/core/src/main/java/org/apache/synapse/endpoints/dispatch/DispatcherContext.java (original)
+++ synapse/branches/1.2/modules/core/src/main/java/org/apache/synapse/endpoints/dispatch/DispatcherContext.java Mon May 26 11:40:57 2008
@@ -42,8 +42,8 @@
* For a non-clustered environment , all data are kept locally.
* <p/>
* This class provide the abstraction need to separate the dynamic data from the static data and
- * improve the high cohesion and provides capability to replicate only required state at a given time
- * This improves the performance when replicate data.
+ * improve the high cohesion and provides capability to replicate only required state at a given
+ * time. This improves the performance when replicate data.
*/
public class DispatcherContext {
@@ -87,15 +87,21 @@
handleException("Cannot find the required key prefix to find the " +
"shared state of one of 'session'");
}
- // gets the value from configuration context (The shared state across all instances )
+ // gets the value from configuration context (The shared state across all instances)
Object value = this.configCtx.getPropertyNonReplicable(this.keyPrefix + sessionID);
if (value != null && value instanceof String) {
+ if (log.isDebugEnabled()) {
+ log.debug("Retrieving the endpoint from the session id " + value);
+ }
return endpointsMap.get(value.toString());
}
} else {
synchronized (sessionMap) {
+ if (log.isDebugEnabled()) {
+ log.debug("Retrieving the endpoint from the session id " + sessionID);
+ }
return sessionMap.get(sessionID);
}
}
@@ -120,8 +126,7 @@
endpointName = endpoint.getName();
}
- if (endpointName == null && !(endpoint instanceof IndirectEndpoint)) {
-
+ if (endpointName == null) {
if (log.isDebugEnabled() && isClusteringEnable()) {
log.warn(SALoadbalanceEndpoint.WARN_MESSAGE);
}
@@ -129,6 +134,10 @@
}
if (keyPrefix != null) {
+ if (log.isDebugEnabled()) {
+ log.debug("Adding the enpoint " + endpointName + " with the session id "
+ + keyPrefix + sessionID + " for replication to the session");
+ }
// replicates the state so that all instances across cluster can see this state
setAndReplicateState(keyPrefix + sessionID, endpointName);
}
@@ -136,6 +145,10 @@
} else {
synchronized (sessionMap) {
+ if (log.isDebugEnabled()) {
+ log.debug("Adding the endpoint " + endpoint
+ + " with the session id " + sessionID + " to the session");
+ }
sessionMap.put(sessionID, endpoint);
}
}
@@ -152,15 +165,21 @@
if (isClusteringEnable) { // if this is a clustering env.
if (keyPrefix != null) {
+ if (log.isDebugEnabled()) {
+ log.debug("Removing and replicating " +
+ "the session with the session id " + keyPrefix + id);
+ }
//Removes the endpoint name and then replicates the current
//state so that all instances
removeAndReplicateState(keyPrefix + id);
-
}
} else {
synchronized (sessionMap) {
+ if (log.isDebugEnabled()) {
+ log.debug("Removing the session with the session id " + id);
+ }
sessionMap.remove(id);
}
}
@@ -221,7 +240,6 @@
* @param msg The error message
*/
protected void handleException(String msg) {
-
log.error(msg);
throw new SynapseException(msg);
}
@@ -233,7 +251,6 @@
* @param e The exception
*/
protected void handleException(String msg, Exception e) {
-
log.error(msg, e);
throw new SynapseException(msg, e);
}
@@ -252,7 +269,7 @@
try {
if (log.isDebugEnabled()) {
- log.debug("Going to replicate the property with key : " +
+ log.debug("Start replicating the property with key : " +
key + " value : " + value);
}
@@ -262,6 +279,10 @@
Replicator.replicate(configCtx, new String[]{key});
}
+ if (log.isDebugEnabled()) {
+ log.debug("Completed replication of the property with key: " + key);
+ }
+
} catch (ClusteringFault clusteringFault) {
handleException("Error during the replicating states ", clusteringFault);
}
@@ -281,12 +302,16 @@
try {
if (log.isDebugEnabled()) {
- log.debug("Going to replicate the property with key : " + key);
+ log.debug("Start replicating the property removal with key : " + key);
}
configCtx.removeProperty(key);
Replicator.replicate(configCtx, new String[]{key});
+ if (log.isDebugEnabled()) {
+ log.debug("Completed replication of the property removal with key : " + key);
+ }
+
} catch (ClusteringFault clusteringFault) {
handleException("Error during the replicating states ", clusteringFault);
}
@@ -327,8 +352,12 @@
endpointName = SynapseConstants.ANONYMOUS_ENDPOINT;
}
+ if (log.isDebugEnabled()) {
+ log.debug("Adding an endpoint with the name/key "
+ + endpointName + " to the endpoints map");
+ }
endpointsMap.put(endpointName, endpoint);
}
}
}
-}
+}
\ No newline at end of file
Modified: synapse/branches/1.2/modules/core/src/main/java/org/apache/synapse/endpoints/dispatch/HttpSessionDispatcher.java
URL: http://svn.apache.org/viewvc/synapse/branches/1.2/modules/core/src/main/java/org/apache/synapse/endpoints/dispatch/HttpSessionDispatcher.java?rev=660268&r1=660267&r2=660268&view=diff
==============================================================================
--- synapse/branches/1.2/modules/core/src/main/java/org/apache/synapse/endpoints/dispatch/HttpSessionDispatcher.java (original)
+++ synapse/branches/1.2/modules/core/src/main/java/org/apache/synapse/endpoints/dispatch/HttpSessionDispatcher.java Mon May 26 11:40:57 2008
@@ -28,9 +28,9 @@
import java.util.Map;
/**
- * Dispatches sessions based on HTTP cookies. Session is initiated by the server in the first response
- * when it sends "Set-Cookie" HTTP header with the session ID. For all successive messages client
- * should send "Cookie" HTTP header with session ID send by the server.
+ * Dispatches sessions based on HTTP cookies. Session is initiated by the server in the first
+ * response when it sends "Set-Cookie" HTTP header with the session ID. For all successive messages
+ * client should send "Cookie" HTTP header with session ID send by the server.
*/
public class HttpSessionDispatcher implements Dispatcher {
@@ -42,9 +42,9 @@
private final static String SET_COOKIE = "Set-Cookie";
/**
- * Check if "Cookie" HTTP header is available. If so, check if that cookie is in the session map.
- * If cookie is available, there is a session for this cookie. return the (server) endpoint for
- * that session.
+ * Check if "Cookie" HTTP header is available. If so, check if that cookie is in the session
+ * map. If cookie is available, there is a session for this cookie. return the (server)
+ * endpoint for that session.
*
* @param synCtx MessageContext possibly containing a "Cookie" HTTP header.
* @return Endpoint Server endpoint for the given HTTP session.
@@ -62,10 +62,23 @@
Object cookie = headerMap.get(COOKIE);
if (cookie != null && cookie instanceof String) {
+
+ if (log.isDebugEnabled()) {
+ log.debug("Using the HTTP header 'Cookie: " + cookie
+ + "' to retrieve the endpoint in the transport session");
+ }
+
Object ep = dispatcherContext.getEndpoint((String) cookie);
if (ep != null && ep instanceof Endpoint) {
endpoint = (Endpoint) ep;
+ } else if (log.isDebugEnabled()) {
+ log.debug("No endpoint found in the transport " +
+ "session for the session id " + cookie);
}
+
+ } else if (log.isDebugEnabled()) {
+ log.debug("No 'Cookie' HTTP headers found to extract the " +
+ "endpoint from the transport session");
}
}
@@ -80,7 +93,8 @@
* @param synCtx MessageContext possibly containing the "Set-Cookie" HTTP header.
* @param endpoint Endpoint to be mapped to the session.
*/
- public void updateSession(MessageContext synCtx, DispatcherContext dispatcherContext, Endpoint endpoint) {
+ public void updateSession(MessageContext synCtx, DispatcherContext dispatcherContext,
+ Endpoint endpoint) {
if (endpoint == null || dispatcherContext == null) {
return;
@@ -103,8 +117,22 @@
// will result in the session id "JSESSIONID=760764CB72E96A7221506823748CF2AE"
// and the client is expected to send the Cookie header as;
// Cookie: JSESSIONID=760764CB72E96A7221506823748CF2AE
+ if (log.isDebugEnabled()) {
+ log.debug("Found the HTTP header 'Set-Cookie: "
+ + cookie + "' for updating the session");
+ }
String sessionId = ((String) cookie).split(";")[0];
+
+ if (log.isDebugEnabled()) {
+ log.debug("Using the session id '" + sessionId +
+ "' extracted from the Set-Cookie header to update the session " +
+ "with the endpoint " + endpoint);
+ }
dispatcherContext.setEndpoint(sessionId, endpoint);
+
+ } else if (log.isDebugEnabled()) {
+ log.debug("No 'Set-Cookie' HTTP header is specified in " +
+ "the message to update the session");
}
}
}
@@ -120,7 +148,16 @@
Object cookie = headerMap.get(COOKIE);
if (cookie != null && cookie instanceof String) {
+
+ if (log.isDebugEnabled()) {
+ log.debug("Using the HTTP header 'Cookie: "
+ + cookie + "' to unbind the session");
+ }
dispatcherContext.removeSession((String) cookie);
+
+ } else if (log.isDebugEnabled()) {
+ log.debug("No 'Cookie' HTTP header is specified in " +
+ "the message to unbind the session");
}
}
}