| /* |
| * 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; |
| |
| import flex.management.runtime.messaging.MessageDestinationControl; |
| import flex.management.runtime.messaging.services.messaging.SubscriptionManagerControl; |
| import flex.management.runtime.messaging.services.messaging.ThrottleManagerControl; |
| import flex.messaging.config.ConfigurationConstants; |
| import flex.messaging.config.ConfigurationException; |
| import flex.messaging.config.DestinationSettings; |
| import flex.messaging.config.ThrottleSettings; |
| import flex.messaging.config.ConfigMap; |
| import flex.messaging.config.NetworkSettings; |
| import flex.messaging.config.ServerSettings; |
| import flex.messaging.config.ThrottleSettings.Policy; |
| import flex.messaging.log.LogCategories; |
| import flex.messaging.services.MessageService; |
| import flex.messaging.services.Service; |
| import flex.messaging.services.messaging.SubscriptionManager; |
| import flex.messaging.services.messaging.RemoteSubscriptionManager; |
| import flex.messaging.services.messaging.ThrottleManager; |
| import flex.messaging.services.messaging.MessagingConstants; |
| import flex.messaging.util.ClassUtil; |
| |
| /** |
| * A logical reference to a MessageDestination. |
| */ |
| public class MessageDestination extends FactoryDestination { |
| static final long serialVersionUID = -2016911808141319012L; |
| |
| /** |
| * Log category for <code>MessageDestination</code>. |
| */ |
| public static final String LOG_CATEGORY = LogCategories.SERVICE_MESSAGE; |
| |
| // Errors |
| private static final int UNSUPPORTED_POLICY = 10124; |
| |
| // Destination properties |
| private transient ServerSettings serverSettings; |
| |
| // Destination internal |
| private transient SubscriptionManager subscriptionManager; |
| private transient RemoteSubscriptionManager remoteSubscriptionManager; |
| private transient ThrottleManager throttleManager; |
| |
| private transient MessageDestinationControl controller; |
| |
| //-------------------------------------------------------------------------- |
| // |
| // Constructor |
| // |
| //-------------------------------------------------------------------------- |
| |
| /** |
| * Constructs an unmanaged <code>MessageDestination</code> instance. |
| */ |
| public MessageDestination() { |
| this(false); |
| } |
| |
| /** |
| * Constructs a <code>MessageDestination</code> with the indicated management. |
| * |
| * @param enableManagement <code>true</code> if the <code>MessageDestination</code> |
| * is manageable; otherwise <code>false</code>. |
| */ |
| public MessageDestination(boolean enableManagement) { |
| super(enableManagement); |
| |
| serverSettings = new ServerSettings(); |
| |
| // Managers |
| subscriptionManager = new SubscriptionManager(this); |
| remoteSubscriptionManager = new RemoteSubscriptionManager(this); |
| } |
| |
| //-------------------------------------------------------------------------- |
| // |
| // Initialize, validate, start, and stop methods. |
| // |
| //-------------------------------------------------------------------------- |
| |
| /** |
| * Initializes the <code>MessageDestination</code> with the properties. |
| * If subclasses override, they must call <code>super.initialize()</code>. |
| * |
| * @param id The id of the destination. |
| * @param properties Properties for the <code>MessageDestination</code>. |
| */ |
| @Override |
| public void initialize(String id, ConfigMap properties) { |
| super.initialize(id, properties); |
| |
| if (properties == null || properties.size() == 0) |
| return; |
| |
| // Network properties |
| network(properties); |
| |
| // Server properties |
| server(properties); |
| } |
| |
| /** |
| * Sets up the throttle manager before it starts. |
| */ |
| @Override |
| public void start() { |
| // Create the throttle manager, only if needed. |
| if (networkSettings.getThrottleSettings() != null) { |
| ThrottleSettings settings = networkSettings.getThrottleSettings(); |
| if (settings.isClientThrottleEnabled() || settings.isDestinationThrottleEnabled()) { |
| settings.setDestinationName(getId()); |
| throttleManager = createThrottleManager(); |
| throttleManager.setThrottleSettings(settings); |
| throttleManager.start(); |
| } |
| } |
| super.start(); |
| } |
| |
| /** |
| * Stops the subscription, remote subscription, and throttle managers and |
| * then calls super class's stop. |
| */ |
| @Override |
| public void stop() { |
| if (isStarted()) { |
| subscriptionManager.stop(); |
| remoteSubscriptionManager.stop(); |
| if (throttleManager != null) |
| throttleManager.stop(); |
| } |
| super.stop(); |
| } |
| |
| //-------------------------------------------------------------------------- |
| // |
| // Public Getters and Setters for Destination properties |
| // |
| //-------------------------------------------------------------------------- |
| |
| /** |
| * Sets the <code>NetworkSettings</code> of the <code>MessageDestination</code>. |
| * |
| * @param networkSettings The <code>NetworkSettings</code> of the <code>MessageDestination</code> |
| */ |
| @Override |
| public void setNetworkSettings(NetworkSettings networkSettings) { |
| super.setNetworkSettings(networkSettings); |
| |
| // Set the subscription manager settings if needed. |
| if (networkSettings.getSubscriptionTimeoutMinutes() > 0) { |
| long subscriptionTimeoutMillis = networkSettings.getSubscriptionTimeoutMinutes() * 60L * 1000L; // Convert to millis. |
| subscriptionManager.setSubscriptionTimeoutMillis(subscriptionTimeoutMillis); |
| } |
| } |
| |
| /** |
| * Returns the <code>ServerSettings</code> of the <code>MessageDestination</code>. |
| * |
| * @return The <code>ServerSettings</code> of the <code>MessageDestination</code>. |
| */ |
| public ServerSettings getServerSettings() { |
| return serverSettings; |
| } |
| |
| /** |
| * Sets the <code>ServerSettings</code> of the <code>MessageDestination</code>. |
| * |
| * @param serverSettings The <code>ServerSettings</code> of the <code>MessageDestination</code> |
| */ |
| public void setServerSettings(ServerSettings serverSettings) { |
| this.serverSettings = serverSettings; |
| } |
| |
| /** |
| * Casts the <code>Service</code> into <code>MessageService</code> |
| * and calls super.setService. |
| * |
| * @param service The <code>Service</code> managing this <code>Destination</code>. |
| */ |
| @Override |
| public void setService(Service service) { |
| MessageService messageService = (MessageService) service; |
| super.setService(messageService); |
| } |
| |
| //-------------------------------------------------------------------------- |
| // |
| // Other Public Methods |
| // |
| //-------------------------------------------------------------------------- |
| |
| /** |
| * Returns a <tt>ConfigMap</tt> of destination properties that the client |
| * needs. This includes properties from <code>super{@link #describeDestination(boolean)}</code> |
| * and it also includes outbound throttling policy that the edge server might need. |
| * |
| * @param onlyReliable Determines whether only reliable destination configuration should be returned. |
| * @return A <tt>ConfigMap</tt> of destination properties that the client needs. |
| */ |
| @Override |
| public ConfigMap describeDestination(boolean onlyReliable) { |
| ConfigMap destinationConfig = super.describeDestination(onlyReliable); |
| if (destinationConfig == null) |
| return null; |
| |
| if (throttleManager == null) |
| return destinationConfig; |
| |
| Policy outboundPolicy = throttleManager.getOutboundPolicy(); |
| if (outboundPolicy == null || outboundPolicy == Policy.NONE) |
| return destinationConfig; |
| |
| // Add the outbound throttle policy to network properties section as appropriate. |
| ConfigMap properties = destinationConfig.getPropertyAsMap(ConfigurationConstants.PROPERTIES_ELEMENT, null); |
| if (properties == null) { |
| properties = new ConfigMap(); |
| destinationConfig.addProperty(ConfigurationConstants.PROPERTIES_ELEMENT, properties); |
| } |
| |
| ConfigMap network = properties.getPropertyAsMap(NetworkSettings.NETWORK_ELEMENT, null); |
| if (network == null) { |
| network = new ConfigMap(); |
| properties.addProperty(NetworkSettings.NETWORK_ELEMENT, network); |
| } |
| |
| ConfigMap throttleOutbound = new ConfigMap(); |
| throttleOutbound.addProperty(ThrottleSettings.ELEMENT_POLICY, throttleManager.getOutboundPolicy().toString()); |
| network.addProperty(ThrottleSettings.ELEMENT_OUTBOUND, throttleOutbound); |
| |
| return destinationConfig; |
| } |
| |
| |
| public SubscriptionManager getSubscriptionManager() { |
| return subscriptionManager; |
| } |
| |
| |
| public RemoteSubscriptionManager getRemoteSubscriptionManager() { |
| return remoteSubscriptionManager; |
| } |
| |
| |
| public ThrottleManager getThrottleManager() { |
| return throttleManager; |
| } |
| |
| |
| @Override |
| public boolean equals(Object o) { |
| if (o instanceof Destination) { |
| Destination d = (Destination) o; |
| String serviceType1 = d.getServiceType(); |
| String serviceType2 = getServiceType(); |
| if ((serviceType1 == null && serviceType2 == null) || (serviceType1 != null && serviceType1.equals(serviceType2))) { |
| String id1 = d.getId(); |
| String id2 = getId(); |
| if ((id1 == null && id2 == null) || (id1 != null && id1.equals(id2))) |
| return true; |
| } |
| } |
| return false; |
| } |
| |
| |
| @Override |
| public int hashCode() { |
| return (getServiceType() == null ? 0 : getServiceType().hashCode()) * 100003 + |
| (getId() == null ? 0 : getId().hashCode()); |
| } |
| |
| |
| @Override |
| public String toString() { |
| return getServiceType() + "#" + getId(); |
| } |
| |
| //-------------------------------------------------------------------------- |
| // |
| // Protected/Private Methods |
| // |
| //-------------------------------------------------------------------------- |
| |
| protected ThrottleManager createThrottleManager() { |
| Service service = getService(); |
| if (service == null || service.getMessageBroker() == null) |
| return new ThrottleManager(); // Return the default. |
| |
| try { |
| Class<? extends ThrottleManager> throttleManagerClass = service.getMessageBroker().getThrottleManagerClass(); |
| Object instance = ClassUtil.createDefaultInstance(throttleManagerClass, null); |
| if (instance instanceof ThrottleManager) |
| return (ThrottleManager) instance; |
| } catch (Throwable t) { |
| // NOWARN |
| } |
| |
| return new ThrottleManager(); // Return the default. |
| } |
| |
| protected void network(ConfigMap properties) { |
| ConfigMap network = properties.getPropertyAsMap(NetworkSettings.NETWORK_ELEMENT, null); |
| if (network == null) |
| return; |
| |
| // Get implementation specific network settings, including subclasses! |
| NetworkSettings ns = getNetworkSettings(); |
| |
| // Subscriber timeout; first check for subscription-timeout-minutes and fallback to legacy session-timeout. |
| int useLegacyPropertyToken = -999999; |
| int subscriptionTimeoutMinutes = network.getPropertyAsInt(NetworkSettings.SUBSCRIPTION_TIMEOUT_MINUTES, useLegacyPropertyToken); |
| if (subscriptionTimeoutMinutes == useLegacyPropertyToken) |
| subscriptionTimeoutMinutes = network.getPropertyAsInt(NetworkSettings.SESSION_TIMEOUT, NetworkSettings.DEFAULT_TIMEOUT); |
| ns.setSubscriptionTimeoutMinutes(subscriptionTimeoutMinutes); |
| |
| // Throttle Settings |
| ThrottleSettings ts = ns.getThrottleSettings(); |
| ts.setDestinationName(getId()); |
| throttle(ts, network); |
| |
| setNetworkSettings(ns); |
| } |
| |
| protected void throttle(ThrottleSettings ts, ConfigMap network) { |
| ConfigMap inbound = network.getPropertyAsMap(ThrottleSettings.ELEMENT_INBOUND, null); |
| if (inbound != null) { |
| ThrottleSettings.Policy policy = getPolicyFromThrottleSettings(inbound); |
| ts.setInboundPolicy(policy); |
| int destFreq = inbound.getPropertyAsInt(ThrottleSettings.ELEMENT_DEST_FREQ, 0); |
| ts.setIncomingDestinationFrequency(destFreq); |
| int clientFreq = inbound.getPropertyAsInt(ThrottleSettings.ELEMENT_CLIENT_FREQ, 0); |
| ts.setIncomingClientFrequency(clientFreq); |
| } |
| |
| ConfigMap outbound = network.getPropertyAsMap(ThrottleSettings.ELEMENT_OUTBOUND, null); |
| if (outbound != null) { |
| ThrottleSettings.Policy policy = getPolicyFromThrottleSettings(outbound); |
| ts.setOutboundPolicy(policy); |
| int destFreq = outbound.getPropertyAsInt(ThrottleSettings.ELEMENT_DEST_FREQ, 0); |
| ts.setOutgoingDestinationFrequency(destFreq); |
| int clientFreq = outbound.getPropertyAsInt(ThrottleSettings.ELEMENT_CLIENT_FREQ, 0); |
| ts.setOutgoingClientFrequency(clientFreq); |
| } |
| } |
| |
| private ThrottleSettings.Policy getPolicyFromThrottleSettings(ConfigMap settings) { |
| String policyString = settings.getPropertyAsString(ThrottleSettings.ELEMENT_POLICY, null); |
| ThrottleSettings.Policy policy = ThrottleSettings.Policy.NONE; |
| if (policyString == null) |
| return policy; |
| try { |
| policy = ThrottleSettings.parsePolicy(policyString); |
| } catch (ConfigurationException exception) { |
| ConfigurationException ce = new ConfigurationException(); |
| ce.setMessage(UNSUPPORTED_POLICY, new Object[]{getId(), policyString}); |
| throw ce; |
| } |
| return policy; |
| } |
| |
| protected void server(ConfigMap properties) { |
| ConfigMap server = properties.getPropertyAsMap(DestinationSettings.SERVER_ELEMENT, null); |
| if (server == null) |
| return; |
| |
| long ttl = server.getPropertyAsLong(MessagingConstants.TIME_TO_LIVE_ELEMENT, -1); |
| serverSettings.setMessageTTL(ttl); |
| |
| boolean durable = server.getPropertyAsBoolean(MessagingConstants.IS_DURABLE_ELEMENT, false); |
| serverSettings.setDurable(durable); |
| |
| boolean allowSubtopics = server.getPropertyAsBoolean(MessagingConstants.ALLOW_SUBTOPICS_ELEMENT, false); |
| serverSettings.setAllowSubtopics(allowSubtopics); |
| |
| boolean disallowWildcardSubtopics = server.getPropertyAsBoolean(MessagingConstants.DISALLOW_WILDCARD_SUBTOPICS_ELEMENT, false); |
| serverSettings.setDisallowWildcardSubtopics(disallowWildcardSubtopics); |
| |
| int priority = server.getPropertyAsInt(MessagingConstants.MESSAGE_PRIORITY, -1); |
| if (priority != -1) |
| serverSettings.setPriority(priority); |
| |
| String subtopicSeparator = server.getPropertyAsString(MessagingConstants.SUBTOPIC_SEPARATOR_ELEMENT, MessagingConstants.DEFAULT_SUBTOPIC_SEPARATOR); |
| serverSettings.setSubtopicSeparator(subtopicSeparator); |
| |
| String routingMode = server.getPropertyAsString(MessagingConstants.CLUSTER_MESSAGE_ROUTING, "server-to-server"); |
| serverSettings.setBroadcastRoutingMode(routingMode); |
| } |
| |
| /** |
| * Returns the log category of the <code>MessageDestination</code>. |
| * |
| * @return The log category of the component. |
| */ |
| @Override |
| protected String getLogCategory() { |
| return LOG_CATEGORY; |
| } |
| |
| /** |
| * Invoked automatically to allow the <code>MessageDestination</code> to setup its corresponding |
| * MBean control. |
| * |
| * @param service The <code>Service</code> that manages this <code>MessageDestination</code>. |
| */ |
| @Override |
| protected void setupDestinationControl(Service service) { |
| controller = new MessageDestinationControl(this, service.getControl()); |
| controller.register(); |
| setControl(controller); |
| setupThrottleManagerControl(controller); |
| setupSubscriptionManagerControl(controller); |
| } |
| |
| protected void setupThrottleManagerControl(MessageDestinationControl destinationControl) { |
| if (throttleManager != null) { |
| ThrottleManagerControl throttleManagerControl = new ThrottleManagerControl(throttleManager, destinationControl); |
| throttleManagerControl.register(); |
| throttleManager.setControl(throttleManagerControl); |
| throttleManager.setManaged(true); |
| destinationControl.setThrottleManager(throttleManagerControl.getObjectName()); |
| } |
| } |
| |
| private void setupSubscriptionManagerControl(MessageDestinationControl destinationControl) { |
| SubscriptionManagerControl subscriptionManagerControl = new SubscriptionManagerControl(getSubscriptionManager(), destinationControl); |
| subscriptionManagerControl.register(); |
| getSubscriptionManager().setControl(subscriptionManagerControl); |
| getSubscriptionManager().setManaged(true); |
| destinationControl.setSubscriptionManager(subscriptionManagerControl.getObjectName()); |
| } |
| } |
