import org.apache.axis2.clustering.ClusteringFault; //导入依赖的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.ClusteringFault; //导入依赖的package包/类
@Override
public void init() throws ClusteringFault {
    try {
        log.info("Initializing clustering membership scheme for Mesos...");
        configureTcpIpParameters();
        loadConfiguration();
        if (MESOS_DNS_DISCOVERY_SCHEME.equals(memberDiscoveryScheme)) {
            addMembersFromMesosDNS();
        } else if (MESOS_MARATHON_DISCOVERY_SCHEME.equals(memberDiscoveryScheme)) {
            addMembersFromMesosMarathon();
        } else {
            throw new ClusteringFault(
                    String.format("Unsupported Mesos member discovery scheme [Discovery-Scheme] %s", memberDiscoveryScheme));
        }
        log.info("Mesos clustering membership scheme initialized successfully");
    } catch (Exception e) {
        throw new ClusteringFault("Mesos clustering membership initialization failed", e);
    }
}
 

import org.apache.axis2.clustering.ClusteringFault; //导入依赖的package包/类
public String greet(String name) {
    ServiceContext serviceContext =
            MessageContext.getCurrentMessageContext().getServiceContext();
    serviceContext.setProperty(HELLO_SERVICE_NAME, name);
    try {
        Replicator.replicate(serviceContext, new String[]{HELLO_SERVICE_NAME});
    } catch (ClusteringFault clusteringFault) {
        clusteringFault.printStackTrace();
    }

    if (name != null) {
        return "Hello World, " + name + " !!!";
    } else {
        return "Hello World !!!";
    }
}
 

import org.apache.axis2.clustering.ClusteringFault; //导入依赖的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.ClusteringFault; //导入依赖的package包/类
public void addSubscription(Subscription subscription) {
      String topicName = getTopicName(subscription.getTopicName());
      SubscriptionContainer subscriptionsContainer = getTopicSubscriptionCache().get(topicName);

      if (subscriptionsContainer == null){
          subscriptionsContainer = new SubscriptionContainer(topicName);
      }
      
      subscriptionsContainer.getSubscriptionsCache().put(subscription.getId(), subscription);
      getTopicSubscriptionCache().put(topicName, subscriptionsContainer);
      getSubscriptionIDTopicNameCache().put(subscription.getId(), topicName);
      
      //send cluster message to inform other nodes about subscription added to the system, so that everyone can add new one.
      try {
	SharedMemoryCacheUtil.sendAddSubscriptionClusterMessage(topicName, subscription.getId(), 
			subscription.getTenantId(),
			subscription.getTenantDomain());
} catch (ClusteringFault e) {
	//This happen due to hazelcast clustering not setup correctly. subscription cluster will not send to other nodes.
	log.error("Subscription cluster message sending failed",e);
}
  }
 

import org.apache.axis2.clustering.ClusteringFault; //导入依赖的package包/类
@Override
public void execute(ConfigurationContext arg0) throws ClusteringFault {

       try {
           PrivilegedCarbonContext.startTenantFlow();
           PrivilegedCarbonContext.getThreadLocalCarbonContext().setTenantId(tenantID);
           PrivilegedCarbonContext.getThreadLocalCarbonContext().setTenantDomain(tenantDomain);
           PrivilegedCarbonContext.getThreadLocalCarbonContext().getTenantDomain(true);
          
           //get the subscription by adding the 
           SubscriptionContainer subscriptionsContainer = SharedMemorySubscriptionStorage.getTopicSubscriptionCache().get(topicName);

           if (subscriptionsContainer != null){
           	subscriptionsContainer.getSubscriptionsCache().get(subscriptionID);
           	log.info("Subscription ID: "+subscriptionID+" for the topic: "+topicName+" is received.");
           }          

       }
       finally{
           PrivilegedCarbonContext.endTenantFlow();
       }
}
 

import org.apache.axis2.clustering.ClusteringFault; //导入依赖的package包/类
public void execute(ConfigurationContext configContext) throws ClusteringFault {
        log.info("Received state initialization message");
        
        // Run this code only if this node is not already initialized
        if (configContext.
                getPropertyNonReplicable(ClusteringConstants.RECD_STATE_INIT_MSG) == null) {
            configContext.
                setNonReplicableProperty(ClusteringConstants.RECD_STATE_INIT_MSG, "true");
//            log.info("Received state initialization message");
            if (commands != null) {
                for (int i = 0; i < commands.length; i++) {
                    commands[i].execute(configContext);
                }
            }
        }
    }
 

import org.apache.axis2.clustering.ClusteringFault; //导入依赖的package包/类
public void execute(ConfigurationContext configCtx) throws ClusteringFault {

        List serviceGroupNames = new ArrayList();
        AxisConfiguration axisConfig = configCtx.getAxisConfiguration();
        for (Iterator iter = axisConfig.getServiceGroups(); iter.hasNext();) {
            AxisServiceGroup serviceGroup = (AxisServiceGroup) iter.next();
            boolean excludeSG = false;
            for (Iterator serviceIter = serviceGroup.getServices(); serviceIter.hasNext();) {
                AxisService service = (AxisService) serviceIter.next();
                if (service.getParameter(AxisModule.MODULE_SERVICE) != null ||
                    service.isClientSide()) { // No need to send services deployed through modules or client side services
                    excludeSG = true;
                    break;
                }
            }

            //TODO: Exclude all services loaded from modules. How to handle data services etc.?
            if (!excludeSG) {
                serviceGroupNames.add(serviceGroup.getServiceGroupName());
            }
        }
        this.serviceGroupNames =
                (String[]) serviceGroupNames.toArray(new String[serviceGroupNames.size()]);
    }
 

import org.apache.axis2.clustering.ClusteringFault; //导入依赖的package包/类
public void execute(ConfigurationContext configContext) throws ClusteringFault {
    ServiceGroupContext sgCtx =
            configContext.getServiceGroupContext(serviceGroupContextId);

    // If the ServiceGroupContext is not found, create it
    if (sgCtx == null) {
        AxisServiceGroup axisServiceGroup =
                configContext.getAxisConfiguration().getServiceGroup(serviceGroupName);
        if(axisServiceGroup == null){
            return;
        }
        sgCtx = new ServiceGroupContext(configContext, axisServiceGroup);
        sgCtx.setId(serviceGroupContextId);
        configContext.addServiceGroupContextIntoSoapSessionTable(sgCtx);  // TODO: Check this
    }
    if (log.isDebugEnabled()) {
        log.debug("Gonna update SG prop in " + serviceGroupContextId + "===" + sgCtx);
    }
    propertyUpdater.updateProperties(sgCtx);
}
 

import org.apache.axis2.clustering.ClusteringFault; //导入依赖的package包/类
public Serializable replyRequest(Serializable msg, Member invoker) {
    if (log.isDebugEnabled()) {
        log.debug("RPC request received by RpcMessagingHandler");
    }
    if (msg instanceof ClusteringMessage) {
        ClusteringMessage clusteringMsg = (ClusteringMessage) msg;
        try {
            clusteringMsg.execute(configurationContext);
        } catch (ClusteringFault e) {
            String errMsg = "Cannot handle RPC message";
            log.error(errMsg, e);
            throw new RemoteProcessException(errMsg, e);
        }
        return clusteringMsg.getResponse();
    } else {
        throw new IllegalArgumentException("Invalid RPC message of type " + msg.getClass() +
                                           " received");
    }
}
 

import org.apache.axis2.clustering.ClusteringFault; //导入依赖的package包/类
public void sendToSelf(ClusteringCommand msg) throws ClusteringFault {
    if (channel == null) {
        return;
    }
    try {
        channel.send(new Member[]{channel.getLocalMember(true)},
                     toByteMessage(msg),
                     Channel.SEND_OPTIONS_USE_ACK |
                     Channel.SEND_OPTIONS_BYTE_MESSAGE);
        if (log.isDebugEnabled()) {
            log.debug("Sent " + msg + " to self");
        }
    } catch (Exception e) {
        throw new ClusteringFault(e);
    }
}
 

import org.apache.axis2.clustering.ClusteringFault; //导入依赖的package包/类
/**
 * 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;
}
 

import org.apache.axis2.clustering.ClusteringFault; //导入依赖的package包/类
/**
 * 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

    // 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;
    }

    // start sending messages in a new thread, since we don't want to block the main thread.
    Thread th = new Thread(new ClusterInfoRetriever(members, sentMembersList, command));
    th.start();
}
 

import org.apache.axis2.clustering.ClusteringFault; //导入依赖的package包/类
/**
 * 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");
}
 

import org.apache.axis2.clustering.ClusteringFault; //导入依赖的package包/类
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);*/
    }
 

import org.apache.axis2.clustering.ClusteringFault; //导入依赖的package包/类
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);*/
    }
 

import org.apache.axis2.clustering.ClusteringFault; //导入依赖的package包/类
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);*/
    }
 

import org.apache.axis2.clustering.ClusteringFault; //导入依赖的package包/类
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);*/
    }
 

import org.apache.axis2.clustering.ClusteringFault; //导入依赖的package包/类
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);*/
    }
 

import org.apache.axis2.clustering.ClusteringFault; //导入依赖的package包/类
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);*/
    }