modules/core/src/flex/messaging/MessageDestination.java - flex-blazeds - Git at Google

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

     /** @exclude **/
     @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;
     }

     /** @exclude **/
     @Override
     public int hashCode()
     {
         return (getServiceType() == null ? 0 : getServiceType().hashCode()) * 100003 +
             (getId() == null ? 0 : getId().hashCode());
     }

     /** @exclude **/
     @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());
     }
 }
	/*
	* 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;
	}

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

	/ @exclude /
	@Override
	public int hashCode()
	{
	return (getServiceType() == null ? 0 : getServiceType().hashCode()) * 100003 +
	(getId() == null ? 0 : getId().hashCode());
	}

	/ @exclude /
	@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());
	}
	}