import org.apache.axis2.clustering.ClusteringAgent; //导入依赖的package包/类
/**
 * Send out policy status change notification to other nodes.
 *
 * @param clusterMessage
 * @param isSync
 */
private void sendClusterMessage(PolicyStatusClusterMessage clusterMessage, boolean isSync) {
    try {
        if (log.isDebugEnabled()) {
            log.debug("Sending policy status change cluster message to all other nodes");
        }

        ClusteringAgent clusteringAgent = EntitlementConfigHolder.getInstance()
                .getConfigurationContextService()
                .getServerConfigContext()
                .getAxisConfiguration()
                .getClusteringAgent();

        if (clusteringAgent != null) {
            clusteringAgent.sendMessage(clusterMessage, isSync);
        } else {
            log.error("Clustering Agent not available.");
        }
    } catch (ClusteringFault clusteringFault) {
        log.error("Error while sending policy status change cluster message", clusteringFault);
    }
}
 

import org.apache.axis2.clustering.ClusteringAgent; //导入依赖的package包/类
private void replicateAssociationInfo(AssociationClusterMessage associationInfoData) {
    if (log.isDebugEnabled()) {
        log.debug("Starting to replicate association : " + associationInfoData.getAssociation().getHandle());
    }

    ClusteringAgent agent = IdentityProviderServiceComponent.getConfigContext().getAxisConfiguration().getClusteringAgent();
    if (log.isDebugEnabled()) {
        log.debug("Clustering Agent: " + agent);
    }

    if (agent != null) {
        try {
            agent.sendMessage(associationInfoData, true);
        } catch (ClusteringFault e) {
            log.error("Unable to send cluster message :" + e.getMessage(), e);
        }
    }
    if (log.isDebugEnabled()) {
        log.debug("Completed replicating association : " + associationInfoData.getAssociation().getHandle());
    }
}
 

import org.apache.axis2.clustering.ClusteringAgent; //导入依赖的package包/类
private void loadWellKnownMembers(ClusteringAgent clusteringAgent, OMElement clusterElement) {
    clusteringAgent.setMembers(new ArrayList<Member>());
    Parameter membershipSchemeParam = clusteringAgent.getParameter("membershipScheme");
    if (membershipSchemeParam != null) {
        String membershipScheme = ((String) membershipSchemeParam.getValue()).trim();
        if (membershipScheme.equals(ClusteringConstants.MembershipScheme.WKA_BASED)) {
            List<Member> members = new ArrayList<Member>();
            OMElement membersEle =
                    clusterElement.getFirstChildWithName(new QName("members"));
            if (membersEle != null) {
                for (Iterator iter = membersEle.getChildrenWithLocalName("member"); iter.hasNext();) {
                    OMElement memberEle = (OMElement) iter.next();
                    String hostName =
                            memberEle.getFirstChildWithName(new QName("hostName")).getText().trim();
                    String port =
                            memberEle.getFirstChildWithName(new QName("port")).getText().trim();
                    members.add(new Member(replaceVariables(hostName),
                                           Integer.parseInt(replaceVariables(port))));
                }
            }
            clusteringAgent.setMembers(members);
        }
    }
}
 

import org.apache.axis2.clustering.ClusteringAgent; //导入依赖的package包/类
/**
 * Will create a configuration context from the avialable data and then it
 * will start the listener manager
 *
 * @throws AxisFault if something went wrong
 */
protected void start() throws AxisFault {
    if (configContext == null) {
        configContext = getConfigurationContext();
    }
    if (!started) {

        ClusteringAgent clusteringAgent =
                configContext.getAxisConfiguration().getClusteringAgent();
        String avoidInit = ClusteringConstants.Parameters.AVOID_INITIATION;
        if (clusteringAgent != null &&
            clusteringAgent.getParameter(avoidInit) != null &&
            ((String) clusteringAgent.getParameter(avoidInit).getValue()).equalsIgnoreCase("true")) {
            clusteringAgent.setConfigurationContext(configContext);
            clusteringAgent.init();
        }

        listenerManager.startSystem(configContext);
        started = true;
    }
}
 

import org.apache.axis2.clustering.ClusteringAgent; //导入依赖的package包/类
/**
 * Initializes the ClusterManager for this ConfigurationContext
 *
 * @throws AxisFault
 */
public void initCluster() throws AxisFault {
    ClusteringAgent clusteringAgent = axisConfiguration.getClusteringAgent();
    if (clusteringAgent != null) {
        StateManager stateManaget = clusteringAgent.getStateManager();
        if (stateManaget != null) {
            stateManaget.setConfigurationContext(this);
        }
        NodeManager nodeManager = clusteringAgent.getNodeManager();
        if (nodeManager != null) {
            nodeManager.setConfigurationContext(this);
        }
        if (shouldClusterBeInitiated(clusteringAgent)) {
            clusteringAgent.setConfigurationContext(this);
            clusteringAgent.init();
        }
    }
}
 

import org.apache.axis2.clustering.ClusteringAgent; //导入依赖的package包/类
/**
 * @return true if we need to store property differences for this 
 * context in this scenario.
 */
private boolean needPropertyDifferences() {
    
    // Don't store property differences if there are no 
    // cluster members.
    
    ConfigurationContext cc = getRootContext();
    if (cc == null) {
        return false;
    }
    // Add the property differences only if Context replication is enabled,
    // and there are members in the cluster
    ClusteringAgent clusteringAgent = cc.getAxisConfiguration().getClusteringAgent();
    if (clusteringAgent == null ||
        clusteringAgent.getStateManager() == null) {
        return false;
    }
    return true;
}
 

import org.apache.axis2.clustering.ClusteringAgent; //导入依赖的package包/类
public GroupMember[] getMembers(String groupName) throws Exception {
    ClusteringAgent clusteringAgent = getClusteringAgent();
    GroupManagementAgent groupManagementAgent =
            clusteringAgent.getGroupManagementAgent(groupName);
    if (groupManagementAgent == null) {
        handleException("No GroupManagementAgent defined for domain " + groupName);
        return null;
    }
    List<Member> members = groupManagementAgent.getMembers();
    GroupMember[] groupMembers = new GroupMember[members.size()];
    int i = 0;
    for (Member member : members) {
        GroupMember groupMember = new GroupMember();
        groupMember.setHostName(member.getHostName());
        groupMember.setHttpPort(member.getHttpPort());
        groupMember.setHttpsPort(member.getHttpsPort());
        Properties properties = member.getProperties();
        groupMember.setBackendServerURL(properties.getProperty("backendServerURL"));
        groupMember.setMgConsoleURL(properties.getProperty("mgtConsoleURL"));
        groupMembers[i] = groupMember;
        i++;
    }

    return groupMembers;
}
 

import org.apache.axis2.clustering.ClusteringAgent; //导入依赖的package包/类
public static Boolean addServiceMappingToCluster(String mapping, String epr) throws AxisFault {
    Boolean isMappingAdded = false;
    try {
        ClusteringAgent agent = getClusteringAgent();
        if(agent != null) {
            agent.sendMessage(new ServiceMappingAddRequest(mapping, epr), true);
            isMappingAdded = true;
        }
        if (log.isDebugEnabled()) {
            log.debug("sent cluster command to to get Active tenants on cluster");
        }
    } catch (AxisFault f) {
        String msg = "Error in getting active tenant by cluster commands";
        log.error(msg, f);
        throw new AxisFault(msg);
    }
    return isMappingAdded;
}
 

import org.apache.axis2.clustering.ClusteringAgent; //导入依赖的package包/类
public static Boolean addVirtualHostsToCluster(String hostName, String uri, String webappPath) throws AxisFault {
    Boolean isMappingAdded = false;
    try {
        ClusteringAgent agent = getClusteringAgent();
        if(agent != null) {
            agent.sendMessage(new VirtualHostAddRequest(hostName, uri, webappPath), true);
            isMappingAdded = true;
        }
        if (log.isDebugEnabled()) {
            log.debug("sent cluster command to to get Active tenants on cluster");
        }
    } catch (AxisFault f) {
        String msg = "Error in getting active tenant by cluster commands";
        log.error(msg, f);
        throw new AxisFault(msg);
    }
    return isMappingAdded;
}
 

import org.apache.axis2.clustering.ClusteringAgent; //导入依赖的package包/类
public static Boolean deleteServiceMappingToCluster(String mapping) throws AxisFault {
    Boolean isMappingAdded = false;
    try {
        ClusteringAgent agent = getClusteringAgent();
        if(agent != null) {
            agent.sendMessage(new ServiceMappingDeleteRequest(mapping), true);
            isMappingAdded = true;
        }
        if (log.isDebugEnabled()) {
            log.debug("sent cluster command to to get Active tenants on cluster");
        }
    } catch (AxisFault f) {
        String msg = "Error in getting active tenant by cluster commands";
        log.error(msg, f);
        throw new AxisFault(msg);
    }
    return isMappingAdded;
}
 

import org.apache.axis2.clustering.ClusteringAgent; //导入依赖的package包/类
public static Boolean deleteVirtualHostsToCluster(String hostName) throws AxisFault {
    Boolean isMappingAdded = false;
    try {
        ClusteringAgent agent = getClusteringAgent();
        if(agent != null) {
            agent.sendMessage(new VirtualHostDeleteMapping(hostName), true);
            isMappingAdded = true;
        }
        if (log.isDebugEnabled()) {
            log.debug("sent cluster command to to get Active tenants on cluster");
        }
    } catch (AxisFault f) {
        String msg = "Error in getting active tenant by cluster commands";
        log.error(msg, f);
        throw new AxisFault(msg);
    }
    return isMappingAdded;
}
 

import org.apache.axis2.clustering.ClusteringAgent; //导入依赖的package包/类
/**
 * Sends a cluster message to other members of the cluster.If message transmission fails it will
 * create a thread which will attempt retry transmission for a predefined amount of retry attempts.
 *
 * @param message The message to be transmitted
 */
public static void sendClusterMessage(ClusteringMessage message) {
    ClusteringAgent agent = createClusteringAgent();
    if (agent == null) {
        //log.error("Unable to send the clustering message as a clustering agent was not obtained.");
        if (log.isDebugEnabled()) {
            log.debug(String.format("Failed to send cluster message :%s ", message));
        }
        return;
    }
    try {
        agent.sendMessage(message, true);
        if (log.isDebugEnabled()) {

            log.debug(String.format("Successfully transmitted cluster message :%s", message));
        }
    } catch (ClusteringFault e) {
        if (log.isDebugEnabled()) {
            log.error("Unable to send the clustering message.The system will now attempt to retry " +
                    "sending the message", e);
        }
    }
}
 

import org.apache.axis2.clustering.ClusteringAgent; //导入依赖的package包/类
public void sendSessionInvalidationClusterMessage(String sessionIndex) {

        SessionClusterMessage clusterMessage = new SessionClusterMessage();
        clusterMessage.setMessageId(UUID.randomUUID());
        clusterMessage.setSessionIndex(sessionIndex);

        ClusteringAgent clusteringAgent = SAML2SSOAuthFEDataHolder.getInstance()
                .getConfigurationContextService().getServerConfigContext().getAxisConfiguration()
                .getClusteringAgent();

        if (clusteringAgent != null) {
            int numberOfRetries = 0;

            while (numberOfRetries < 60) {
                try {
                    clusteringAgent.sendMessage(clusterMessage, true);
                    log.info("Sent [" + clusterMessage + "]");
                    break;
                } catch (ClusteringFault e) {
                    numberOfRetries++;

                    if (numberOfRetries < 60) {
                        log.warn(
                                "Could not send SSOSessionInvalidationClusterMessage. Retry will be attempted in 2s. Request: "
                                        + clusterMessage, e);
                    } else {
                        log.error(
                                "Could not send SSOSessionInvalidationClusterMessage. Several retries failed. Request:"
                                        + clusterMessage, e);
                    }

                    try {
                        Thread.sleep(2000);
                    } catch (InterruptedException ignored) {
                    }
                }
            }
        }
    }
 

import org.apache.axis2.clustering.ClusteringAgent; //导入依赖的package包/类
/**
    * This is to send clusterMessage to inform other nodes about subscription added to the system, so that everyone can add new one.
    * @param topicName
    * @param subsciptionID
    * @param tenantID
    * @param tenantName
    * @throws ClusteringFault
    */
public static void sendAddSubscriptionClusterMessage(String topicName,String subsciptionID, 
		int tenantID, String tenantName) throws ClusteringFault{
	ConfigurationContextService configContextService = (ConfigurationContextService) PrivilegedCarbonContext
			.getThreadLocalCarbonContext().getOSGiService(ConfigurationContextService.class);
	ConfigurationContext configContext = configContextService.getServerConfigContext();
	ClusteringAgent agent = configContext.getAxisConfiguration().getClusteringAgent();

	agent.sendMessage(new SubscriptionClusterMessage(topicName,subsciptionID,tenantID, tenantName), false);
}
 

import org.apache.axis2.clustering.ClusteringAgent; //导入依赖的package包/类
protected ClusteringAgent getClusterManager(ConfigurationContext configCtx) {
    TribesClusteringAgent tribesClusterManager = new TribesClusteringAgent();
    tribesClusterManager.setConfigurationContext(configCtx);
    DefaultNodeManager configurationManager = new DefaultNodeManager();
    tribesClusterManager.setNodeManager(configurationManager);
    DefaultStateManager contextManager = new DefaultStateManager();
    tribesClusterManager.setStateManager(contextManager);
    return tribesClusterManager;
}
 

import org.apache.axis2.clustering.ClusteringAgent; //导入依赖的package包/类
private void loadNodeManager(OMElement clusterElement,
                               ClusteringAgent clusteringAgent) throws DeploymentException,
                                                                     InstantiationException,
                                                                     IllegalAccessException {
    OMElement configManagerEle =
            clusterElement.getFirstChildWithName(new QName(TAG_NODE_MANAGER));
    if (configManagerEle != null) {
        if (!isEnabled(configManagerEle)) {
            log.info("Clustering configuration management has been disabled");
            return;
        }
        log.info("Clustering configuration management has been enabled");

        OMAttribute classNameAttr = configManagerEle.getAttribute(new QName(ATTRIBUTE_CLASS));
        if (classNameAttr == null) {
            throw new DeploymentException(Messages.getMessage("classAttributeNotFound",
                                                              TAG_NODE_MANAGER));
        }

        String className = classNameAttr.getAttributeValue();
        Class clazz;
        try {
            clazz = Class.forName(className);
        } catch (ClassNotFoundException e) {
            throw new DeploymentException(Messages.getMessage("clusterImplNotFound",
                                                              className));
        }

        NodeManager nodeManager = (NodeManager) clazz.newInstance();
        clusteringAgent.setNodeManager(nodeManager);

        //updating the NodeManager with the new ConfigurationContext
        nodeManager.setConfigurationContext(configCtx);

        //loading the parameters.
        processParameters(configManagerEle.getChildrenWithName(new QName(TAG_PARAMETER)),
                          nodeManager,
                          null);
    }
}
 

import org.apache.axis2.clustering.ClusteringAgent; //导入依赖的package包/类
private static StateManager getStateManager(AxisConfiguration axisConfiguration) {
    ClusteringAgent clusteringAgent = axisConfiguration.getClusteringAgent();
    if (clusteringAgent != null) {
        return clusteringAgent.getStateManager();
    }
    return null;
}
 

import org.apache.axis2.clustering.ClusteringAgent; //导入依赖的package包/类
private void sendLoggingConfigSyncClusterMessage() {
    ClusteringAgent clusteringAgent =
            DataHolder.getInstance().getServerConfigContext().getAxisConfiguration().getClusteringAgent();
    if (clusteringAgent != null) {
        try {
            clusteringAgent.sendMessage(new LoggingConfigSyncRequest(), true);
        } catch (ClusteringFault clusteringFault) {
            log.error("Logging configuration synchronization cluster message failed.", clusteringFault);
        }
    }
}
 

import org.apache.axis2.clustering.ClusteringAgent; //导入依赖的package包/类
private ClusteringAgent getClusteringAgent() throws TaskException {
    ConfigurationContextService configCtxService = TasksDSComponent
            .getConfigurationContextService();
    if (configCtxService == null) {
        throw new TaskException("ConfigurationContextService not available "
                + "for notifying the cluster", Code.UNKNOWN);
    }
    ConfigurationContext configCtx = configCtxService.getServerConfigContext();
    ClusteringAgent agent = configCtx.getAxisConfiguration().getClusteringAgent();
    if (log.isDebugEnabled()) {
        log.debug("Clustering Agent: " + agent);
    }
    return agent;
}
 

import org.apache.axis2.clustering.ClusteringAgent; //导入依赖的package包/类
private TaskState getTaskStateFromLocalCluster(String taskName) throws TaskException {
    /* first check local server */
    if (this.isTaskRunning(taskName)) {
        return TaskState.BLOCKED;
    }
    ClusteringAgent agent = this.getClusteringAgent();
    if (agent == null) {
        return TaskState.UNKNOWN;
    }
    TaskStatusMessage msg = new TaskStatusMessage();
    msg.setTaskName(taskName);
    msg.setTaskType(this.getTaskType());
    msg.setTenantId(this.getTenantId());
    try {
        List<ClusteringCommand> result = agent.sendMessage(msg, true);
        TaskStatusResult status;
        for (ClusteringCommand entry : result) {
            status = (TaskStatusResult) entry;
            if (status.isRunning()) {
                return TaskState.BLOCKED;
            }
        }
        return TaskState.NORMAL;
    } catch (ClusteringFault e) {
        throw new TaskException(e.getMessage(), Code.UNKNOWN, e);
    }
}