| /* |
| * 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 flex.messaging.cluster; |
| |
| import java.io.InputStream; |
| import java.lang.reflect.Constructor; |
| import java.util.HashMap; |
| import java.util.Iterator; |
| import java.util.LinkedHashMap; |
| import java.util.List; |
| import java.util.Arrays; |
| import java.util.ArrayList; |
| import java.util.Map; |
| import java.util.Map.Entry; |
| import java.util.Collections; |
| import java.util.TreeSet; |
| |
| import javax.xml.parsers.DocumentBuilder; |
| import javax.xml.parsers.DocumentBuilderFactory; |
| |
| import org.w3c.dom.Document; |
| import org.w3c.dom.Element; |
| |
| import flex.messaging.Destination; |
| import flex.messaging.MessageBroker; |
| import flex.messaging.config.ClusterSettings; |
| import flex.messaging.config.ConfigMap; |
| import flex.messaging.endpoints.Endpoint; |
| import flex.messaging.util.ClassUtil; |
| |
| /** |
| * The manager of all clusters defined in services-config.xml, and the broker |
| * for the clusters created for clustered destinations. |
| */ |
| public class ClusterManager { |
| /** |
| * Supported operations. |
| */ |
| public static final String OPERATION_ADD_ENDPOINT_FOR_CHANNEL = "addEndpointForChannel"; |
| public static final String OPERATION_SEND_ENDPOINT_URL = "sendEndpointUrl"; |
| public static final String OPERATION_RECEIVE_ENDPOINT_URL = "receiveEndpointUrl"; |
| |
| public static final String OPERATION_PUSH_MESSAGE_FROM_PEER = "pushMessageFromPeer"; |
| public static final String OPERATION_PEER_SYNC_AND_PUSH = "peerSyncAndPush"; |
| public static final String OPERATION_REQUEST_ADAPTER_STATE = "requestAdapterState"; |
| public static final String OPERATION_RECEIVE_ADAPTER_STATE = "receiveAdapterState"; |
| public static final String OPERATION_SEND_SUBSCRIPTIONS = "sendSubscriptions"; |
| public static final String OPERATION_RECEIVE_SUBSCRIPTIONS = "receiveSubscriptions"; |
| public static final String OPERATION_SUBSCRIBE_FROM_PEER = "subscribeFromPeer"; |
| public static final String OPERATION_PUSH_MESSAGE_FROM_PEER_TO_PEER = "pushMessageFromPeerToPeer"; |
| public static final String OPERATION_PEER_SYNC_AND_PUSH_ONE_TO_PEER = "peerSyncAndPushOneToPeer"; |
| |
| /** |
| * A link to the MessageBroker. |
| */ |
| private MessageBroker broker; |
| |
| /** |
| * A mapping between the cluster ids and the Cluster instances. |
| * name=clusterId value=clusterInstance |
| */ |
| private LinkedHashMap<String, Cluster> clusters; |
| |
| /** |
| * A mapping between destinations and the Cluster instances. |
| */ |
| private Map<String, Cluster> clustersForDestination; |
| |
| /** |
| * A mapping between cluster ids and their configuration files. |
| * name=clusterId value=propsFile |
| */ |
| private Map<String, Element> clusterConfig; |
| |
| /** |
| * A mapping between cluster ids and ClusterSettings instances. |
| * name=clusterId value=ClusterSettings |
| */ |
| private Map<String, ClusterSettings> clusterSettings; |
| |
| /** |
| * A mapped between destinations and a boolean representing |
| * whether or not the backend for the destination is shared. |
| */ |
| private Map<String, Boolean> backendSharedForDestination; |
| |
| /** |
| * The default cluster when the cluster id for the destination |
| * is unspecified. |
| */ |
| private Cluster defaultCluster; |
| |
| /** |
| * The id of the default cluster. |
| */ |
| private String defaultClusterId; |
| |
| /** |
| * The manager of all clusters defined in services-config.xml, and the broker |
| * for the clusters created for clustered destinations. This class provides |
| * an entry point and abstraction to the logical cluster implementation as |
| * well as the specific cluster implementation. |
| * |
| * @param broker the message broker which uses the cluster manager |
| */ |
| public ClusterManager(MessageBroker broker) { |
| this.broker = broker; |
| clusters = new LinkedHashMap<String, Cluster>(); |
| clusterConfig = new HashMap<String, Element>(); |
| clusterSettings = new HashMap<String, ClusterSettings>(); |
| clustersForDestination = new HashMap<String, Cluster>(); |
| backendSharedForDestination = new HashMap<String, Boolean>(); |
| } |
| |
| /** |
| * The MessageBroker for this cluster. |
| * |
| * @return The defined MessageBroker. |
| */ |
| public MessageBroker getMessageBroker() { |
| return broker; |
| } |
| |
| /** |
| * The default cluster when the cluster id for the destination |
| * is unspecified. |
| * |
| * @return Cluster the default Cluster to use |
| */ |
| public Cluster getDefaultCluster() { |
| return defaultCluster; |
| } |
| |
| /** |
| * The id of the default cluster. |
| * |
| * @return String the default cluster ID |
| */ |
| public String getDefaultClusterId() { |
| return defaultClusterId; |
| } |
| |
| /** |
| * Invoke an endpoint operation across the cluster. |
| * <p> |
| * NOTE: Endpoints don't reference a specific cluster so the default cluster is used for the broadcast. |
| * If no default cluster is defined the operation is broadcast over all defined clusters. |
| * </p> |
| * |
| * @param endpointId The id of the remote endpoint across the cluster to invoke an operation on. |
| * @param operationName The name of the operation to invoke. |
| * @param params The arguments to use for operation invocation. |
| */ |
| public void invokeEndpointOperation(String endpointId, String operationName, Object[] params) { |
| Object[] arguments = new Object[2 + params.length]; |
| arguments[0] = endpointId; |
| arguments[1] = operationName; |
| int n = params.length; |
| for (int i = 2, j = 0; j < n; ++i, ++j) |
| arguments[i] = params[j]; |
| |
| if (defaultCluster != null) { |
| defaultCluster.broadcastServiceOperation(operationName, arguments); |
| } else { |
| for (Cluster cluster : clusters.values()) |
| cluster.broadcastServiceOperation(operationName, arguments); |
| } |
| } |
| |
| /** |
| * Invoke an endpoint operation on a specific peer within the cluster. |
| * <p> |
| * NOTE: Endpoints don't reference a specific cluster so the default cluster is used for the broadcast. |
| * If no default cluster is defined the operation is broadcast over all defined clusters. |
| * </p> |
| * |
| * @param endpointId The id of the remote endpoint across the cluster to invoke an operation on. |
| * @param operationName The name of the operation to invoke. |
| * @param params The arguments to use for operation invocation. |
| * @param targetAddress The peer node that the operation should be invoked on. |
| */ |
| public void invokePeerToPeerEndpointOperation(String endpointId, String operationName, Object[] params, Object targetAddress) { |
| Object[] arguments = new Object[2 + params.length]; |
| arguments[0] = endpointId; |
| arguments[1] = operationName; |
| int n = params.length; |
| for (int i = 2, j = 0; j < n; ++i, ++j) |
| arguments[i] = params[j]; |
| |
| if (defaultCluster != null) { |
| defaultCluster.sendPointToPointServiceOperation(operationName, arguments, targetAddress); |
| } else { |
| for (Cluster cluster : clusters.values()) { |
| cluster.sendPointToPointServiceOperation(operationName, arguments, targetAddress); |
| } |
| } |
| } |
| |
| /** |
| * Invoke a service-related operation, which usually includes a Message as a method parameter. This method |
| * allows a local service to process a Message and then send the Message to the services on all peer nodes |
| * so that they may perform the same processing. Invoke the service operation for the cluster, identified by |
| * serviceType and destinationName. |
| * |
| * @param serviceType The name for the service for this destination. |
| * @param destinationName The name of the destination. |
| * @param operationName The name of the service operation to invoke. |
| * @param params Parameters needed for the service operation. |
| */ |
| public void invokeServiceOperation(String serviceType, String destinationName, |
| String operationName, Object[] params) { |
| Cluster c = getCluster(serviceType, destinationName); |
| ArrayList newParams = new ArrayList(Arrays.asList(params)); |
| newParams.add(0, serviceType); |
| newParams.add(1, destinationName); |
| c.broadcastServiceOperation(operationName, newParams.toArray()); |
| } |
| |
| /** |
| * Send a service-related operation in point-to-point fashion to one and only one member of the cluster. |
| * This is similar to the invokeServiceOperation except that this invocation is sent to the node, |
| * identified by targetAddress. |
| * |
| * @param serviceType The name for the service for this destination. |
| * @param destinationName The name of the destination. |
| * @param operationName The name of the service operation to invoke. |
| * @param params Parameters needed for the service operation. |
| * @param targetAddress The node that the operation should be passed to. |
| */ |
| public void invokePeerToPeerOperation(String serviceType, String destinationName, |
| String operationName, Object[] params, Object targetAddress) { |
| Cluster c = getCluster(serviceType, destinationName); |
| ArrayList newParams = new ArrayList(Arrays.asList(params)); |
| newParams.add(0, serviceType); |
| newParams.add(1, destinationName); |
| c.sendPointToPointServiceOperation(operationName, newParams.toArray(), targetAddress); |
| } |
| |
| /** |
| * Determines whether the given destination is clustered. |
| * |
| * @param serviceType The name for the service for this destination. |
| * @param destinationName The name of the destination. |
| * @return Whether the destination is a clustered destination. |
| */ |
| public boolean isDestinationClustered(String serviceType, String destinationName) { |
| return getCluster(serviceType, destinationName) != null; |
| } |
| |
| /** |
| * Checks whether the give destination is configured for a shared backend. |
| * |
| * @param serviceType The name of the service for this destination. |
| * @param destinationName The name of the destination. |
| * @return Whether the destination is configured for shared backend. |
| */ |
| public boolean isBackendShared(String serviceType, String destinationName) { |
| String destKey = Cluster.getClusterDestinationKey(serviceType, destinationName); |
| Boolean shared = backendSharedForDestination.get(destKey); |
| return shared != null ? shared.booleanValue() : false; |
| } |
| |
| /** |
| * Retrieves a list of cluster nodes for the given cluster. |
| * |
| * @param serviceType The name of the service for the clustered destination. |
| * @param destinationName The name of the destination. |
| * @return List of cluster nodes for the given cluster. |
| */ |
| public List getClusterMemberAddresses(String serviceType, String destinationName) { |
| Cluster c = getCluster(serviceType, destinationName); |
| return c != null ? c.getMemberAddresses() : Collections.EMPTY_LIST; |
| } |
| |
| /** |
| * Used for targeted endpoint operation invocations across the cluster. |
| * If a default cluster is defined, its list of member addresses is returned. |
| * Otherwise, a de-duped list of all member addresses from all registered clusters is returned. |
| * |
| * @return The list of cluster nodes that endpoint operation invocations can be issued against. |
| */ |
| public List getClusterMemberAddresses() { |
| if (defaultCluster != null) |
| return defaultCluster.getMemberAddresses(); |
| |
| TreeSet uniqueAddresses = new TreeSet(); |
| for (Cluster cluster : clusters.values()) |
| uniqueAddresses.addAll(cluster.getMemberAddresses()); |
| |
| return new ArrayList(uniqueAddresses); |
| } |
| |
| /** |
| * Find the properties file in the given cluster settings. Read the XML based |
| * cluster configuration file and save the settings and configuration for the |
| * given cluster for retrieval later. |
| * |
| * @param settings The cluster settings for a specific cluster. |
| */ |
| public void prepareCluster(ClusterSettings settings) { |
| String propsFileName = settings.getPropsFileName(); |
| |
| checkForNullPropertiesFile(settings.getClusterName(), propsFileName); |
| |
| InputStream propsFile = resolveInternalPath(propsFileName); |
| |
| if (propsFile == null) |
| propsFile = resolveExternalPath(propsFileName); |
| |
| if (propsFile == null) |
| throwClusterException(10208, new Object[]{propsFileName}, null); |
| |
| try { |
| DocumentBuilderFactory factory = DocumentBuilderFactory.newInstance(); |
| factory.setNamespaceAware(false); |
| factory.setValidating(false); |
| DocumentBuilder builder = factory.newDocumentBuilder(); |
| Document doc = builder.parse(propsFile); |
| if (settings.isDefault()) |
| defaultClusterId = settings.getClusterName(); |
| clusterConfig.put(settings.getClusterName(), doc.getDocumentElement()); |
| clusterSettings.put(settings.getClusterName(), settings); |
| } catch (Exception ex) { |
| throwClusterException(10213, new Object[]{propsFileName}, ex); |
| } |
| } |
| |
| /** |
| * Retrieve the local address for the specified clustered destination. |
| * |
| * @param serviceType The service type of the clustered destination. |
| * @param destinationName The name of the clustered destination. |
| * @return The local address of the clustered destination. |
| */ |
| public Object getLocalAddress(String serviceType, String destinationName) { |
| Cluster c = getCluster(serviceType, destinationName); |
| return c != null ? c.getLocalAddress() : null; |
| } |
| |
| /** |
| * Retrieve the local address for the default cluster or if no default cluster is defined |
| * return the local address derived from the first cluster of any defined. |
| * |
| * @return The local address for this cluster node, or <code>null</code> if this node |
| * is not a member of any cluster. |
| */ |
| public Object getLocalAddress() { |
| if (defaultCluster != null) |
| return defaultCluster.getLocalAddress(); |
| |
| // Else, use first defined cluster. |
| for (Entry<String, Cluster> entry : clusters.entrySet()) |
| return entry.getValue().getLocalAddress(); |
| |
| return null; // No cluster defined. |
| } |
| |
| /** |
| * Find the cluster for the specified cluster id. |
| * |
| * @param clusterId the cluster ID |
| * @return The cluster identified by the given id. |
| */ |
| public Cluster getClusterById(String clusterId) { |
| return clusters.get(clusterId); |
| } |
| |
| /** |
| * Find the cluster identified by the service type and destination name. |
| * |
| * @param serviceType The service type of the clustered destination. |
| * @param destinationName The name of the clustered destination. |
| * @return The cluster identified by the serviec type and destination naem. |
| */ |
| public Cluster getCluster(String serviceType, String destinationName) { |
| Cluster cluster = null; |
| try { |
| String destKey = Cluster.getClusterDestinationKey(serviceType, destinationName); |
| |
| cluster = clustersForDestination.get(destKey); |
| |
| if (cluster == null) |
| cluster = defaultCluster; |
| } catch (NoClassDefFoundError nex) { |
| ClusterException cx = new ClusterException(); |
| cx.setMessage(10202, new Object[]{destinationName}); |
| cx.setRootCause(nex); |
| throw cx; |
| } |
| return cluster; |
| } |
| |
| /** |
| * Call destroy on each of the managed clusters. |
| */ |
| public void destroyClusters() { |
| for (Iterator<Cluster> iter = clusters.values().iterator(); iter.hasNext(); ) { |
| Cluster cluster = iter.next(); |
| cluster.destroy(); |
| iter.remove(); |
| } |
| } |
| |
| /** |
| * Add the specified destination to the cluster, identitied by clusterId if available. If the cluster |
| * is not currently defined, create the cluster. Also, setup the load balancing urls and shared |
| * backend information for this clustered destination and endpoint. |
| * |
| * @param clusterId The cluster id that this destination wants to be associated with. |
| * @param serviceType The service type for the clustered destination. |
| * @param destinationName The name of the clustered destination. |
| * @param channelId The channel id that should be added to the cluster load balancing. |
| * @param endpointUrl The endpoint url that should be added to the cluster load balancing. |
| * @param endpointPort The endpoint port that should be added to the cluster load balancing. |
| * @param sharedBackend Whether the destination has shared backend set to true or not. |
| */ |
| public void clusterDestinationChannel(String clusterId, String serviceType, String destinationName, |
| String channelId, String endpointUrl, int endpointPort, boolean sharedBackend) { |
| Cluster cluster = getClusterById(clusterId); |
| String destKey = Cluster.getClusterDestinationKey(serviceType, destinationName); |
| if (cluster == null) { |
| if (!clusterConfig.containsKey(clusterId)) { |
| ClusterException cx = new ClusterException(); |
| cx.setMessage(10207, new Object[]{destinationName, clusterId}); |
| throw cx; |
| } |
| cluster = createCluster(clusterId, serviceType, destinationName); |
| } else { |
| clustersForDestination.put(destKey, cluster); |
| } |
| backendSharedForDestination.put(destKey, sharedBackend ? Boolean.TRUE : Boolean.FALSE); |
| |
| if (cluster.getURLLoadBalancing()) |
| cluster.addLocalEndpointForChannel(serviceType, destinationName, |
| channelId, endpointUrl, endpointPort); |
| } |
| |
| /** |
| * Adds the destination to the cluster. The settings for the clustered destination are |
| * available from the <code>Destination</code> object. |
| * |
| * @param destination The destination to be clustered. |
| */ |
| public void clusterDestination(Destination destination) { |
| String clusterId = destination.getNetworkSettings().getClusterId(); |
| if (clusterId == null) |
| clusterId = getDefaultClusterId(); |
| |
| ClusterSettings cls = clusterSettings.get(clusterId); |
| if (cls == null) { |
| ClusterException ce = new ClusterException(); |
| ce.setMessage(10217, new Object[]{destination.getId(), clusterId}); |
| throw ce; |
| } |
| |
| for (String channelId : destination.getChannels()) { |
| Endpoint endpoint = broker.getEndpoint(channelId); |
| String endpointUrl = endpoint.getUrl(); |
| int endpointPort = endpoint.getPort(); |
| |
| // This is only an error if we are using client side url-based load balancing. If |
| // there is a HW load balancer, then we can assume the server.name served up by the |
| // SWF can be used to access the cluster members. With client side load balancing, |
| // the clients need the direct URLs of all of the servers. |
| if (cls.getURLLoadBalancing()) { |
| // Ensure that the endpoint URI does not contain any replacement tokens. |
| int tokenStart = endpointUrl.indexOf('{'); |
| if (tokenStart != -1) { |
| int tokenEnd = endpointUrl.indexOf('}', tokenStart); |
| if (tokenEnd == -1) |
| tokenEnd = endpointUrl.length(); |
| else |
| tokenEnd++; |
| |
| ClusterException ce = new ClusterException(); |
| ce.setMessage(10209, new Object[]{destination.getId(), channelId, endpointUrl.substring(tokenStart, tokenEnd)}); |
| throw ce; |
| } |
| } |
| |
| clusterDestinationChannel(clusterId, destination.getServiceType(), |
| destination.getId(), channelId, endpointUrl, endpointPort, destination.getNetworkSettings().isSharedBackend()); |
| } |
| } |
| |
| /** |
| * Get a list of endpoints for the destination. |
| * |
| * @param serviceType the service type |
| * @param destinationName destination name |
| * @return List the list endpoints that the destination can use |
| */ |
| public List getEndpointsForDestination(String serviceType, String destinationName) { |
| Cluster c = getCluster(serviceType, destinationName); |
| return c != null ? c.getAllEndpoints(serviceType, destinationName) : null; |
| } |
| |
| |
| private void checkForNullPropertiesFile(String clusterName, String propsFileName) { |
| if (propsFileName == null) |
| throwClusterException(10201, new Object[]{clusterName, propsFileName}, null); |
| } |
| |
| /** |
| * Create the cluster based on the cluster settings already available. The cluster |
| * is added to the cluster managers list of clusters indexed by the cluster id. |
| * The cluster is also associated with the specified service type and destination |
| * name. The cluster id is unique across all clusters managed by this cluster |
| * manager. The cluster may be associated with more than one cluster destination. |
| * |
| * @param clusterId The cluster id. |
| * @param serviceType The service type of the clustered destination. |
| * @param destinationName The destination name for the clustered destination. |
| * @return The new cluster. |
| */ |
| private Cluster createCluster(String clusterId, String serviceType, String destinationName) { |
| String destKey = Cluster.getClusterDestinationKey(serviceType, destinationName); |
| Element propsFile = clusterConfig.get(clusterId); |
| ClusterSettings cls = clusterSettings.get(clusterId); |
| Cluster cluster = null; |
| Class clusterClass = ClassUtil.createClass(cls.getImplementationClass()); |
| Constructor clusterConstructor = null; |
| try { |
| clusterConstructor = clusterClass.getConstructor(ClusterManager.class); |
| } catch (Exception e) { |
| ClusterException cx = new ClusterException(); |
| cx.setMessage(10210); |
| cx.setRootCause(e); |
| throw cx; |
| } |
| try { |
| cluster = (Cluster) clusterConstructor.newInstance(this); |
| cluster.setClusterPropertiesFile(propsFile); |
| cluster.setURLLoadBalancing(cls.getURLLoadBalancing()); |
| cluster.initialize(clusterId, cls.getProperties()); |
| } catch (Exception e) { |
| ClusterException cx = new ClusterException(); |
| cx.setMessage(10211); |
| cx.setRootCause(e); |
| throw cx; |
| } |
| clustersForDestination.put(destKey, cluster); |
| clusters.put(clusterId, cluster); |
| |
| if (defaultClusterId != null && defaultClusterId.equals(clusterId)) |
| defaultCluster = cluster; |
| |
| return cluster; |
| } |
| |
| private InputStream resolveExternalPath(String propsFileName) { |
| try { |
| return broker.resolveExternalPath(propsFileName); |
| } catch (Throwable t) { |
| throwClusterException(10208, new Object[]{propsFileName}, t); |
| } |
| return null; |
| } |
| |
| private InputStream resolveInternalPath(String propsFileName) { |
| try { |
| return broker.resolveInternalPath(propsFileName); |
| } catch (Throwable t) { |
| throwClusterException(10208, new Object[]{propsFileName}, t); |
| } |
| return null; |
| } |
| |
| private void throwClusterException(int number, Object[] args, Throwable t) { |
| ClusterException cx = new ClusterException(); |
| cx.setMessage(number, args); |
| if (t != null) |
| cx.setRootCause(t); |
| throw cx; |
| } |
| |
| /** |
| * Return a {@link ConfigMap} describing the clusters that have been added to the cluster manager |
| * |
| * @return a ConfigMap of the clusters |
| */ |
| public ConfigMap describeClusters() { |
| ConfigMap result = new ConfigMap(); |
| for (Entry<String, Cluster> entry : clusters.entrySet()) { |
| Cluster cluster = entry.getValue(); |
| ConfigMap clusterMap = new ConfigMap(); |
| clusterMap.put("id", entry.getKey()); |
| ClusterSettings settings = clusterSettings.get(entry.getKey()); |
| clusterMap.put("properties", settings.getPropsFileName()); |
| if (settings.isDefault()) { |
| clusterMap.put("default", "true"); |
| } |
| clusterMap.put("class", cluster.getClass().getCanonicalName()); |
| |
| result.addProperty("cluster", clusterMap); |
| } |
| return result; |
| } |
| } |
