| /* |
| * 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.*; |
| 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.clustering.management.DefaultGroupManagementAgent; |
| import org.apache.axis2.clustering.management.DefaultNodeManager; |
| import org.apache.axis2.clustering.management.GroupManagementAgent; |
| import org.apache.axis2.clustering.management.NodeManager; |
| import org.apache.axis2.clustering.state.ClusteringContextListener; |
| import org.apache.axis2.clustering.state.DefaultStateManager; |
| import org.apache.axis2.clustering.state.StateManager; |
| import org.apache.axis2.context.ConfigurationContext; |
| import org.apache.axis2.util.JavaUtils; |
| 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.ErrorHandler; |
| import org.apache.catalina.tribes.ManagedChannel; |
| import org.apache.catalina.tribes.Member; |
| import org.apache.catalina.tribes.UniqueId; |
| 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.group.interceptors.NonBlockingCoordinator; |
| 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.InvocationTargetException; |
| import java.lang.reflect.Method; |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.Collections; |
| 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); |
| public static final String DEFAULT_SUB_DOMAIN = "__$default"; |
| |
| private DefaultNodeManager configurationManager; |
| private DefaultStateManager contextManager; |
| |
| private final HashMap<String, Parameter> parameters; |
| private ManagedChannel channel; |
| /** |
| * RpcChannel used for cluster initialization interactions |
| */ |
| private RpcChannel rpcInitChannel; |
| /** |
| * RpcChannel used for RPC messaging interactions |
| */ |
| private RpcChannel rpcMessagingChannel; |
| private ConfigurationContext configurationContext; |
| private Axis2ChannelListener axis2ChannelListener; |
| private ChannelSender channelSender; |
| private MembershipManager primaryMembershipManager; |
| private RpcInitializationRequestHandler rpcInitRequestHandler; |
| private MembershipScheme membershipScheme; |
| |
| private NonBlockingCoordinator coordinator; |
| |
| /** |
| * 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; |
| |
| /** |
| * Map[key, value=Map[key, value]] = [domain, [subDomain, GroupManagementAgent]] |
| */ |
| private final Map<String, Map<String, GroupManagementAgent>> groupManagementAgents = |
| new HashMap<String, Map<String, GroupManagementAgent>>(); |
| private boolean clusterManagementMode; |
| private RpcMessagingHandler rpcMessagingHandler; |
| |
| 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 int getAliveMemberCount() { |
| return primaryMembershipManager.getMembers().length; |
| } |
| |
| public void addGroupManagementAgent(GroupManagementAgent agent, String applicationDomain) { |
| addGroupManagementAgent(agent, applicationDomain, null); |
| } |
| |
| public void addGroupManagementAgent(GroupManagementAgent agent, String applicationDomain, |
| String applicationSubDomain) { |
| if (applicationSubDomain == null) { |
| applicationSubDomain = DEFAULT_SUB_DOMAIN; // default sub-domain since a sub-domain is not specified |
| } |
| log.info("Managing group application domain:" + applicationDomain + ", sub-domain:" + |
| applicationSubDomain + " using agent " + agent.getClass()); |
| if(!groupManagementAgents.containsKey(applicationDomain)){ |
| groupManagementAgents.put(applicationDomain, new HashMap<String, GroupManagementAgent>()); |
| } |
| groupManagementAgents.get(applicationDomain).put(applicationSubDomain, agent); |
| clusterManagementMode = true; |
| } |
| |
| public GroupManagementAgent getGroupManagementAgent(String applicationDomain) { |
| return getGroupManagementAgent(applicationDomain, null); |
| } |
| |
| public GroupManagementAgent getGroupManagementAgent(String applicationDomain, |
| String applicationSubDomain) { |
| if (applicationSubDomain == null) { |
| applicationSubDomain = DEFAULT_SUB_DOMAIN; // default sub-domain since a sub-domain is not specified |
| } |
| Map<String, GroupManagementAgent> groupManagementAgentMap = groupManagementAgents.get(applicationDomain); |
| if (groupManagementAgentMap != null) { |
| return groupManagementAgentMap.get(applicationSubDomain); |
| } |
| return null; |
| } |
| |
| public Set<String> getDomains() { |
| return groupManagementAgents.keySet(); |
| } |
| |
| public StateManager getStateManager() { |
| return contextManager; |
| } |
| |
| public NodeManager getNodeManager() { |
| return configurationManager; |
| } |
| |
| public boolean isCoordinator(){ |
| return coordinator.isCoordinator(); |
| } |
| |
| /** |
| * Initialize the cluster. |
| * |
| * @throws ClusteringFault If initialization fails |
| */ |
| public void init() throws ClusteringFault { |
| log.info("Initializing cluster..."); |
| addRequestBlockingHandlerToInFlows(); |
| primaryMembershipManager = new MembershipManager(configurationContext); |
| |
| channel = new Axis2GroupChannel(); |
| coordinator = new NonBlockingCoordinator(); |
| channel.addInterceptor(coordinator); |
| channel.setHeartbeat(true); |
| channelSender = new ChannelSender(channel, primaryMembershipManager, synchronizeAllMembers()); |
| axis2ChannelListener = |
| new Axis2ChannelListener(configurationContext, configurationManager, contextManager); |
| channel.addChannelListener(axis2ChannelListener); |
| |
| 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 Init Channel for domain " + new String(domain)); |
| } |
| |
| // Initialize RpcChannel used for messaging |
| rpcMessagingHandler = new RpcMessagingHandler(configurationContext); |
| rpcMessagingChannel = |
| new RpcChannel(TribesUtil.getRpcMessagingChannelId(domain), channel, |
| rpcMessagingHandler); |
| if (log.isDebugEnabled()) { |
| log.debug("Created RPC Messaging 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.")) { |
| log.warn("Local member advertising its IP address as 127.0.0.1. " + |
| "Remote members will not be able to connect to this member."); |
| } |
| } 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(); |
| |
| configurationContext.getAxisConfiguration().addObservers(new TribesAxisObserver()); |
| ClassLoaderUtil.init(configurationContext.getAxisConfiguration()); |
| |
| // 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); |
| axis2ChannelListener.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); |
| } |
| } |
| } |
| |
| public List<ClusteringCommand> sendMessage(ClusteringMessage message, |
| boolean isRpcMessage) throws ClusteringFault { |
| List<ClusteringCommand> responseList = new ArrayList<ClusteringCommand>(); |
| Member[] members = primaryMembershipManager.getMembers(); |
| if (members.length == 0) { |
| return responseList; |
| } |
| if (isRpcMessage) { |
| try { |
| Response[] responses = rpcMessagingChannel.send(members, message, RpcChannel.ALL_REPLY, |
| Channel.SEND_OPTIONS_SYNCHRONIZED_ACK, |
| 10000); |
| for (Response response : responses) { |
| responseList.add((ClusteringCommand)response.getMessage()); |
| } |
| } catch (ChannelException e) { |
| String msg = "Error occurred while sending RPC message to cluster."; |
| log.error(msg, e); |
| throw new ClusteringFault(msg, e); |
| } |
| } else { |
| try { |
| channel.send(members, message, 10000, new ErrorHandler(){ |
| public void handleError(ChannelException e, UniqueId uniqueId) { |
| log.error("Sending failed " + uniqueId, e ); |
| } |
| |
| public void handleCompletion(UniqueId uniqueId) { |
| if(log.isDebugEnabled()){ |
| log.debug("Sending successful " + uniqueId); |
| } |
| } |
| }); |
| } catch (ChannelException e) { |
| String msg = "Error occurred while sending message to cluster."; |
| log.error(msg, e); |
| throw new ClusteringFault(msg, e); |
| } |
| } |
| return responseList; |
| } |
| |
| private void setMemberInfo() throws ClusteringFault { |
| Properties memberInfo = new Properties(); |
| AxisConfiguration axisConfig = configurationContext.getAxisConfiguration(); |
| TransportInDescription httpTransport = axisConfig.getTransportIn("http"); |
| int portOffset = 0; |
| Parameter param = getParameter(ClusteringConstants.Parameters.AVOID_INITIATION); |
| if(param != null && !JavaUtils.isTrueExplicitly(param.getValue())){ |
| //AvoidInitialization = false, Hence we set the portOffset |
| if(System.getProperty("portOffset") != null){ |
| portOffset = Integer.parseInt(System.getProperty("portOffset")); |
| } |
| } |
| |
| if (httpTransport != null) { |
| Parameter port = httpTransport.getParameter("port"); |
| if (port != null) { |
| memberInfo.put("httpPort", |
| String.valueOf(Integer.valueOf((String)port.getValue()) + portOffset)); |
| } |
| } |
| TransportInDescription httpsTransport = axisConfig.getTransportIn("https"); |
| if (httpsTransport != null) { |
| Parameter port = httpsTransport.getParameter("port"); |
| if (port != null) { |
| memberInfo.put("httpsPort", |
| String.valueOf(Integer.valueOf((String)port.getValue()) + portOffset)); |
| } |
| } |
| 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){ |
| String attribName = nameAttrib.getAttributeValue(); |
| attribName = replaceProperty(attribName, memberInfo); |
| |
| OMAttribute valueAttrib = propEle.getAttribute(new QName("value")); |
| if (valueAttrib != null) { |
| String attribVal = valueAttrib.getAttributeValue(); |
| attribVal = replaceProperty(attribVal, memberInfo); |
| memberInfo.setProperty(attribName, 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) { |
| propValue = System.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 (Map<String, GroupManagementAgent> agents : groupManagementAgents.values()) { |
| for (GroupManagementAgent agent : agents.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 (InvocationTargetException e) { |
| handleException("Error invoking setter method named : " + mName + |
| "() that takes a single String, int, long, float, double " + |
| "or boolean parameter", e); |
| } catch (NoSuchMethodException e) { |
| handleException("Error invoking setter method named : " + mName + |
| "() that takes a single String, int, long, float, double " + |
| "or boolean parameter", e); |
| } catch (IllegalAccessException 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/> |
| * rpcInitChannel 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 initialization 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 | |
| Channel.SEND_OPTIONS_BYTE_MESSAGE, |
| 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(rpcMessagingChannel); |
| channel.removeChannelListener(axis2ChannelListener); |
| 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 (rpcMessagingHandler!= null) { |
| rpcMessagingHandler.setConfigurationContext(configurationContext); |
| } |
| if (axis2ChannelListener != null) { |
| axis2ChannelListener.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()); |
| } |
| } |
| |
