You are viewing a plain text version of this content. The canonical link for it is here.
Posted to axis-cvs@ws.apache.org by az...@apache.org on 2009/02/16 13:03:45 UTC
svn commit: r744900 [2/3] - in /webservices/axis2/trunk/java/modules:
clustering/src/org/apache/axis2/clustering/
clustering/src/org/apache/axis2/clustering/configuration/
clustering/src/org/apache/axis2/clustering/context/
clustering/src/org/apache/ax...
Added: webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/ClusterManagementMode.java
URL: http://svn.apache.org/viewvc/webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/ClusterManagementMode.java?rev=744900&view=auto
==============================================================================
--- webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/ClusterManagementMode.java (added)
+++ webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/ClusterManagementMode.java Mon Feb 16 12:02:42 2009
@@ -0,0 +1,152 @@
+/*
+ * Copyright 2004,2005 The Apache Software Foundation.
+ *
+ * Licensed 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.axis2.clustering.tribes;
+
+import org.apache.axis2.clustering.management.GroupManagementAgent;
+import org.apache.axis2.clustering.management.DefaultGroupManagementAgent;
+import org.apache.axis2.clustering.control.wka.MemberJoinedCommand;
+import org.apache.catalina.tribes.Channel;
+import org.apache.catalina.tribes.ChannelException;
+import org.apache.catalina.tribes.Member;
+import org.apache.catalina.tribes.MembershipListener;
+import org.apache.catalina.tribes.RemoteProcessException;
+import org.apache.catalina.tribes.group.RpcChannel;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Represents a member running in load balance mode
+ */
+public class ClusterManagementMode implements OperationMode {
+
+ private static final Log log = LogFactory.getLog(ClusterManagementMode.class);
+
+ private byte[] clusterManagerDomain;
+ private Map<String, GroupManagementAgent> groupManagementAgents;
+ private List<MembershipManager> membershipManagers = new ArrayList<MembershipManager>();
+ private MembershipManager primaryMembershipManager;
+
+ public ClusterManagementMode(byte[] clusterManagerDomain,
+ Map<String, GroupManagementAgent> groupManagementAgents,
+ MembershipManager primaryMembershipManager) {
+ this.clusterManagerDomain = clusterManagerDomain;
+ this.groupManagementAgents = groupManagementAgents;
+ this.primaryMembershipManager = primaryMembershipManager;
+ }
+
+ public void addInterceptors(Channel channel) {
+ ClusterManagementInterceptor interceptor =
+ new ClusterManagementInterceptor(clusterManagerDomain);
+ interceptor.setOptionFlag(TribesConstants.MEMBERSHIP_MSG_OPTION);
+ channel.addInterceptor(interceptor);
+ if (log.isDebugEnabled()) {
+ log.debug("Added ClusterManagementInterceptor");
+ }
+ }
+
+ public void init(Channel channel) {
+ // Have multiple RPC channels with multiple RPC request handlers for each domain
+ // This is needed only when this member is running as a load balancer
+ for (Object o : groupManagementAgents.keySet()) {
+ String domain = (String) o;
+ final MembershipManager membershipManager = new MembershipManager();
+ membershipManager.setDomain(domain.getBytes());
+ GroupManagementAgent groupMgtAgent = groupManagementAgents.get(domain);
+ membershipManager.setGroupManagementAgent(groupMgtAgent);
+ if(groupMgtAgent instanceof DefaultGroupManagementAgent) {
+ ((DefaultGroupManagementAgent) groupMgtAgent).setMembershipManager(membershipManager);
+ }
+
+ MembershipListener membershipListener = new MembershipListener() {
+ public void memberAdded(org.apache.catalina.tribes.Member member) {
+ membershipManager.memberAdded(member);
+ }
+
+ public void memberDisappeared(org.apache.catalina.tribes.Member member) {
+ membershipManager.memberDisappeared(member);
+ }
+ };
+ channel.addMembershipListener(membershipListener);
+ membershipManagers.add(membershipManager);
+ }
+ }
+
+ public List<MembershipManager> getMembershipManagers() {
+ return membershipManagers;
+ }
+
+ public void notifyMemberJoin(final Member member) {
+
+ if (TribesUtil.isInDomain(member, clusterManagerDomain)) { // A peer load balancer has joined
+
+ // Notify all members in the LB group
+ primaryMembershipManager.sendMemberJoinedToAll(member);
+
+ // Send the MEMBER_LISTS of all the groups to the the new LB member
+ for (MembershipManager manager : membershipManagers) {
+ manager.sendMemberList(member);
+ }
+ } else { // An application member has joined.
+
+ // Need to notify all members in the group of the new app member
+ Thread th = new Thread() {
+ public void run() {
+ for (MembershipManager manager : membershipManagers) {
+ if (TribesUtil.isInDomain(member, manager.getDomain())) {
+
+ // Send MEMBER_JOINED to the group of the new member
+ manager.sendMemberJoinedToAll(member);
+
+ // Send MEMBER_JOINED to the load balancer group
+ sendMemberJoinedToLoadBalancerGroup(manager.getRpcMembershipChannel(),
+ member);
+ break;
+ }
+ }
+ }
+
+ /**
+ * Send MEMBER_JOINED to the load balancer group
+ * @param rpcChannel The RpcChannel corresponding to the member's group
+ * @param member The member who joined
+ */
+ private void sendMemberJoinedToLoadBalancerGroup(RpcChannel rpcChannel,
+ Member member) {
+ MemberJoinedCommand cmd = new MemberJoinedCommand();
+ cmd.setMember(member);
+ try {
+ rpcChannel.send(primaryMembershipManager.getMembers(),
+ cmd,
+ RpcChannel.ALL_REPLY,
+ Channel.SEND_OPTIONS_ASYNCHRONOUS,
+ 10000);
+ } catch (ChannelException e) {
+ String errMsg = "Could not send MEMBER_JOINED[" +
+ TribesUtil.getName(member) +
+ "] to all load balancer members ";
+ log.error(errMsg, e);
+ throw new RemoteProcessException(errMsg, e);
+ }
+ }
+ };
+ th.start();
+ }
+ }
+}
Modified: webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/MembershipManager.java
URL: http://svn.apache.org/viewvc/webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/MembershipManager.java?rev=744900&r1=744899&r2=744900&view=diff
==============================================================================
--- webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/MembershipManager.java (original)
+++ webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/MembershipManager.java Mon Feb 16 12:02:42 2009
@@ -20,7 +20,7 @@
package org.apache.axis2.clustering.tribes;
import org.apache.axis2.clustering.ClusteringConstants;
-import org.apache.axis2.clustering.LoadBalanceEventHandler;
+import org.apache.axis2.clustering.management.GroupManagementAgent;
import org.apache.axis2.clustering.control.wka.MemberJoinedCommand;
import org.apache.axis2.clustering.control.wka.MemberListCommand;
import org.apache.axis2.context.ConfigurationContext;
@@ -52,7 +52,7 @@
* The domain corresponding to the membership handled by this MembershipManager
*/
private byte[] domain;
- private LoadBalanceEventHandler loadBalanceEventHandler;
+ private GroupManagementAgent groupManagementAgent;
private ConfigurationContext configContext;
public MembershipManager(ConfigurationContext configContext) {
@@ -75,8 +75,8 @@
this.staticMembershipInterceptor = staticMembershipInterceptor;
}
- public void setLoadBalanceEventHandler(LoadBalanceEventHandler loadBalanceEventHandler) {
- this.loadBalanceEventHandler = loadBalanceEventHandler;
+ public void setGroupManagementAgent(GroupManagementAgent groupManagementAgent) {
+ this.groupManagementAgent = groupManagementAgent;
}
public void setDomain(byte[] domain) {
@@ -149,10 +149,10 @@
TribesUtil.areInSameDomain(localMember, member);
// If this member is a load balancer, notify the respective load balance event handler?
- if (loadBalanceEventHandler != null) {
+ if (groupManagementAgent != null) {
log.info("Application member " + TribesUtil.getName(member) + " joined group " +
new String(member.getDomain()));
- loadBalanceEventHandler.applicationMemberAdded(TribesUtil.toAxis2Member(member));
+ groupManagementAgent.applicationMemberAdded(TribesUtil.toAxis2Member(member));
}
if (shouldAddMember) {
@@ -279,8 +279,8 @@
members.remove(member);
// Is this an application domain member?
- if (loadBalanceEventHandler != null) {
- loadBalanceEventHandler.applicationMemberRemoved(TribesUtil.toAxis2Member(member));
+ if (groupManagementAgent != null) {
+ groupManagementAgent.applicationMemberRemoved(TribesUtil.toAxis2Member(member));
}
}
Added: webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/TribesClusteringAgent.java
URL: http://svn.apache.org/viewvc/webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/TribesClusteringAgent.java?rev=744900&view=auto
==============================================================================
--- webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/TribesClusteringAgent.java (added)
+++ webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/TribesClusteringAgent.java Mon Feb 16 12:02:42 2009
@@ -0,0 +1,746 @@
+/*
+ * 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.axis2.clustering.tribes;
+
+import org.apache.axiom.om.OMAttribute;
+import org.apache.axiom.om.OMElement;
+import org.apache.axis2.AxisFault;
+import org.apache.axis2.clustering.ClusteringAgent;
+import org.apache.axis2.clustering.ClusteringConstants;
+import org.apache.axis2.clustering.ClusteringFault;
+import org.apache.axis2.clustering.management.GroupManagementAgent;
+import org.apache.axis2.clustering.MembershipListener;
+import org.apache.axis2.clustering.MembershipScheme;
+import org.apache.axis2.clustering.RequestBlockingHandler;
+import org.apache.axis2.clustering.management.NodeManager;
+import org.apache.axis2.clustering.management.DefaultNodeManager;
+import org.apache.axis2.clustering.management.DefaultGroupManagementAgent;
+import org.apache.axis2.clustering.state.ClusteringContextListener;
+import org.apache.axis2.clustering.state.StateManager;
+import org.apache.axis2.clustering.state.DefaultStateManager;
+import org.apache.axis2.clustering.control.ControlCommand;
+import org.apache.axis2.clustering.control.GetConfigurationCommand;
+import org.apache.axis2.clustering.control.GetStateCommand;
+import org.apache.axis2.context.ConfigurationContext;
+import org.apache.axis2.description.HandlerDescription;
+import org.apache.axis2.description.Parameter;
+import org.apache.axis2.description.PhaseRule;
+import org.apache.axis2.description.TransportInDescription;
+import org.apache.axis2.engine.AxisConfiguration;
+import org.apache.axis2.engine.DispatchPhase;
+import org.apache.axis2.engine.Phase;
+import org.apache.catalina.tribes.Channel;
+import org.apache.catalina.tribes.ChannelException;
+import org.apache.catalina.tribes.ManagedChannel;
+import org.apache.catalina.tribes.Member;
+import org.apache.catalina.tribes.group.GroupChannel;
+import org.apache.catalina.tribes.group.Response;
+import org.apache.catalina.tribes.group.RpcChannel;
+import org.apache.catalina.tribes.transport.MultiPointSender;
+import org.apache.catalina.tribes.transport.ReplicationTransmitter;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import javax.xml.namespace.QName;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.lang.reflect.Method;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+
+/**
+ * The main ClusteringAgent class for the Tribes based clustering implementation
+ */
+public class TribesClusteringAgent implements ClusteringAgent {
+
+ private static final Log log = LogFactory.getLog(TribesClusteringAgent.class);
+
+ private DefaultNodeManager configurationManager;
+ private DefaultStateManager contextManager;
+
+ private HashMap<String, Parameter> parameters;
+ private ManagedChannel channel;
+ private RpcChannel rpcInitChannel;
+ private ConfigurationContext configurationContext;
+ private ChannelListener channelListener;
+ private ChannelSender channelSender;
+ private MembershipManager primaryMembershipManager;
+ private RpcInitializationRequestHandler rpcInitRequestHandler;
+ private MembershipScheme membershipScheme;
+
+ /**
+ * The mode in which this member operates such as "loadBalance" or "application"
+ */
+ private OperationMode mode;
+
+ /**
+ * Static members
+ */
+ private List<org.apache.axis2.clustering.Member> members;
+
+ private final Map<String, GroupManagementAgent> groupManagementAgents =
+ new HashMap<String, GroupManagementAgent>();
+ private boolean clusterManagementMode;
+
+ public TribesClusteringAgent() {
+ parameters = new HashMap<String, Parameter>();
+ }
+
+ public void setMembers(List<org.apache.axis2.clustering.Member> members) {
+ this.members = members;
+ }
+
+ public List<org.apache.axis2.clustering.Member> getMembers() {
+ return members;
+ }
+
+ public void addGroupManagementAgent(GroupManagementAgent agent,
+ String applicationDomain) {
+ log.info("Managing group application domain " + applicationDomain +
+ " using agent " + agent.getClass());
+ groupManagementAgents.put(applicationDomain, agent);
+ clusterManagementMode = true;
+ }
+
+ public GroupManagementAgent getGroupManagementAgent(String applicationDomain) {
+ return groupManagementAgents.get(applicationDomain);
+ }
+
+ public Set<String> getDomains() {
+ return groupManagementAgents.keySet();
+ }
+
+ public StateManager getStateManager() {
+ return contextManager;
+ }
+
+ public NodeManager getNodeManager() {
+ return configurationManager;
+ }
+
+ /**
+ * Initialize the cluster.
+ *
+ * @throws ClusteringFault If initialization fails
+ */
+ public void init() throws ClusteringFault {
+ log.info("Initializing cluster...");
+ addRequestBlockingHandlerToInFlows();
+ primaryMembershipManager = new MembershipManager(configurationContext);
+
+ channel = new GroupChannel();
+ channelSender = new ChannelSender(channel, primaryMembershipManager, synchronizeAllMembers());
+ channelListener =
+ new ChannelListener(configurationContext, configurationManager, contextManager);
+ channel.addChannelListener(channelListener);
+
+ byte[] domain = getClusterDomain();
+ log.info("Cluster domain: " + new String(domain));
+ primaryMembershipManager.setDomain(domain);
+
+ // RpcChannel is a ChannelListener. When the reply to a particular request comes back, it
+ // picks it up. Each RPC is given a UUID, hence can correlate the request-response pair
+ rpcInitRequestHandler = new RpcInitializationRequestHandler(configurationContext);
+ rpcInitChannel =
+ new RpcChannel(TribesUtil.getRpcInitChannelId(domain),
+ channel, rpcInitRequestHandler);
+ if (log.isDebugEnabled()) {
+ log.debug("Created RPC Channel for domain " + new String(domain));
+ }
+
+ setMaximumRetries();
+ configureMode(domain);
+ configureMembershipScheme(domain, mode.getMembershipManagers());
+ setMemberInfo();
+
+ TribesMembershipListener membershipListener = new TribesMembershipListener(primaryMembershipManager);
+ channel.addMembershipListener(membershipListener);
+ try {
+ channel.start(Channel.DEFAULT); // At this point, this member joins the group
+ String localHost = TribesUtil.getLocalHost(channel);
+ if (localHost.startsWith("127.0.")) {
+ channel.stop(Channel.DEFAULT);
+ throw new ClusteringFault("Cannot join cluster using IP " + localHost +
+ ". Please set an IP address other than " +
+ localHost + " in the axis2.xml file");
+ }
+ } catch (ChannelException e) {
+ String msg = "Error starting Tribes channel";
+ log.error(msg, e);
+ throw new ClusteringFault(msg, e);
+ }
+
+ log.info("Local Member " + TribesUtil.getLocalHost(channel));
+ TribesUtil.printMembers(primaryMembershipManager);
+
+ membershipScheme.joinGroup();
+
+ // If configuration management is enabled, get the latest config from a neighbour
+ if (configurationManager != null) {
+ configurationManager.setSender(channelSender);
+ initializeSystem(new GetConfigurationCommand());
+ }
+
+ // If context replication is enabled, get the latest state from a neighbour
+ if (contextManager != null) {
+ contextManager.setSender(channelSender);
+ channelListener.setStateManager(contextManager);
+ initializeSystem(new GetStateCommand());
+ ClusteringContextListener contextListener = new ClusteringContextListener(channelSender);
+ configurationContext.addContextListener(contextListener);
+ }
+
+ configurationContext.
+ setNonReplicableProperty(ClusteringConstants.CLUSTER_INITIALIZED, "true");
+ log.info("Cluster initialization completed.");
+ }
+
+ public void finalize(){
+ if (channel != null){
+ log.info("Stopping Tribes channel...");
+ try {
+ channel.stop(Channel.DEFAULT);
+ } catch (ChannelException e) {
+ String msg = "Error occurred while stopping channel";
+ log.error(msg, e);
+ }
+ }
+ }
+
+ private void setMemberInfo() throws ClusteringFault {
+ Properties memberInfo = new Properties();
+ AxisConfiguration axisConfig = configurationContext.getAxisConfiguration();
+ TransportInDescription httpTransport = axisConfig.getTransportIn("http");
+ if (httpTransport != null) {
+ Parameter port = httpTransport.getParameter("port");
+ if (port != null) {
+ memberInfo.put("httpPort", port.getValue());
+ }
+ }
+ TransportInDescription httpsTransport = axisConfig.getTransportIn("https");
+ if (httpsTransport != null) {
+ Parameter port = httpsTransport.getParameter("port");
+ if (port != null) {
+ memberInfo.put("httpsPort", port.getValue());
+ }
+ }
+ Parameter isActiveParam = getParameter(ClusteringConstants.Parameters.IS_ACTIVE);
+ if (isActiveParam != null) {
+ memberInfo.setProperty(ClusteringConstants.Parameters.IS_ACTIVE,
+ (String) isActiveParam.getValue());
+ }
+
+ memberInfo.setProperty("hostName",
+ TribesUtil.getLocalHost(getParameter(TribesConstants.LOCAL_MEMBER_HOST)));
+
+ Parameter propsParam = getParameter("properties");
+ if(propsParam != null){
+ OMElement paramEle = propsParam.getParameterElement();
+ for(Iterator iter = paramEle.getChildrenWithLocalName("property"); iter.hasNext();){
+ OMElement propEle = (OMElement) iter.next();
+ OMAttribute nameAttrib = propEle.getAttribute(new QName("name"));
+ if(nameAttrib != null){
+ OMAttribute valueAttrib = propEle.getAttribute(new QName("value"));
+ if (valueAttrib != null) {
+ String attribVal = valueAttrib.getAttributeValue();
+ attribVal = replaceProperty(attribVal, memberInfo);
+ memberInfo.setProperty(nameAttrib.getAttributeValue(), attribVal);
+ }
+ }
+ }
+ }
+
+ memberInfo.remove("hostName"); // this was needed only to populate other properties. No need to send it.
+
+ ByteArrayOutputStream bout = new ByteArrayOutputStream();
+ try {
+ memberInfo.store(bout, "");
+ } catch (IOException e) {
+ String msg = "Cannot store member transport properties in the ByteArrayOutputStream";
+ log.error(msg, e);
+ throw new ClusteringFault(msg, e);
+ }
+ channel.getMembershipService().setPayload(bout.toByteArray());
+ }
+
+ private static String replaceProperty(String text, Properties props) {
+ int indexOfStartingChars = -1;
+ int indexOfClosingBrace;
+
+ // The following condition deals with properties.
+ // Properties are specified as ${system.property},
+ // and are assumed to be System properties
+ while (indexOfStartingChars < text.indexOf("${") &&
+ (indexOfStartingChars = text.indexOf("${")) != -1 &&
+ (indexOfClosingBrace = text.indexOf("}")) != -1) { // Is a property used?
+ String sysProp = text.substring(indexOfStartingChars + 2,
+ indexOfClosingBrace);
+ String propValue = props.getProperty(sysProp);
+ if (propValue != null) {
+ text = text.substring(0, indexOfStartingChars) + propValue +
+ text.substring(indexOfClosingBrace + 1);
+ }
+ }
+ return text;
+ }
+
+ /**
+ * Get the membership scheme applicable to this cluster
+ *
+ * @return The membership scheme. Only "wka" & "multicast" are valid return values.
+ * @throws ClusteringFault If the membershipScheme specified in the axis2.xml file is invalid
+ */
+ private String getMembershipScheme() throws ClusteringFault {
+ Parameter membershipSchemeParam =
+ getParameter(ClusteringConstants.Parameters.MEMBERSHIP_SCHEME);
+ String mbrScheme = ClusteringConstants.MembershipScheme.MULTICAST_BASED;
+ if (membershipSchemeParam != null) {
+ mbrScheme = ((String) membershipSchemeParam.getValue()).trim();
+ }
+ if (!mbrScheme.equals(ClusteringConstants.MembershipScheme.MULTICAST_BASED) &&
+ !mbrScheme.equals(ClusteringConstants.MembershipScheme.WKA_BASED)) {
+ String msg = "Invalid membership scheme '" + mbrScheme + "'. Supported schemes are " +
+ ClusteringConstants.MembershipScheme.MULTICAST_BASED + " & " +
+ ClusteringConstants.MembershipScheme.WKA_BASED;
+ log.error(msg);
+ throw new ClusteringFault(msg);
+ }
+ return mbrScheme;
+ }
+
+ /**
+ * Get the clustering domain to which this node belongs to
+ *
+ * @return The clustering domain to which this node belongs to
+ */
+ private byte[] getClusterDomain() {
+ Parameter domainParam = getParameter(ClusteringConstants.Parameters.DOMAIN);
+ byte[] domain;
+ if (domainParam != null) {
+ domain = ((String) domainParam.getValue()).getBytes();
+ } else {
+ domain = ClusteringConstants.DEFAULT_DOMAIN.getBytes();
+ }
+ return domain;
+ }
+
+ /**
+ * Set the maximum number of retries, if message sending to a particular node fails
+ */
+ private void setMaximumRetries() {
+ Parameter maxRetriesParam = getParameter(TribesConstants.MAX_RETRIES);
+ int maxRetries = 10;
+ if (maxRetriesParam != null) {
+ maxRetries = Integer.parseInt((String) maxRetriesParam.getValue());
+ }
+ ReplicationTransmitter replicationTransmitter =
+ (ReplicationTransmitter) channel.getChannelSender();
+ MultiPointSender multiPointSender = replicationTransmitter.getTransport();
+ multiPointSender.setMaxRetryAttempts(maxRetries);
+ }
+
+ /**
+ * A RequestBlockingHandler, which is an implementation of
+ * {@link org.apache.axis2.engine.Handler} is added to the InFlow & InFaultFlow. This handler
+ * is used for rejecting Web service requests until this node has been initialized. This handler
+ * can also be used for rejecting requests when this node is reinitializing or is in an
+ * inconsistent state (which can happen when a configuration change is taking place).
+ */
+ private void addRequestBlockingHandlerToInFlows() {
+ AxisConfiguration axisConfig = configurationContext.getAxisConfiguration();
+ for (Object o : axisConfig.getInFlowPhases()) {
+ Phase phase = (Phase) o;
+ if (phase instanceof DispatchPhase) {
+ RequestBlockingHandler requestBlockingHandler = new RequestBlockingHandler();
+ if (!phase.getHandlers().contains(requestBlockingHandler)) {
+ PhaseRule rule = new PhaseRule("Dispatch");
+ rule.setAfter("SOAPMessageBodyBasedDispatcher");
+ rule.setBefore("InstanceDispatcher");
+ HandlerDescription handlerDesc = requestBlockingHandler.getHandlerDesc();
+ handlerDesc.setHandler(requestBlockingHandler);
+ handlerDesc.setName(ClusteringConstants.REQUEST_BLOCKING_HANDLER);
+ handlerDesc.setRules(rule);
+ phase.addHandler(requestBlockingHandler);
+
+ log.debug("Added " + ClusteringConstants.REQUEST_BLOCKING_HANDLER +
+ " between SOAPMessageBodyBasedDispatcher & InstanceDispatcher to InFlow");
+ break;
+ }
+ }
+ }
+ for (Object o : axisConfig.getInFaultFlowPhases()) {
+ Phase phase = (Phase) o;
+ if (phase instanceof DispatchPhase) {
+ RequestBlockingHandler requestBlockingHandler = new RequestBlockingHandler();
+ if (!phase.getHandlers().contains(requestBlockingHandler)) {
+ PhaseRule rule = new PhaseRule("Dispatch");
+ rule.setAfter("SOAPMessageBodyBasedDispatcher");
+ rule.setBefore("InstanceDispatcher");
+ HandlerDescription handlerDesc = requestBlockingHandler.getHandlerDesc();
+ handlerDesc.setHandler(requestBlockingHandler);
+ handlerDesc.setName(ClusteringConstants.REQUEST_BLOCKING_HANDLER);
+ handlerDesc.setRules(rule);
+ phase.addHandler(requestBlockingHandler);
+
+ log.debug("Added " + ClusteringConstants.REQUEST_BLOCKING_HANDLER +
+ " between SOAPMessageBodyBasedDispatcher & InstanceDispatcher to InFaultFlow");
+ break;
+ }
+ }
+ }
+ }
+
+ private void configureMode(byte[] domain) {
+ if (clusterManagementMode) {
+ mode = new ClusterManagementMode(domain, groupManagementAgents, primaryMembershipManager);
+ for (GroupManagementAgent agent : groupManagementAgents.values()) {
+
+ if (agent instanceof DefaultGroupManagementAgent) {
+ ((DefaultGroupManagementAgent) agent).setSender(channelSender);
+ }
+ }
+ } else {
+ mode = new ApplicationMode(domain, primaryMembershipManager);
+ }
+ mode.init(channel);
+ }
+
+ /**
+ * Handle specific configurations related to different membership management schemes.
+ *
+ * @param localDomain The clustering loadBalancerDomain to which this member belongs to
+ * @param membershipManagers MembershipManagers for different domains
+ * @throws ClusteringFault If the membership scheme is invalid, or if an error occurs
+ * while configuring membership scheme
+ */
+ private void configureMembershipScheme(byte[] localDomain,
+ List<MembershipManager> membershipManagers)
+ throws ClusteringFault {
+ MembershipListener membershipListener;
+ Parameter parameter = getParameter(ClusteringConstants.Parameters.MEMBERSHIP_LISTENER);
+ if (parameter != null) {
+ OMElement paramEle = parameter.getParameterElement();
+ String clazz =
+ paramEle.getFirstChildWithName(new QName("class")).getText().trim();
+ try {
+ membershipListener = (MembershipListener) Class.forName(clazz).newInstance();
+ } catch (Exception e) {
+ String msg = "Cannot instantiate MembershipListener " + clazz;
+ log.error(msg, e);
+ throw new ClusteringFault(msg, e);
+ }
+ OMElement propsEle = paramEle.getFirstChildWithName(new QName("properties"));
+ if (propsEle != null) {
+ for (Iterator iter = propsEle.getChildElements(); iter.hasNext();) {
+ OMElement propEle = (OMElement) iter.next();
+ OMAttribute nameAttrib = propEle.getAttribute(new QName("name"));
+ if (nameAttrib != null) {
+ String name = nameAttrib.getAttributeValue();
+ setInstanceProperty(name, propEle.getText().trim(), membershipListener);
+ }
+ }
+ }
+ }
+
+ String scheme = getMembershipScheme();
+ log.info("Using " + scheme + " based membership management scheme");
+ if (scheme.equals(ClusteringConstants.MembershipScheme.WKA_BASED)) {
+ membershipScheme =
+ new WkaBasedMembershipScheme(channel, mode,
+ membershipManagers,
+ primaryMembershipManager,
+ parameters, localDomain, members,
+ getBooleanParam(ClusteringConstants.Parameters.ATMOST_ONCE_MSG_SEMANTICS),
+ getBooleanParam(ClusteringConstants.Parameters.PRESERVE_MSG_ORDER));
+ } else if (scheme.equals(ClusteringConstants.MembershipScheme.MULTICAST_BASED)) {
+ membershipScheme =
+ new MulticastBasedMembershipScheme(channel, mode, parameters,
+ localDomain,
+ getBooleanParam(ClusteringConstants.Parameters.ATMOST_ONCE_MSG_SEMANTICS),
+ getBooleanParam(ClusteringConstants.Parameters.PRESERVE_MSG_ORDER));
+ } else {
+ String msg = "Invalid membership scheme '" + scheme +
+ "'. Supported schemes are multicast & wka";
+ log.error(msg);
+ throw new ClusteringFault(msg);
+ }
+ membershipScheme.init();
+ }
+
+ private boolean getBooleanParam(String name) {
+ boolean result = false;
+ Parameter parameter = getParameter(name);
+ if (parameter != null) {
+ Object value = parameter.getValue();
+ if (value != null) {
+ result = Boolean.valueOf(((String) value).trim());
+ }
+ }
+ return result;
+ }
+
+ /**
+ * Find and invoke the setter method with the name of form setXXX passing in the value given
+ * on the POJO object
+ *
+ * @param name name of the setter field
+ * @param val value to be set
+ * @param obj POJO instance
+ * @throws ClusteringFault If an error occurs while setting the property
+ */
+ private void setInstanceProperty(String name, Object val, Object obj) throws ClusteringFault {
+
+ String mName = "set" + Character.toUpperCase(name.charAt(0)) + name.substring(1);
+ Method method;
+ try {
+ Method[] methods = obj.getClass().getMethods();
+ boolean invoked = false;
+ for (Method method1 : methods) {
+ if (mName.equals(method1.getName())) {
+ Class[] params = method1.getParameterTypes();
+ if (params.length != 1) {
+ handleException("Did not find a setter method named : " + mName +
+ "() that takes a single String, int, long, float, double " +
+ "or boolean parameter");
+ } else if (val instanceof String) {
+ String value = (String) val;
+ if (params[0].equals(String.class)) {
+ method = obj.getClass().getMethod(mName, String.class);
+ method.invoke(obj, new String[]{value});
+ } else if (params[0].equals(int.class)) {
+ method = obj.getClass().getMethod(mName, int.class);
+ method.invoke(obj, new Integer[]{new Integer(value)});
+ } else if (params[0].equals(long.class)) {
+ method = obj.getClass().getMethod(mName, long.class);
+ method.invoke(obj, new Long[]{new Long(value)});
+ } else if (params[0].equals(float.class)) {
+ method = obj.getClass().getMethod(mName, float.class);
+ method.invoke(obj, new Float[]{new Float(value)});
+ } else if (params[0].equals(double.class)) {
+ method = obj.getClass().getMethod(mName, double.class);
+ method.invoke(obj, new Double[]{new Double(value)});
+ } else if (params[0].equals(boolean.class)) {
+ method = obj.getClass().getMethod(mName, boolean.class);
+ method.invoke(obj, new Boolean[]{Boolean.valueOf(value)});
+ } else {
+ handleException("Did not find a setter method named : " + mName +
+ "() that takes a single String, int, long, float, double " +
+ "or boolean parameter");
+ }
+ } else {
+ if (params[0].equals(OMElement.class)) {
+ method = obj.getClass().getMethod(mName, OMElement.class);
+ method.invoke(obj, new OMElement[]{(OMElement) val});
+ }
+ }
+ invoked = true;
+ }
+ }
+
+ if (!invoked) {
+ handleException("Did not find a setter method named : " + mName +
+ "() that takes a single String, int, long, float, double " +
+ "or boolean parameter");
+ }
+
+ } catch (Exception e) {
+ handleException("Error invoking setter method named : " + mName +
+ "() that takes a single String, int, long, float, double " +
+ "or boolean parameter", e);
+ }
+ }
+
+ private void handleException(String msg, Exception e) throws ClusteringFault {
+ log.error(msg, e);
+ throw new ClusteringFault(msg, e);
+ }
+
+ private void handleException(String msg) throws ClusteringFault {
+ log.error(msg);
+ throw new ClusteringFault(msg);
+ }
+
+ /**
+ * Get some information from a neighbour. This information will be used by this node to
+ * initialize itself
+ * <p/>
+ * rpcChannel is The utility for sending RPC style messages to the channel
+ *
+ * @param command The control command to send
+ * @throws ClusteringFault If initialization code failed on this node
+ */
+ private void initializeSystem(ControlCommand command) throws ClusteringFault {
+ // If there is at least one member in the cluster,
+ // get the current initialization info from a member
+ int numberOfTries = 0; // Don't keep on trying indefinitely
+
+ // Keep track of members to whom we already sent an initialization command
+ // Do not send another request to these members
+ List<String> sentMembersList = new ArrayList<String>();
+ sentMembersList.add(TribesUtil.getLocalHost(channel));
+ Member[] members = primaryMembershipManager.getMembers();
+ if (members.length == 0) {
+ return;
+ }
+
+ while (members.length > 0 && numberOfTries < 5) {
+ Member member = (numberOfTries == 0) ?
+ primaryMembershipManager.getLongestLivingMember() : // First try to get from the longest member alive
+ primaryMembershipManager.getRandomMember(); // Else get from a random member
+ String memberHost = TribesUtil.getName(member);
+ log.info("Trying to send intialization request to " + memberHost);
+ try {
+ if (!sentMembersList.contains(memberHost)) {
+ Response[] responses;
+// do {
+ responses = rpcInitChannel.send(new Member[]{member},
+ command,
+ RpcChannel.FIRST_REPLY,
+ Channel.SEND_OPTIONS_ASYNCHRONOUS,
+ 10000);
+ if (responses.length == 0) {
+ try {
+ Thread.sleep(500);
+ } catch (InterruptedException ignored) {
+ }
+ }
+ // TODO: If we do not get a response within some time, try to recover from this fault
+// }
+// while (responses.length == 0 || responses[0] == null || responses[0].getMessage() == null); // TODO: #### We will need to check this
+ if (responses.length != 0 && responses[0] != null && responses[0].getMessage() != null) {
+ ((ControlCommand) responses[0].getMessage()).execute(configurationContext); // Do the initialization
+ break;
+ }
+ }
+ } catch (Exception e) {
+ log.error("Cannot get initialization information from " +
+ memberHost + ". Will retry in 2 secs.", e);
+ sentMembersList.add(memberHost);
+ try {
+ Thread.sleep(2000);
+ } catch (InterruptedException ignored) {
+ log.debug("Interrupted", ignored);
+ }
+ }
+ numberOfTries++;
+ members = primaryMembershipManager.getMembers();
+ if (numberOfTries >= members.length) {
+ break;
+ }
+ }
+ }
+
+ public void setNodeManager(NodeManager nodeManager) {
+ this.configurationManager = (DefaultNodeManager) nodeManager;
+ this.configurationManager.setSender(channelSender);
+ }
+
+ public void setStateManager(StateManager stateManager) {
+ this.contextManager = (DefaultStateManager) stateManager;
+ this.contextManager.setSender(channelSender);
+ }
+
+ public void addParameter(Parameter param) throws AxisFault {
+ parameters.put(param.getName(), param);
+ }
+
+ public void deserializeParameters(OMElement parameterElement) throws AxisFault {
+ throw new UnsupportedOperationException();
+ }
+
+ public Parameter getParameter(String name) {
+ return parameters.get(name);
+ }
+
+ public ArrayList getParameters() {
+ ArrayList<Parameter> list = new ArrayList<Parameter>();
+ for (String msg : parameters.keySet()) {
+ list.add(parameters.get(msg));
+ }
+ return list;
+ }
+
+ public boolean isParameterLocked(String parameterName) {
+ Parameter parameter = parameters.get(parameterName);
+ return parameter != null && parameter.isLocked();
+ }
+
+ public void removeParameter(Parameter param) throws AxisFault {
+ parameters.remove(param.getName());
+ }
+
+ /**
+ * Shutdown the cluster. This member will leave the cluster when this method is called.
+ *
+ * @throws ClusteringFault If an error occurs while shutting down
+ */
+ public void shutdown() throws ClusteringFault {
+ log.debug("Enter: TribesClusteringAgent::shutdown");
+ if (channel != null) {
+ try {
+ channel.removeChannelListener(rpcInitChannel);
+ channel.removeChannelListener(channelListener);
+ channel.stop(Channel.DEFAULT);
+ } catch (ChannelException e) {
+
+ if (log.isDebugEnabled()) {
+ log.debug("Exit: TribesClusteringAgent::shutdown");
+ }
+
+ throw new ClusteringFault(e);
+ }
+ }
+ log.debug("Exit: TribesClusteringAgent::shutdown");
+ }
+
+ public void setConfigurationContext(ConfigurationContext configurationContext) {
+ this.configurationContext = configurationContext;
+ if (rpcInitRequestHandler != null) {
+ rpcInitRequestHandler.setConfigurationContext(configurationContext);
+ }
+ if (channelListener != null) {
+ channelListener.setConfigurationContext(configurationContext);
+ }
+ if (configurationManager != null) {
+ configurationManager.setConfigurationContext(configurationContext);
+ }
+ if (contextManager != null) {
+ contextManager.setConfigurationContext(configurationContext);
+ }
+ }
+
+ /**
+ * Method to check whether all members in the cluster have to be kept in sync at all times.
+ * Typically, this will require each member in the cluster to ACKnowledge receipt of a
+ * particular message, which may have a significant performance hit.
+ *
+ * @return true - if all members in the cluster should be kept in sync at all times, false
+ * otherwise
+ */
+ public boolean synchronizeAllMembers() {
+ Parameter syncAllParam = getParameter(ClusteringConstants.Parameters.SYNCHRONIZE_ALL_MEMBERS);
+ return syncAllParam == null || Boolean.parseBoolean((String) syncAllParam.getValue());
+ }
+}
Modified: webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/TribesMembershipListener.java
URL: http://svn.apache.org/viewvc/webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/TribesMembershipListener.java?rev=744900&r1=744899&r2=744900&view=diff
==============================================================================
--- webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/TribesMembershipListener.java (original)
+++ webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/TribesMembershipListener.java Mon Feb 16 12:02:42 2009
@@ -43,14 +43,14 @@
} else {
}*/
}
- // System.err.println("++++++ IS COORD="+TribesClusterManager.nbc.isCoordinator());
+ // System.err.println("++++++ IS COORD="+TribesClusteringAgent.nbc.isCoordinator());
}
public void memberDisappeared(Member member) {
log.info("Member " + TribesUtil.getName(member) + " left cluster");
membershipManager.memberDisappeared(member);
-// System.err.println("++++++ IS COORD="+TribesClusterManager.nbc.isCoordinator());
+// System.err.println("++++++ IS COORD="+TribesClusteringAgent.nbc.isCoordinator());
}
}
Modified: webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/TribesUtil.java
URL: http://svn.apache.org/viewvc/webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/TribesUtil.java?rev=744900&r1=744899&r2=744900&view=diff
==============================================================================
--- webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/TribesUtil.java (original)
+++ webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/TribesUtil.java Mon Feb 16 12:02:42 2009
@@ -146,4 +146,4 @@
}
return props;
}
-}
\ No newline at end of file
+}
Modified: webservices/axis2/trunk/java/modules/clustering/test/org/apache/axis2/clustering/ClusterManagerTestCase.java
URL: http://svn.apache.org/viewvc/webservices/axis2/trunk/java/modules/clustering/test/org/apache/axis2/clustering/ClusterManagerTestCase.java?rev=744900&r1=744899&r2=744900&view=diff
==============================================================================
--- webservices/axis2/trunk/java/modules/clustering/test/org/apache/axis2/clustering/ClusterManagerTestCase.java (original)
+++ webservices/axis2/trunk/java/modules/clustering/test/org/apache/axis2/clustering/ClusterManagerTestCase.java Mon Feb 16 12:02:42 2009
@@ -20,10 +20,6 @@
package org.apache.axis2.clustering;
import junit.framework.TestCase;
-import org.apache.axis2.clustering.configuration.ConfigurationManagerListener;
-import org.apache.axis2.clustering.configuration.DefaultConfigurationManagerListener;
-import org.apache.axis2.clustering.configuration.TestConfigurationManagerListener;
-import org.apache.axis2.clustering.context.DefaultContextManagerListener;
import org.apache.axis2.context.ConfigurationContext;
import org.apache.axis2.context.ConfigurationContextFactory;
import org.apache.axis2.description.AxisService;
@@ -33,25 +29,21 @@
public abstract class ClusterManagerTestCase extends TestCase {
- protected ClusterManager clusterManager1 = null;
- protected ClusterManager clusterManager2 = null;
+ protected ClusteringAgent clusterManager1 = null;
+ protected ClusteringAgent clusterManager2 = null;
protected AxisConfiguration axisConfiguration1 = null;
protected AxisConfiguration axisConfiguration2 = null;
protected ConfigurationContext configurationContext1 = null;
protected ConfigurationContext configurationContext2 = null;
- protected ConfigurationManagerListener configManagerListener1;
- protected ConfigurationManagerListener configManagerListener2;
protected AxisServiceGroup serviceGroup1 = null;
protected AxisServiceGroup serviceGroup2 = null;
protected AxisService service1 = null;
protected AxisService service2 = null;
protected String serviceName = "testService";
- protected abstract ClusterManager getClusterManager(ConfigurationContext configCtx);
+ protected abstract ClusteringAgent getClusterManager(ConfigurationContext configCtx);
protected boolean skipChannelTests = false;
- protected TestConfigurationManagerListener configurationManagerListener;
- protected DefaultContextManagerListener contextManagerListener;
protected void setUp() throws Exception {
@@ -63,16 +55,11 @@
clusterManager1 = getClusterManager(configurationContext1);
clusterManager2 = getClusterManager(configurationContext2);
- clusterManager1.getContextManager().setConfigurationContext(configurationContext1);
- clusterManager2.getContextManager().setConfigurationContext(configurationContext2);
+ clusterManager1.getStateManager().setConfigurationContext(configurationContext1);
+ clusterManager2.getStateManager().setConfigurationContext(configurationContext2);
- clusterManager1.getConfigurationManager().setConfigurationContext(configurationContext1);
- clusterManager2.getConfigurationManager().setConfigurationContext(configurationContext2);
-
- configManagerListener1 = new DefaultConfigurationManagerListener();
- clusterManager1.getConfigurationManager().setConfigurationManagerListener(configManagerListener1);
- configManagerListener2 = new DefaultConfigurationManagerListener();
- clusterManager2.getConfigurationManager().setConfigurationManagerListener(configManagerListener2);
+ clusterManager1.getNodeManager().setConfigurationContext(configurationContext1);
+ clusterManager2.getNodeManager().setConfigurationContext(configurationContext2);
//giving both Nodes the same deployment configuration
axisConfiguration1 = configurationContext1.getAxisConfiguration();
@@ -91,10 +78,10 @@
System.setProperty(ClusteringConstants.LOCAL_IP_ADDRESS, Utils.getIpAddress());
try {
clusterManager1.init();
- System.out.println("ClusterManager-1 successfully initialized");
+ System.out.println("ClusteringAgent-1 successfully initialized");
System.out.println("*** PLEASE IGNORE THE java.net.ConnectException STACKTRACES. THIS IS EXPECTED ***");
clusterManager2.init();
- System.out.println("ClusterManager-2 successfully initialized");
+ System.out.println("ClusteringAgent-2 successfully initialized");
} catch (ClusteringFault e) {
String message =
"Could not initialize ClusterManagers. Please check the network connection";
Modified: webservices/axis2/trunk/java/modules/clustering/test/org/apache/axis2/clustering/ContextReplicationTest.java
URL: http://svn.apache.org/viewvc/webservices/axis2/trunk/java/modules/clustering/test/org/apache/axis2/clustering/ContextReplicationTest.java?rev=744900&r1=744899&r2=744900&view=diff
==============================================================================
--- webservices/axis2/trunk/java/modules/clustering/test/org/apache/axis2/clustering/ContextReplicationTest.java (original)
+++ webservices/axis2/trunk/java/modules/clustering/test/org/apache/axis2/clustering/ContextReplicationTest.java Mon Feb 16 12:02:42 2009
@@ -23,13 +23,11 @@
import org.apache.axiom.om.util.UUIDGenerator;
import org.apache.axis2.AxisFault;
import org.apache.axis2.util.Utils;
-import org.apache.axis2.clustering.configuration.ConfigurationManager;
-import org.apache.axis2.clustering.configuration.DefaultConfigurationManager;
-import org.apache.axis2.clustering.configuration.DefaultConfigurationManagerListener;
-import org.apache.axis2.clustering.context.ContextManager;
-import org.apache.axis2.clustering.context.DefaultContextManager;
-import org.apache.axis2.clustering.context.DefaultContextManagerListener;
-import org.apache.axis2.clustering.tribes.TribesClusterManager;
+import org.apache.axis2.clustering.management.NodeManager;
+import org.apache.axis2.clustering.management.DefaultNodeManager;
+import org.apache.axis2.clustering.state.StateManager;
+import org.apache.axis2.clustering.state.DefaultStateManager;
+import org.apache.axis2.clustering.tribes.TribesClusteringAgent;
import org.apache.axis2.context.ConfigurationContext;
import org.apache.axis2.context.ConfigurationContextFactory;
import org.apache.axis2.context.ServiceContext;
@@ -61,18 +59,18 @@
"axis2.domain." + UUIDGenerator.getUUID());
// --------------- Cluster-1 ------------------------------------------------------
- private ClusterManager clusterManager1;
- private ContextManager ctxMan1;
- private ConfigurationManager configMan1;
+ private ClusteringAgent clusterManager1;
+ private StateManager ctxMan1;
+ private NodeManager configMan1;
private ConfigurationContext configurationContext1;
private AxisServiceGroup serviceGroup1;
private AxisService service1;
//---------------------------------------------------------------------------------
// --------------- Cluster-2 ------------------------------------------------------
- private ClusterManager clusterManager2;
- private ContextManager ctxMan2;
- private ConfigurationManager configMan2;
+ private ClusteringAgent clusterManager2;
+ private StateManager ctxMan2;
+ private NodeManager configMan2;
private ConfigurationContext configurationContext2;
private AxisServiceGroup serviceGroup2;
private AxisService service2;
@@ -202,7 +200,7 @@
clusterManager1 = getClusterManager(configurationContext1, ctxMan1, configMan1);
clusterManager1.addParameter(domainParam);
clusterManager1.init();
- System.out.println("---------- ClusterManager-1 successfully initialized -----------");
+ System.out.println("---------- ClusteringAgent-1 successfully initialized -----------");
// Second cluster
configurationContext2 =
@@ -214,26 +212,20 @@
clusterManager2 = getClusterManager(configurationContext2, ctxMan2, configMan2);
clusterManager2.addParameter(domainParam);
clusterManager2.init();
- System.out.println("---------- ClusterManager-2 successfully initialized -----------");
+ System.out.println("---------- ClusteringAgent-2 successfully initialized -----------");
}
- protected ClusterManager getClusterManager(ConfigurationContext configCtx,
- ContextManager contextManager,
- ConfigurationManager configManager)
+ protected ClusteringAgent getClusterManager(ConfigurationContext configCtx,
+ StateManager stateManager,
+ NodeManager configManager)
throws AxisFault {
- ClusterManager clusterManager = new TribesClusterManager();
- configCtx.getAxisConfiguration().setClusterManager(clusterManager);
+ ClusteringAgent clusteringAgent = new TribesClusteringAgent();
+ configCtx.getAxisConfiguration().setClusteringAgent(clusteringAgent);
+ clusteringAgent.setNodeManager(configManager);
+ clusteringAgent.setStateManager(stateManager);
+ clusteringAgent.setConfigurationContext(configCtx);
- configManager.
- setConfigurationManagerListener(new DefaultConfigurationManagerListener());
- clusterManager.setConfigurationManager(configManager);
-
- contextManager.setContextManagerListener(new DefaultContextManagerListener());
- clusterManager.setContextManager(contextManager);
-
- clusterManager.setConfigurationContext(configCtx);
-
- return clusterManager;
+ return clusteringAgent;
}
protected AxisServiceGroup createAxisServiceGroup(ConfigurationContext configCtx)
@@ -250,15 +242,13 @@
return service;
}
- protected ContextManager getContextManager() throws AxisFault {
- ContextManager contextManager = new DefaultContextManager();
- contextManager.setContextManagerListener(new DefaultContextManagerListener());
- return contextManager;
+ protected StateManager getContextManager() throws AxisFault {
+ StateManager stateManager = new DefaultStateManager();
+ return stateManager;
}
- protected ConfigurationManager getConfigurationManager() throws AxisFault {
- ConfigurationManager contextManager = new DefaultConfigurationManager();
- contextManager.setConfigurationManagerListener(new DefaultConfigurationManagerListener());
+ protected NodeManager getConfigurationManager() throws AxisFault {
+ NodeManager contextManager = new DefaultNodeManager();
return contextManager;
}
Added: webservices/axis2/trunk/java/modules/clustering/test/org/apache/axis2/clustering/management/ConfigurationManagerTestCase.java
URL: http://svn.apache.org/viewvc/webservices/axis2/trunk/java/modules/clustering/test/org/apache/axis2/clustering/management/ConfigurationManagerTestCase.java?rev=744900&view=auto
==============================================================================
--- webservices/axis2/trunk/java/modules/clustering/test/org/apache/axis2/clustering/management/ConfigurationManagerTestCase.java (added)
+++ webservices/axis2/trunk/java/modules/clustering/test/org/apache/axis2/clustering/management/ConfigurationManagerTestCase.java Mon Feb 16 12:02:42 2009
@@ -0,0 +1,192 @@
+/*
+ * 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.axis2.clustering.management;
+
+import org.apache.axis2.clustering.ClusterManagerTestCase;
+import org.apache.axis2.clustering.ClusteringFault;
+
+import javax.xml.stream.XMLStreamException;
+
+
+public abstract class ConfigurationManagerTestCase extends ClusterManagerTestCase {
+
+ public void testLoadServiceGroup() throws ClusteringFault {
+
+ String serviceGroupName = "testService";
+// clusterManager1.getNodeManager().loadServiceGroups(new String[]{serviceGroupName});
+
+ /*try {
+ Thread.sleep(3000);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+
+ List eventList = configurationManagerListener2.getEventList();
+ assertNotNull(eventList);
+ assertEquals(eventList.size(), 1);
+ ConfigurationEvent event = (ConfigurationEvent) eventList.get(0);
+
+ assertNotNull(event);
+
+ String[] serviceGroupNames = event.getServiceGroupNames();
+ assertNotNull(serviceGroupNames);
+ assertEquals(serviceGroupNames[0], serviceGroupName);*/
+ }
+
+ public void testUnloadServiceGroup() throws ClusteringFault {
+
+ String serviceGroupName = "testService";
+// clusterManager1.getNodeManager().unloadServiceGroups(new String[]{serviceGroupName});
+
+ /*try {
+ Thread.sleep(3000);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+
+ List eventList = configurationManagerListener2.getEventList();
+ assertNotNull(eventList);
+ assertEquals(eventList.size(), 1);
+ ConfigurationEvent event = (ConfigurationEvent) eventList.get(0);
+
+ assertNotNull(event);
+
+ String[] serviceGroupNames = event.getServiceGroupNames();
+ assertNotNull(serviceGroupNames);
+ assertEquals(serviceGroupNames[0], serviceGroupName);*/
+ }
+
+ public void testApplyPolicy() throws ClusteringFault, XMLStreamException {
+
+ String serviceGroupName = "testService";
+// clusterManager1.getNodeManager().loadServiceGroups(new String[]{serviceGroupName});
+ String policyID = "policy1";
+
+ /*try {
+ Thread.sleep(3000);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+
+ Policy policy = new Policy();
+ policy.setId(policyID);
+
+ StringWriter writer = new StringWriter();
+ XMLStreamWriter xmlStreamWriter = XMLOutputFactory.newInstance()
+ .createXMLStreamWriter(writer);
+
+ policy.serialize(xmlStreamWriter);
+ xmlStreamWriter.flush();
+
+ clusterManager1.getConfigurationManager().applyPolicy(serviceGroupName, writer.toString());
+
+ try {
+ Thread.sleep(3000);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+
+ List eventList = configurationManagerListener2.getEventList();
+ assertNotNull(eventList);
+ assertEquals(eventList.size(), 2);
+ ConfigurationEvent event = (ConfigurationEvent) eventList.get(1);
+ assertNotNull(event);
+ assertEquals(event.getServiceName(), serviceName);
+ assertNotNull(event.getPolicy());*/
+
+ }
+
+ public void testPrepare() throws ClusteringFault {
+
+ String serviceGroupName = "testService";
+ clusterManager1.getNodeManager().prepare();
+
+ /*try {
+ Thread.sleep(3000);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+
+ List eventList = configurationManagerListener2.getEventList();
+ assertNotNull(eventList);
+ assertEquals(eventList.size(), 1);
+ ConfigurationEvent event = (ConfigurationEvent) eventList.get(0);
+
+ assertNotNull(event);*/
+ }
+
+ public void testCommit() throws ClusteringFault {
+
+ String serviceGroupName = "testService";
+ clusterManager1.getNodeManager().commit();
+
+ /*try {
+ Thread.sleep(3000);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+
+ List eventList = configurationManagerListener2.getEventList();
+ assertNotNull(eventList);
+ assertEquals(eventList.size(), 1);
+ ConfigurationEvent event = (ConfigurationEvent) eventList.get(0);
+
+ assertNotNull(event);*/
+ }
+
+ public void testRollback() throws ClusteringFault {
+
+ String serviceGroupName = "testService";
+ clusterManager1.getNodeManager().rollback();
+
+ /*try {
+ Thread.sleep(3000);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+
+ List eventList = configurationManagerListener2.getEventList();
+ assertNotNull(eventList);
+ assertEquals(eventList.size(), 1);
+ ConfigurationEvent event = (ConfigurationEvent) eventList.get(0);
+
+ assertNotNull(event);*/
+ }
+
+ public void testReloadConfiguration() throws ClusteringFault {
+
+ String serviceGroupName = "testService";
+// clusterManager1.getNodeManager().reloadConfiguration();
+
+ try {
+ Thread.sleep(3000);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+
+ /*List eventList = configurationManagerListener2.getEventList();
+ assertNotNull(eventList);
+ assertEquals(eventList.size(), 1);
+ ConfigurationEvent event = (ConfigurationEvent) eventList.get(0);
+
+ assertNotNull(event);*/
+ }
+
+}
Modified: webservices/axis2/trunk/java/modules/clustering/test/org/apache/axis2/clustering/tribes/ConfigurationManagerTest.java
URL: http://svn.apache.org/viewvc/webservices/axis2/trunk/java/modules/clustering/test/org/apache/axis2/clustering/tribes/ConfigurationManagerTest.java?rev=744900&r1=744899&r2=744900&view=diff
==============================================================================
--- webservices/axis2/trunk/java/modules/clustering/test/org/apache/axis2/clustering/tribes/ConfigurationManagerTest.java (original)
+++ webservices/axis2/trunk/java/modules/clustering/test/org/apache/axis2/clustering/tribes/ConfigurationManagerTest.java Mon Feb 16 12:02:42 2009
@@ -19,26 +19,21 @@
package org.apache.axis2.clustering.tribes;
-import org.apache.axis2.clustering.ClusterManager;
-import org.apache.axis2.clustering.configuration.DefaultConfigurationManager;
-import org.apache.axis2.clustering.configuration.DefaultConfigurationManagerListener;
-import org.apache.axis2.clustering.context.DefaultContextManager;
-import org.apache.axis2.clustering.context.DefaultContextManagerListener;
+import org.apache.axis2.clustering.ClusteringAgent;
+import org.apache.axis2.clustering.management.DefaultNodeManager;
+import org.apache.axis2.clustering.state.DefaultStateManager;
import org.apache.axis2.context.ConfigurationContext;
public class ConfigurationManagerTest extends
- org.apache.axis2.clustering.configuration.ConfigurationManagerTestCase {
+ org.apache.axis2.clustering.management.ConfigurationManagerTestCase {
- protected ClusterManager getClusterManager(ConfigurationContext configCtx) {
- TribesClusterManager tribesClusterManager = new TribesClusterManager();
+ protected ClusteringAgent getClusterManager(ConfigurationContext configCtx) {
+ TribesClusteringAgent tribesClusterManager = new TribesClusteringAgent();
tribesClusterManager.setConfigurationContext(configCtx);
- DefaultConfigurationManager configurationManager = new DefaultConfigurationManager();
- configurationManager.
- setConfigurationManagerListener(new DefaultConfigurationManagerListener());
- tribesClusterManager.setConfigurationManager(configurationManager);
- DefaultContextManager contextManager = new DefaultContextManager();
- contextManager.setContextManagerListener(new DefaultContextManagerListener());
- tribesClusterManager.setContextManager(contextManager);
+ DefaultNodeManager configurationManager = new DefaultNodeManager();
+ tribesClusterManager.setNodeManager(configurationManager);
+ DefaultStateManager contextManager = new DefaultStateManager();
+ tribesClusterManager.setStateManager(contextManager);
return tribesClusterManager;
}
Modified: webservices/axis2/trunk/java/modules/kernel/conf/axis2.xml
URL: http://svn.apache.org/viewvc/webservices/axis2/trunk/java/modules/kernel/conf/axis2.xml?rev=744900&r1=744899&r2=744900&view=diff
==============================================================================
--- webservices/axis2/trunk/java/modules/kernel/conf/axis2.xml (original)
+++ webservices/axis2/trunk/java/modules/kernel/conf/axis2.xml Mon Feb 16 12:02:42 2009
@@ -251,40 +251,21 @@
<!-- Clustering -->
<!-- ================================================= -->
<!--
- To enable clustering for this node, set the value of "enable" attribute of the "cluster"
- element to "true". The initilization of a node in the cluster is handled by the class
- corresponding to the "class" attribute of the "cluster" element. It is also responsible for
+ To enable clustering for this node, set the value of "enable" attribute of the "clustering"
+ element to "true". The initialization of a node in the cluster is handled by the class
+ corresponding to the "class" attribute of the "clustering" element. It is also responsible for
getting this node to join the cluster.
-->
- <cluster class="org.apache.axis2.clustering.tribes.TribesClusterManager" enable="false">
+ <clustering class="org.apache.axis2.clustering.tribes.TribesClusteringAgent" enable="false">
<!--
- This parameter indicates whther the cluster has to be automatically initalized
+ This parameter indicates whether the cluster has to be automatically initalized
when the AxisConfiguration is built. If set to "true" the initialization will not be
done at that stage, and some other party will have to explictly initialize the cluster.
-->
<parameter name="AvoidInitiation">true</parameter>
<!--
- Indicates whether this member is active or not. This will be useful in a setup where
- members are deployed in ACTIVE and/or PASSIVE configuration
- -->
- <parameter name="isActive">true</parameter>
-
- <!--
- An implementation of org.apache.axis2.clustering.MembershipListener
- which will be called on membership changes
- -->
- <!--
- <parameter name="membershipListener">
- <class>org.apache.axis2.clustering.MembershipListenerImpl</class>
- <properties>
- <property name="foo">bar</property>
- </properties>
- </parameter>
- -->
-
- <!--
The membership scheme used in this setup. The only values supported at the moment are
"multicast" and "wka"
@@ -303,13 +284,7 @@
The clustering domain/group. Nodes in the same group will belong to the same multicast
domain. There will not be interference between nodes in different groups.
-->
- <parameter name="domain">apache.axis2.domain</parameter>
-
- <!--
- This parameter is only valid when the "mode" parameter is set to application. This indicates
- the domain in which the the applications being load balanced are deployed.
- -->
- <parameter name="applicationDomain">apache.axis2.application.domain</parameter>
+ <parameter name="domain">wso2.carbon.domain</parameter>
<!--
When a Web service request is received, and processed, before the response is sent to the
@@ -367,6 +342,14 @@
<parameter name="atmostOnceMessageSemantics">true</parameter>
<!--
+ Properties specific to this member
+ -->
+ <parameter name="properties">
+ <property name="backendServerURL" value="https://${hostName}:${httpsPort}/services/"/>
+ <property name="mgtConsoleURL" value="https://${hostName}:${httpsPort}/"/>
+ </parameter>
+
+ <!--
The list of static or well-known members. These entries will only be valid if the
"membershipScheme" above is set to "wka"
-->
@@ -382,35 +365,25 @@
</members>
<!--
- Enable the load balancer entry if you need to run this node as a load balancer.
- Multiple application domains with different LoadBalanceEventHandler implementations
+ Enable the groupManagement entry if you need to run this node as a cluster manager.
+ Multiple application domains with different GroupManagementAgent implementations
can be defined in this section.
-->
- <loadBalancer enable="false">
+ <groupManagement enable="false">
<applicationDomain name="apache.axis2.application.domain"
- handler="org.apache.axis2.clustering.DefaultLoadBalanceEventHandler"/>
- </loadBalancer>
+ description="Axis2 group"
+ agent="org.apache.axis2.clustering.management.DefaultGroupManagementAgent"/>
+ </groupManagement>
<!--
- This interface is responsible for handling configuration management.
- Configuraion changes include:
-
- 1. Rebooting an entire cluster, in which case, all nodes have to load the new
- Axis2 configuration in a consistent manner
- 2. Deploying a new service to a cluster or undeploying a service from a cluster
- 3. Changing the policies of a service deployed on the cluster
-
- The "enable" attribute indicates whether Configuration management has been enabled
+ This interface is responsible for handling management of a specific node in the cluster
+ The "enable" attribute indicates whether Node management has been enabled
-->
- <configurationManager
- class="org.apache.axis2.clustering.configuration.DefaultConfigurationManager"
- enable="true">
- <listener
- class="org.apache.axis2.clustering.configuration.DefaultConfigurationManagerListener"/>
- </configurationManager>
+ <nodeManager class="org.apache.axis2.clustering.management.DefaultNodeManager"
+ enable="true"/>
<!--
- This interface is responsible for handling context replication. The property changes in
+ This interface is responsible for handling state replication. The property changes in
the Axis2 context hierarchy in this node, are propagated to all other nodes in the cluster.
The "excludes" patterns can be used to specify the prefixes (e.g. local_*) or
@@ -419,27 +392,25 @@
The "enable" attribute indicates whether context replication has been enabled
-->
- <contextManager class="org.apache.axis2.clustering.context.DefaultContextManager"
- enable="true">
- <listener class="org.apache.axis2.clustering.context.DefaultContextManagerListener"/>
+ <stateManager class="org.apache.axis2.clustering.state.DefaultStateManager"
+ enable="true">
<replication>
<defaults>
<exclude name="local_*"/>
<exclude name="LOCAL_*"/>
</defaults>
<context class="org.apache.axis2.context.ConfigurationContext">
- <exclude name="UseAsyncOperations"/>
- <exclude name="SequencePropertyBeanMap"/>
+ <exclude name="local_*"/>
</context>
<context class="org.apache.axis2.context.ServiceGroupContext">
- <exclude name="my.sandesha.*"/>
+ <exclude name="local_*"/>
</context>
<context class="org.apache.axis2.context.ServiceContext">
- <exclude name="my.sandesha.*"/>
+ <exclude name="local_*"/>
</context>
</replication>
- </contextManager>
- </cluster>
+ </stateManager>
+ </clustering>
<!-- ================================================= -->
<!-- Phases -->
Added: webservices/axis2/trunk/java/modules/kernel/src/org/apache/axis2/clustering/ClusteringAgent.java
URL: http://svn.apache.org/viewvc/webservices/axis2/trunk/java/modules/kernel/src/org/apache/axis2/clustering/ClusteringAgent.java?rev=744900&view=auto
==============================================================================
--- webservices/axis2/trunk/java/modules/kernel/src/org/apache/axis2/clustering/ClusteringAgent.java (added)
+++ webservices/axis2/trunk/java/modules/kernel/src/org/apache/axis2/clustering/ClusteringAgent.java Mon Feb 16 12:02:42 2009
@@ -0,0 +1,187 @@
+/*
+ * 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.axis2.clustering;
+
+import org.apache.axis2.clustering.management.NodeManager;
+import org.apache.axis2.clustering.management.GroupManagementAgent;
+import org.apache.axis2.clustering.state.StateManager;
+import org.apache.axis2.context.ConfigurationContext;
+import org.apache.axis2.description.ParameterInclude;
+
+import java.util.List;
+import java.util.Set;
+
+/**
+ * <p>
+ * This is the main interface in the Axis2 clustering implementation.
+ * In order to plug-in a new clustering implementation, this interface has to be
+ * implemented.
+ * </p>
+ * <p>
+ * The initilization of a node in the cluster is handled here. It is also responsible for getting
+ * this node to join the cluster. This node should not process any Web services requests until it
+ * successfully joins the cluster. Generally, this node will also need to obtain the state
+ * information and/or configuration information from a neighboring node.
+ * This interface is also responsible for
+ * properly instantiating a {@link org.apache.axis2.clustering.state.StateManager} &
+ * {@link org.apache.axis2.clustering.management.NodeManager}. In the case of
+ * a static <a href="http://afkham.org/2008/05/group-membership-management-schemes.html">
+ * membership scheme</a>,
+ * this members are read from the axis2.xml file and added to the ClusteringAgent.
+ * </p>
+ * <p>
+ * In the axis2.xml, the instance of this interface is specified using the "clustering"
+ * class attribute.
+ * e.g.
+ * <code><b>
+ * <nobr><clustering class="org.apache.axis2.clustering.tribes.TribesClusterAgent"></nobr>
+ * </b>
+ * </code>
+ * specifies that the TribesClusterAgent class is the instance of this interface that
+ * needs to be used.
+ * </p>
+ * <p>
+ * There can also be several "parameter" elements, which are children of the "clustering" element
+ * in the axis2.xml file. Generally, these parameters will be specific to the ClusteringAgent
+ * implementation.
+ * </p>
+ */
+public interface ClusteringAgent extends ParameterInclude {
+
+ /**
+ * Initialize this node, and join the cluster
+ *
+ * @throws ClusteringFault If an error occurs while initializing this node or joining the cluster
+ */
+ void init() throws ClusteringFault;
+
+ /**
+ * Do cleanup & leave the cluster
+ */
+ void finalize();
+
+ /**
+ * @return The StateManager
+ */
+ StateManager getStateManager();
+
+ /**
+ * @return The NodeManager
+ */
+ NodeManager getNodeManager();
+
+ /**
+ * Set the StateManager corresponding to this ClusteringAgent. This is an optional attribute.
+ * We can have a cluster with no context replication, in which case the contextManager will be
+ * null. This value is set by the {@link org.apache.axis2.deployment.ClusterBuilder}, by
+ * reading the "contextManager" element in the axis2.xml
+ * <p/>
+ * e.g.
+ * <code>
+ * <b>
+ * <contextManager class="org.apache.axis2.cluster.configuration.TribesContextManager">
+ * </b>
+ * </code>
+ *
+ * @param stateManager The StateManager instance
+ */
+ void setStateManager(StateManager stateManager);
+
+ /**
+ * Set the NodeManager corresponding to this ClusteringAgent. This is an optional attribute.
+ * We can have a cluster with no configuration management, in which case the configurationManager
+ * will be null. This value is set by the {@link org.apache.axis2.deployment.ClusterBuilder}, by
+ * reading the "configurationManager" element in the axis2.xml
+ * <p/>
+ * e.g.
+ * <code>
+ * <b>
+ * <configurationManager class="org.apache.axis2.cluster.configuration.TribesConfigurationManager">
+ * </b>
+ * </code>
+ *
+ * @param nodeManager The NodeManager instance
+ */
+ void setNodeManager(NodeManager nodeManager);
+
+ /**
+ * Disconnect this node from the cluster. This node will no longer receive membership change
+ * notifications, state change messages or configuration change messages. The node will be "
+ * "standing alone" once it is shutdown. However, it has to continue to process Web service
+ * requests.
+ *
+ * @throws ClusteringFault If an error occurs while leaving the cluster
+ */
+ void shutdown() throws ClusteringFault;
+
+ /**
+ * Set the system's configuration context. This will be used by the clustering implementations
+ * to get information about the Axis2 environment and to correspond with the Axis2 environment
+ *
+ * @param configurationContext The configuration context
+ */
+ void setConfigurationContext(ConfigurationContext configurationContext);
+
+ /**
+ * Set the static members of the cluster. This is used only with
+ * <a href="http://afkham.org/2008/05/group-membership-management-schemes.html">
+ * static group membership </a>
+ *
+ * @param members Members to be added
+ */
+ void setMembers(List<Member> members);
+
+ /**
+ * Get the list of members in a
+ * <a href="http://afkham.org/2008/05/group-membership-management-schemes.html">
+ * static group
+ * </a>
+ *
+ * @return The members if static group membership is used. If any other membership scheme is used,
+ * the values returned may not be valid
+ */
+ List<Member> getMembers();
+
+ /**
+ * Set the load balance event handler which will be notified when load balance events occur.
+ * This will be valid only when this node is running in loadBalance mode
+ *
+ * @param agent The load balance event handler
+ * @param applicationDomain The application domain which is handled by the eventHandler
+ */
+ void addGroupManagementAgent(GroupManagementAgent agent, String applicationDomain);
+
+ /**
+ * Get the LoadBalanceEventHandler which corresponds to the <code>applicationDomain</code>
+ * This will be valid only when this node is running in loadBalance mode
+ *
+ * @param applicationDomain The application domain to which the application nodes being
+ * load balanced belong to
+ * @return LoadBalanceEventHandler which corresponds to the <code>applicationDomain</code>
+ */
+ GroupManagementAgent getGroupManagementAgent(String applicationDomain);
+
+ /**
+ * Get all the domains that this ClusteringAgent belongs to
+ *
+ * @return the domains of this ClusteringAgent
+ */
+ Set<String> getDomains();
+}
\ No newline at end of file
Modified: webservices/axis2/trunk/java/modules/kernel/src/org/apache/axis2/clustering/ClusteringCommand.java
URL: http://svn.apache.org/viewvc/webservices/axis2/trunk/java/modules/kernel/src/org/apache/axis2/clustering/ClusteringCommand.java?rev=744900&r1=744899&r2=744900&view=diff
==============================================================================
--- webservices/axis2/trunk/java/modules/kernel/src/org/apache/axis2/clustering/ClusteringCommand.java (original)
+++ webservices/axis2/trunk/java/modules/kernel/src/org/apache/axis2/clustering/ClusteringCommand.java Mon Feb 16 12:02:42 2009
@@ -19,8 +19,11 @@
package org.apache.axis2.clustering;
+import org.apache.axis2.context.ConfigurationContext;
+
import java.io.Serializable;
public abstract class ClusteringCommand implements Serializable {
+ public abstract void execute(ConfigurationContext configContext) throws ClusteringFault;
}
Modified: webservices/axis2/trunk/java/modules/kernel/src/org/apache/axis2/clustering/Member.java
URL: http://svn.apache.org/viewvc/webservices/axis2/trunk/java/modules/kernel/src/org/apache/axis2/clustering/Member.java?rev=744900&r1=744899&r2=744900&view=diff
==============================================================================
--- webservices/axis2/trunk/java/modules/kernel/src/org/apache/axis2/clustering/Member.java (original)
+++ webservices/axis2/trunk/java/modules/kernel/src/org/apache/axis2/clustering/Member.java Mon Feb 16 12:02:42 2009
@@ -1,17 +1,17 @@
-/*
- * Copyright 2004,2005 The Apache Software Foundation.
- *
- * Licensed 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.
+/*
+ * Copyright 2004,2005 The Apache Software Foundation.
+ *
+ * Licensed 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.axis2.clustering;