You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by an...@apache.org on 2014/11/11 19:41:42 UTC
[12/51] [abbrv] [partial] activemq-6 git commit: ACTIVEMQ6-4 - Rename
packages to ActiveMQ
http://git-wip-us.apache.org/repos/asf/activemq-6/blob/23e8edd9/activemq6-jms-server/src/main/java/org/apache/activemq6/jms/server/impl/JMSServerManagerImpl.java
----------------------------------------------------------------------
diff --git a/activemq6-jms-server/src/main/java/org/apache/activemq6/jms/server/impl/JMSServerManagerImpl.java b/activemq6-jms-server/src/main/java/org/apache/activemq6/jms/server/impl/JMSServerManagerImpl.java
new file mode 100644
index 0000000..5ffff6d
--- /dev/null
+++ b/activemq6-jms-server/src/main/java/org/apache/activemq6/jms/server/impl/JMSServerManagerImpl.java
@@ -0,0 +1,1971 @@
+/*
+ * Copyright 2005-2014 Red Hat, Inc.
+ * Red Hat 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 org.apache.activemq6.jms.server.impl;
+
+import javax.naming.Context;
+import javax.naming.InitialContext;
+import javax.naming.NamingException;
+import javax.transaction.xa.Xid;
+
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.activemq6.api.core.DiscoveryGroupConfiguration;
+import org.apache.activemq6.api.core.HornetQException;
+import org.apache.activemq6.api.core.SimpleString;
+import org.apache.activemq6.api.core.TransportConfiguration;
+import org.apache.activemq6.api.core.management.AddressControl;
+import org.apache.activemq6.api.core.management.ResourceNames;
+import org.apache.activemq6.api.jms.HornetQJMSClient;
+import org.apache.activemq6.api.jms.JMSFactoryType;
+import org.apache.activemq6.core.config.Configuration;
+import org.apache.activemq6.core.deployers.DeploymentManager;
+import org.apache.activemq6.core.deployers.impl.FileDeploymentManager;
+import org.apache.activemq6.core.deployers.impl.XmlDeployer;
+import org.apache.activemq6.core.postoffice.Binding;
+import org.apache.activemq6.core.postoffice.BindingType;
+import org.apache.activemq6.core.registry.JndiBindingRegistry;
+import org.apache.activemq6.core.remoting.impl.netty.NettyConnectorFactory;
+import org.apache.activemq6.core.remoting.impl.netty.TransportConstants;
+import org.apache.activemq6.core.security.Role;
+import org.apache.activemq6.core.server.ActivateCallback;
+import org.apache.activemq6.core.server.HornetQServer;
+import org.apache.activemq6.core.server.Queue;
+import org.apache.activemq6.core.server.impl.HornetQServerImpl;
+import org.apache.activemq6.core.server.management.Notification;
+import org.apache.activemq6.core.settings.impl.AddressSettings;
+import org.apache.activemq6.core.transaction.ResourceManager;
+import org.apache.activemq6.core.transaction.Transaction;
+import org.apache.activemq6.core.transaction.TransactionDetail;
+import org.apache.activemq6.jms.client.HornetQConnectionFactory;
+import org.apache.activemq6.jms.client.HornetQDestination;
+import org.apache.activemq6.jms.client.HornetQQueue;
+import org.apache.activemq6.jms.client.HornetQTopic;
+import org.apache.activemq6.jms.client.SelectorTranslator;
+import org.apache.activemq6.jms.persistence.JMSStorageManager;
+import org.apache.activemq6.jms.persistence.config.PersistedConnectionFactory;
+import org.apache.activemq6.jms.persistence.config.PersistedDestination;
+import org.apache.activemq6.jms.persistence.config.PersistedJNDI;
+import org.apache.activemq6.jms.persistence.config.PersistedType;
+import org.apache.activemq6.jms.persistence.impl.journal.JMSJournalStorageManagerImpl;
+import org.apache.activemq6.jms.persistence.impl.nullpm.NullJMSStorageManagerImpl;
+import org.apache.activemq6.jms.server.HornetQJMSServerBundle;
+import org.apache.activemq6.jms.server.HornetQJMSServerLogger;
+import org.apache.activemq6.jms.server.JMSServerManager;
+import org.apache.activemq6.jms.server.config.ConnectionFactoryConfiguration;
+import org.apache.activemq6.jms.server.config.JMSConfiguration;
+import org.apache.activemq6.jms.server.config.JMSQueueConfiguration;
+import org.apache.activemq6.jms.server.config.TopicConfiguration;
+import org.apache.activemq6.jms.server.config.impl.ConnectionFactoryConfigurationImpl;
+import org.apache.activemq6.jms.server.management.JMSManagementService;
+import org.apache.activemq6.jms.server.management.JMSNotificationType;
+import org.apache.activemq6.jms.server.management.impl.JMSManagementServiceImpl;
+import org.apache.activemq6.jms.transaction.JMSTransactionDetail;
+import org.apache.activemq6.spi.core.naming.BindingRegistry;
+import org.apache.activemq6.utils.TimeAndCounterIDGenerator;
+import org.apache.activemq6.utils.TypedProperties;
+import org.apache.activemq6.utils.json.JSONArray;
+import org.apache.activemq6.utils.json.JSONObject;
+
+/**
+ * A Deployer used to create and add to JNDI queues, topics and connection
+ * factories. Typically this would only be used in an app server env.
+ * <p>
+ * JMS Connection Factories and Destinations can be configured either
+ * using configuration files or using a JMSConfiguration object.
+ * <p>
+ * If configuration files are used, JMS resources are redeployed if the
+ * files content is changed.
+ * If a JMSConfiguration object is used, the JMS resources can not be
+ * redeployed.
+ *
+ * @author <a href="ataylor@redhat.com">Andy Taylor</a>
+ * @author <a href="jmesnil@redhat.com">Jeff Mesnil</a>
+ * @author <a href="tim.fox@jboss.com">Tim Fox</a>
+ */
+public class JMSServerManagerImpl implements JMSServerManager, ActivateCallback
+{
+ private static final String REJECT_FILTER = HornetQServerImpl.GENERIC_IGNORED_FILTER;
+
+ private BindingRegistry registry;
+
+ private final Map<String, HornetQQueue> queues = new HashMap<String, HornetQQueue>();
+
+ private final Map<String, HornetQTopic> topics = new HashMap<String, HornetQTopic>();
+
+ private final Map<String, HornetQConnectionFactory> connectionFactories = new HashMap<String, HornetQConnectionFactory>();
+
+ private final Map<String, List<String>> queueJNDI = new HashMap<String, List<String>>();
+
+ private final Map<String, List<String>> topicJNDI = new HashMap<String, List<String>>();
+
+ private final Map<String, List<String>> connectionFactoryJNDI = new HashMap<String, List<String>>();
+
+ // We keep things cached if objects are created while the JMS is not active
+ private final List<Runnable> cachedCommands = new ArrayList<Runnable>();
+
+ private final HornetQServer server;
+
+ private JMSManagementService jmsManagementService;
+
+ private XmlDeployer jmsDeployer;
+
+ private boolean startCalled;
+
+ private boolean active;
+
+ private DeploymentManager deploymentManager;
+
+ private final String configFileName;
+
+ private boolean contextSet;
+
+ private JMSConfiguration config;
+
+ private Configuration coreConfig;
+
+ private JMSStorageManager storage;
+
+ private final Map<String, List<String>> unRecoveredJndi = new HashMap<String, List<String>>();
+
+ public JMSServerManagerImpl(final HornetQServer server) throws Exception
+ {
+ this.server = server;
+
+ this.coreConfig = server.getConfiguration();
+
+ configFileName = null;
+ }
+
+ /**
+ * This constructor is used by the Application Server's integration
+ *
+ * @param server
+ * @param registry
+ * @throws Exception
+ */
+ public JMSServerManagerImpl(final HornetQServer server, final BindingRegistry registry) throws Exception
+ {
+ this.server = server;
+
+ this.coreConfig = server.getConfiguration();
+
+ configFileName = null;
+
+ this.registry = registry;
+ }
+
+ public JMSServerManagerImpl(final HornetQServer server, final String configFileName) throws Exception
+ {
+ this.server = server;
+
+ this.coreConfig = server.getConfiguration();
+
+ this.configFileName = configFileName;
+ }
+
+ public JMSServerManagerImpl(final HornetQServer server, final JMSConfiguration configuration) throws Exception
+ {
+ this.server = server;
+
+ this.coreConfig = server.getConfiguration();
+
+ configFileName = null;
+
+ config = configuration;
+ }
+
+ /**
+ * Unused
+ */
+ @Deprecated
+ public JMSServerManagerImpl(HornetQServer server, String configFilename, JMSStorageManager storageManager)
+ {
+ this.server = server;
+
+ configFileName = null;
+
+ storage = storageManager;
+ }
+
+ // ActivateCallback implementation -------------------------------------
+
+ public void preActivate()
+ {
+
+ }
+
+ public synchronized void activated()
+ {
+ if (!startCalled)
+ {
+ return;
+ }
+
+ try
+ {
+
+ jmsManagementService = new JMSManagementServiceImpl(server.getManagementService(), server, this);
+
+ jmsManagementService.registerJMSServer(this);
+
+ // Must be set to active before calling initJournal
+ active = true;
+
+ initJournal();
+
+ // start the JMS deployer only if the configuration is not done using the JMSConfiguration object
+ if (config == null)
+ {
+ if (server.getConfiguration().isFileDeploymentEnabled())
+ {
+ jmsDeployer = new JMSServerDeployer(this, deploymentManager);
+
+ if (configFileName != null)
+ {
+ jmsDeployer.setConfigFileNames(new String[]{configFileName});
+ }
+
+ jmsDeployer.start();
+
+ deploymentManager.start();
+ }
+ }
+ else
+ {
+ deploy();
+ }
+
+ for (Runnable run : cachedCommands)
+ {
+ HornetQJMSServerLogger.LOGGER.serverRunningCachedCommand(run);
+ run.run();
+ }
+
+ // do not clear the cachedCommands - HORNETQ-1047
+
+ recoverJndiBindings();
+ }
+ catch (Exception e)
+ {
+ active = false;
+ HornetQJMSServerLogger.LOGGER.jmsDeployerStartError(e);
+ }
+ }
+
+ @Override
+ public void deActivate()
+ {
+ try
+ {
+ synchronized (this)
+ {
+ if (!active)
+ {
+ return;
+ }
+
+ if (jmsDeployer != null)
+ {
+ jmsDeployer.stop();
+ }
+
+ if (deploymentManager != null)
+ {
+ deploymentManager.stop();
+ }
+
+ // Storage could be null on a shared store backup server before initialization
+ if (storage != null && storage.isStarted())
+ {
+ storage.stop();
+ }
+
+ unbindJNDI(queueJNDI);
+
+ unbindJNDI(topicJNDI);
+
+ unbindJNDI(connectionFactoryJNDI);
+
+ for (String connectionFactory : new HashSet<String>(connectionFactories.keySet()))
+ {
+ shutdownConnectionFactory(connectionFactory);
+ }
+
+ connectionFactories.clear();
+ connectionFactoryJNDI.clear();
+
+ queueJNDI.clear();
+ queues.clear();
+
+ topicJNDI.clear();
+ topics.clear();
+
+ // it could be null if a backup
+ if (jmsManagementService != null)
+ {
+ jmsManagementService.unregisterJMSServer();
+
+ jmsManagementService.stop();
+ }
+
+ jmsDeployer = null;
+ jmsManagementService = null;
+
+ active = false;
+ }
+ }
+ catch (Exception e)
+ {
+ e.printStackTrace();
+ }
+ }
+
+ @Override
+ public void activationComplete()
+ {
+
+ }
+
+ public void recoverJndiBindings(String name, PersistedType type) throws NamingException
+ {
+ List<String> bindings = unRecoveredJndi.get(name);
+ if ((bindings != null) && (bindings.size() > 0))
+ {
+ Map<String, List<String>> mapJNDI;
+ Map<String, ?> objects;
+
+ switch (type)
+ {
+ case Queue:
+ mapJNDI = queueJNDI;
+ objects = queues;
+ break;
+ case Topic:
+ mapJNDI = topicJNDI;
+ objects = topics;
+ break;
+ default:
+ case ConnectionFactory:
+ mapJNDI = connectionFactoryJNDI;
+ objects = connectionFactories;
+ break;
+ }
+
+ Object objectToBind = objects.get(name);
+
+ List<String> jndiList = mapJNDI.get(name);
+
+ if (objectToBind == null)
+ {
+ return;
+ }
+
+ if (jndiList == null)
+ {
+ jndiList = new ArrayList<String>();
+ mapJNDI.put(name, jndiList);
+ }
+
+ for (String jndi : bindings)
+ {
+ jndiList.add(jndi);
+ bindToJndi(jndi, objectToBind);
+ }
+
+ unRecoveredJndi.remove(name);
+ }
+ }
+
+ private void recoverJndiBindings() throws Exception
+ {
+ //now its time to add journal recovered stuff
+ List<PersistedJNDI> jndiSpace = storage.recoverPersistedJNDI();
+
+ for (PersistedJNDI record : jndiSpace)
+ {
+ Map<String, List<String>> mapJNDI;
+ Map<String, ?> objects;
+
+ switch (record.getType())
+ {
+ case Queue:
+ mapJNDI = queueJNDI;
+ objects = queues;
+ break;
+ case Topic:
+ mapJNDI = topicJNDI;
+ objects = topics;
+ break;
+ default:
+ case ConnectionFactory:
+ mapJNDI = connectionFactoryJNDI;
+ objects = connectionFactories;
+ break;
+ }
+
+ Object objectToBind = objects.get(record.getName());
+ List<String> jndiList = mapJNDI.get(record.getName());
+
+ if (objectToBind == null)
+ {
+ unRecoveredJndi.put(record.getName(), record.getJndi());
+ continue;
+ }
+
+ if (jndiList == null)
+ {
+ jndiList = new ArrayList<String>();
+ mapJNDI.put(record.getName(), jndiList);
+ }
+
+ for (String jndi : record.getJndi())
+ {
+ jndiList.add(jndi);
+ bindToJndi(jndi, objectToBind);
+ }
+ }
+
+
+ }
+
+ // HornetQComponent implementation -----------------------------------
+
+ /**
+ * Notice that this component has a {@link #startCalled} boolean to control its internal
+ * life-cycle, but its {@link #isStarted()} returns the value of {@code server.isStarted()} and
+ * not the value of {@link #startCalled}.
+ * <p>
+ * This method and {@code server.start()} are interdependent in the following way:
+ * <ol>
+ * <li>{@link JMSServerManagerImpl#start()} is called, it sets {@code start_called=true}, and
+ * calls {@link HornetQServerImpl#start()}
+ * <li>{@link HornetQServerImpl#start()} will call {@link JMSServerManagerImpl#activated()}
+ * <li>{@link JMSServerManagerImpl#activated()} checks the value of {@link #startCalled}, which
+ * must already be true.
+ * </ol>
+ */
+ public synchronized void start() throws Exception
+ {
+ if (startCalled)
+ {
+ return;
+ }
+
+ if (registry == null)
+ {
+ if (!contextSet)
+ {
+ registry = new JndiBindingRegistry(new InitialContext());
+ }
+ }
+
+ deploymentManager = new FileDeploymentManager(server.getConfiguration().getFileDeployerScanPeriod());
+ server.registerActivateCallback(this);
+ /**
+ * See this method's javadoc.
+ * <p>
+ * start_called MUST be set to true BEFORE calling server.start().
+ * <p>
+ * start_called is NOT used at {@link JMSServerManager#isStarted()}
+ */
+ startCalled = true;
+ server.start();
+
+
+ }
+
+ public void stop() throws Exception
+ {
+ synchronized (this)
+ {
+ if (!startCalled)
+ {
+ return;
+ }
+ startCalled = false;
+ //deactivate in case we haven't been already
+ deActivate();
+ if (registry != null)
+ {
+ registry.close();
+ }
+ }
+ // We have to perform the server.stop outside of the lock because of backup activation issues.
+ // See https://bugzilla.redhat.com/show_bug.cgi?id=959616
+ // And org.apache.activemq6.byteman.tests.StartStopDeadlockTest which is validating for this case here
+ server.stop();
+ }
+
+ public boolean isStarted()
+ {
+ return server.isStarted();
+ }
+
+ // JMSServerManager implementation -------------------------------
+
+ public BindingRegistry getRegistry()
+ {
+ return registry;
+ }
+
+ public void setRegistry(BindingRegistry registry)
+ {
+ this.registry = registry;
+ }
+
+ public HornetQServer getHornetQServer()
+ {
+ return server;
+ }
+
+ public void addAddressSettings(final String address, final AddressSettings addressSettings)
+ {
+ server.getAddressSettingsRepository().addMatch(address, addressSettings);
+ }
+
+ public AddressSettings getAddressSettings(final String address)
+ {
+ return server.getAddressSettingsRepository().getMatch(address);
+ }
+
+ public void addSecurity(final String addressMatch, final Set<Role> roles)
+ {
+ server.getSecurityRepository().addMatch(addressMatch, roles);
+ }
+
+ public Set<Role> getSecurity(final String addressMatch)
+ {
+ return server.getSecurityRepository().getMatch(addressMatch);
+ }
+
+ public synchronized void setContext(final Context context)
+ {
+ if (registry == null || registry instanceof JndiBindingRegistry)
+ {
+ registry = new JndiBindingRegistry(context);
+ registry.setContext(context);
+ }
+
+ contextSet = true;
+ }
+
+ public synchronized String getVersion()
+ {
+ checkInitialised();
+
+ return server.getVersion().getFullVersion();
+ }
+
+ public synchronized boolean createQueue(final boolean storeConfig,
+ final String queueName,
+ final String selectorString,
+ final boolean durable,
+ final String... jndi) throws Exception
+ {
+
+ if (active && queues.get(queueName) != null)
+ {
+ return false;
+ }
+
+ runAfterActive(new WrappedRunnable()
+ {
+ @Override
+ public String toString()
+ {
+ return "createQueue for " + queueName;
+ }
+
+ @Override
+ public void runException() throws Exception
+ {
+ checkJNDI(jndi);
+
+ if (internalCreateQueue(queueName, selectorString, durable))
+ {
+
+ HornetQDestination destination = queues.get(queueName);
+ if (destination == null)
+ {
+ // sanity check. internalCreateQueue should already have done this check
+ throw new IllegalArgumentException("Queue does not exist");
+ }
+
+ ArrayList<String> bindings = new ArrayList<String>();
+
+ for (String jndiItem : jndi)
+ {
+ if (bindToJndi(jndiItem, destination))
+ {
+ bindings.add(jndiItem);
+ }
+ }
+
+ String[] usedJNDI = bindings.toArray(new String[bindings.size()]);
+ addToBindings(queueJNDI, queueName, usedJNDI);
+
+ if (storeConfig && durable)
+ {
+ storage.storeDestination(new PersistedDestination(PersistedType.Queue,
+ queueName,
+ selectorString,
+ durable));
+ storage.addJNDI(PersistedType.Queue, queueName, usedJNDI);
+ }
+ }
+ }
+ });
+
+ sendNotification(JMSNotificationType.QUEUE_CREATED, queueName);
+ return true;
+ }
+
+ public synchronized boolean createTopic(final boolean storeConfig, final String topicName, final String... jndi) throws Exception
+ {
+ if (active && topics.get(topicName) != null)
+ {
+ return false;
+ }
+
+ runAfterActive(new WrappedRunnable()
+ {
+ @Override
+ public String toString()
+ {
+ return "createTopic for " + topicName;
+ }
+
+ @Override
+ public void runException() throws Exception
+ {
+ checkJNDI(jndi);
+
+ if (internalCreateTopic(topicName))
+ {
+ HornetQDestination destination = topics.get(topicName);
+
+ if (destination == null)
+ {
+ // sanity check. internalCreateQueue should already have done this check
+ throw new IllegalArgumentException("Queue does not exist");
+ }
+
+ ArrayList<String> bindings = new ArrayList<String>();
+
+ for (String jndiItem : jndi)
+ {
+ if (bindToJndi(jndiItem, destination))
+ {
+ bindings.add(jndiItem);
+ }
+ }
+
+ String[] usedJNDI = bindings.toArray(new String[bindings.size()]);
+ addToBindings(topicJNDI, topicName, usedJNDI);
+
+ if (storeConfig)
+ {
+ storage.storeDestination(new PersistedDestination(PersistedType.Topic, topicName));
+ storage.addJNDI(PersistedType.Topic, topicName, usedJNDI);
+ }
+ }
+ }
+ });
+
+ sendNotification(JMSNotificationType.TOPIC_CREATED, topicName);
+ return true;
+
+ }
+
+ public boolean addTopicToJndi(final String topicName, final String jndiBinding) throws Exception
+ {
+ checkInitialised();
+
+ checkJNDI(jndiBinding);
+
+ HornetQTopic destination = topics.get(topicName);
+ if (destination == null)
+ {
+ throw new IllegalArgumentException("Topic does not exist");
+ }
+ if (destination.getTopicName() == null)
+ {
+ throw new IllegalArgumentException(topicName + " is not a topic");
+ }
+ boolean added = bindToJndi(jndiBinding, destination);
+
+ if (added)
+ {
+ addToBindings(topicJNDI, topicName, jndiBinding);
+ storage.addJNDI(PersistedType.Topic, topicName, jndiBinding);
+ }
+ return added;
+ }
+
+ public String[] getJNDIOnQueue(String queue)
+ {
+ return getJNDIList(queueJNDI, queue);
+ }
+
+ public String[] getJNDIOnTopic(String topic)
+ {
+ return getJNDIList(topicJNDI, topic);
+ }
+
+ public String[] getJNDIOnConnectionFactory(String factoryName)
+ {
+ return getJNDIList(connectionFactoryJNDI, factoryName);
+ }
+
+ public boolean addQueueToJndi(final String queueName, final String jndiBinding) throws Exception
+ {
+ checkInitialised();
+
+ checkJNDI(jndiBinding);
+
+ HornetQQueue destination = queues.get(queueName);
+ if (destination == null)
+ {
+ throw new IllegalArgumentException("Queue does not exist");
+ }
+ if (destination.getQueueName() == null)
+ {
+ throw new IllegalArgumentException(queueName + " is not a queue");
+ }
+ boolean added = bindToJndi(jndiBinding, destination);
+ if (added)
+ {
+ addToBindings(queueJNDI, queueName, jndiBinding);
+ storage.addJNDI(PersistedType.Queue, queueName, jndiBinding);
+ }
+ return added;
+ }
+
+ public boolean addConnectionFactoryToJNDI(final String name, final String jndiBinding) throws Exception
+ {
+ checkInitialised();
+
+ checkJNDI(jndiBinding);
+
+ HornetQConnectionFactory factory = connectionFactories.get(name);
+ if (factory == null)
+ {
+ throw new IllegalArgumentException("Factory does not exist");
+ }
+ if (registry.lookup(jndiBinding) != null)
+ {
+ throw HornetQJMSServerBundle.BUNDLE.cfJndiExists(name);
+ }
+ boolean added = bindToJndi(jndiBinding, factory);
+ if (added)
+ {
+ addToBindings(connectionFactoryJNDI, name, jndiBinding);
+ storage.addJNDI(PersistedType.ConnectionFactory, name, jndiBinding);
+ }
+ return added;
+ }
+
+ @Override
+ public boolean removeQueueFromJNDI(String name, String jndi) throws Exception
+ {
+ checkInitialised();
+
+ boolean removed = removeFromJNDI(queueJNDI, name, jndi);
+
+ if (removed)
+ {
+ storage.deleteJNDI(PersistedType.Queue, name, jndi);
+ }
+
+ return removed;
+ }
+
+ @Override
+ public boolean removeQueueFromJNDI(final String name) throws Exception
+ {
+ final AtomicBoolean valueReturn = new AtomicBoolean(false);
+
+ // HORNETQ-911 - make this runAfterActive to prevent WARN messages on shutdown/undeployment when the backup was never activated
+ runAfterActive(new WrappedRunnable()
+ {
+ @Override
+ public String toString()
+ {
+ return "removeQueueFromJNDI for " + name;
+ }
+
+ @Override
+ public void runException() throws Exception
+ {
+ checkInitialised();
+
+ if (removeFromJNDI(queues, queueJNDI, name))
+ {
+ storage.deleteDestination(PersistedType.Queue, name);
+ valueReturn.set(true);
+ }
+ }
+ });
+
+ return valueReturn.get();
+ }
+
+ @Override
+ public boolean removeTopicFromJNDI(String name, String jndi) throws Exception
+ {
+ checkInitialised();
+
+ if (removeFromJNDI(topicJNDI, name, jndi))
+ {
+ storage.deleteJNDI(PersistedType.Topic, name, jndi);
+ return true;
+ }
+ else
+ {
+ return false;
+ }
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.activemq6.jms.server.JMSServerManager#removeTopicFromJNDI(java.lang.String, java.lang.String)
+ */
+ public boolean removeTopicFromJNDI(final String name) throws Exception
+ {
+ final AtomicBoolean valueReturn = new AtomicBoolean(false);
+
+ // HORNETQ-911 - make this runAfterActive to prevent WARN messages on shutdown/undeployment when the backup was never activated
+ runAfterActive(new WrappedRunnable()
+ {
+ @Override
+ public String toString()
+ {
+ return "removeTopicFromJNDI for " + name;
+ }
+
+ @Override
+ public void runException() throws Exception
+ {
+ checkInitialised();
+
+ if (removeFromJNDI(topics, topicJNDI, name))
+ {
+ storage.deleteDestination(PersistedType.Topic, name);
+ valueReturn.set(true);
+ }
+ }
+ });
+
+ return valueReturn.get();
+ }
+
+ @Override
+ public boolean removeConnectionFactoryFromJNDI(String name, String jndi) throws Exception
+ {
+ checkInitialised();
+
+ removeFromJNDI(connectionFactoryJNDI, name, jndi);
+
+ storage.deleteJNDI(PersistedType.ConnectionFactory, name, jndi);
+
+ return true;
+ }
+
+ @Override
+ public boolean removeConnectionFactoryFromJNDI(String name) throws Exception
+ {
+ checkInitialised();
+
+ removeFromJNDI(connectionFactories, connectionFactoryJNDI, name);
+
+ storage.deleteConnectionFactory(name);
+
+ return true;
+ }
+
+ public synchronized boolean destroyQueue(final String name) throws Exception
+ {
+ return destroyQueue(name, true);
+ }
+
+ public synchronized boolean destroyQueue(final String name, final boolean removeConsumers) throws Exception
+ {
+ checkInitialised();
+
+ server.destroyQueue(HornetQDestination.createQueueAddressFromName(name), null, !removeConsumers, removeConsumers);
+
+ // if the queue has consumers and 'removeConsumers' is false then the queue won't actually be removed
+ // therefore only remove the queue from JNDI, etc. if the queue is actually removed
+ if (this.server.getPostOffice().getBinding(HornetQDestination.createQueueAddressFromName(name)) == null)
+ {
+ removeFromJNDI(queues, queueJNDI, name);
+
+ queues.remove(name);
+ queueJNDI.remove(name);
+
+ jmsManagementService.unregisterQueue(name);
+
+ storage.deleteDestination(PersistedType.Queue, name);
+
+ sendNotification(JMSNotificationType.QUEUE_DESTROYED, name);
+ return true;
+ }
+ else
+ {
+ return false;
+ }
+ }
+
+ public synchronized boolean destroyTopic(final String name) throws Exception
+ {
+ return destroyTopic(name, true);
+ }
+
+ public synchronized boolean destroyTopic(final String name, final boolean removeConsumers) throws Exception
+ {
+ checkInitialised();
+ AddressControl addressControl = (AddressControl) server.getManagementService()
+ .getResource(ResourceNames.CORE_ADDRESS + HornetQDestination.createTopicAddressFromName(name));
+ if (addressControl != null)
+ {
+ for (String queueName : addressControl.getQueueNames())
+ {
+ Binding binding = server.getPostOffice().getBinding(new SimpleString(queueName));
+ if (binding == null)
+ {
+ HornetQJMSServerLogger.LOGGER.noQueueOnTopic(queueName, name);
+ continue;
+ }
+
+ // We can't remove the remote binding. As this would be the bridge associated with the topic on this case
+ if (binding.getType() != BindingType.REMOTE_QUEUE)
+ {
+ server.destroyQueue(SimpleString.toSimpleString(queueName), null, !removeConsumers, removeConsumers);
+ }
+ }
+
+ if (addressControl.getQueueNames().length == 0)
+ {
+ removeFromJNDI(topics, topicJNDI, name);
+
+ topics.remove(name);
+ topicJNDI.remove(name);
+
+ jmsManagementService.unregisterTopic(name);
+
+ storage.deleteDestination(PersistedType.Topic, name);
+
+ sendNotification(JMSNotificationType.TOPIC_DESTROYED, name);
+ return true;
+ }
+ else
+ {
+ return false;
+ }
+ }
+ else
+ {
+ return false;
+ }
+ }
+
+ public synchronized void createConnectionFactory(final String name,
+ final boolean ha,
+ final JMSFactoryType cfType,
+ final List<String> connectorNames,
+ String... jndiBindings) throws Exception
+ {
+ checkInitialised();
+ HornetQConnectionFactory cf = connectionFactories.get(name);
+ if (cf == null)
+ {
+ ConnectionFactoryConfiguration configuration = new ConnectionFactoryConfigurationImpl()
+ .setName(name)
+ .setHA(ha)
+ .setConnectorNames(connectorNames)
+ .setFactoryType(cfType);
+
+ createConnectionFactory(true, configuration, jndiBindings);
+ }
+ }
+
+ public synchronized void createConnectionFactory(final String name,
+ final boolean ha,
+ JMSFactoryType cfType,
+ final List<String> connectorNames,
+ final String clientID,
+ final long clientFailureCheckPeriod,
+ final long connectionTTL,
+ final long callTimeout,
+ final long callFailoverTimeout,
+ final boolean cacheLargeMessagesClient,
+ final int minLargeMessageSize,
+ final boolean compressLargeMessage,
+ final int consumerWindowSize,
+ final int consumerMaxRate,
+ final int confirmationWindowSize,
+ final int producerWindowSize,
+ final int producerMaxRate,
+ final boolean blockOnAcknowledge,
+ final boolean blockOnDurableSend,
+ final boolean blockOnNonDurableSend,
+ final boolean autoGroup,
+ final boolean preAcknowledge,
+ final String loadBalancingPolicyClassName,
+ final int transactionBatchSize,
+ final int dupsOKBatchSize,
+ final boolean useGlobalPools,
+ final int scheduledThreadPoolMaxSize,
+ final int threadPoolMaxSize,
+ final long retryInterval,
+ final double retryIntervalMultiplier,
+ final long maxRetryInterval,
+ final int reconnectAttempts,
+ final boolean failoverOnInitialConnection,
+ final String groupId,
+ String... jndiBindings) throws Exception
+ {
+ checkInitialised();
+ HornetQConnectionFactory cf = connectionFactories.get(name);
+ if (cf == null)
+ {
+ ConnectionFactoryConfiguration configuration = new ConnectionFactoryConfigurationImpl()
+ .setName(name)
+ .setHA(ha)
+ .setConnectorNames(connectorNames)
+ .setClientID(clientID)
+ .setClientFailureCheckPeriod(clientFailureCheckPeriod)
+ .setConnectionTTL(connectionTTL)
+ .setFactoryType(cfType)
+ .setCallTimeout(callTimeout)
+ .setCallFailoverTimeout(callFailoverTimeout)
+ .setCacheLargeMessagesClient(cacheLargeMessagesClient)
+ .setMinLargeMessageSize(minLargeMessageSize)
+ .setConsumerWindowSize(consumerWindowSize)
+ .setConsumerMaxRate(consumerMaxRate)
+ .setConfirmationWindowSize(confirmationWindowSize)
+ .setProducerWindowSize(producerWindowSize)
+ .setProducerMaxRate(producerMaxRate)
+ .setBlockOnAcknowledge(blockOnAcknowledge)
+ .setBlockOnDurableSend(blockOnDurableSend)
+ .setBlockOnNonDurableSend(blockOnNonDurableSend)
+ .setAutoGroup(autoGroup)
+ .setPreAcknowledge(preAcknowledge)
+ .setLoadBalancingPolicyClassName(loadBalancingPolicyClassName)
+ .setTransactionBatchSize(transactionBatchSize)
+ .setDupsOKBatchSize(dupsOKBatchSize)
+ .setUseGlobalPools(useGlobalPools)
+ .setScheduledThreadPoolMaxSize(scheduledThreadPoolMaxSize)
+ .setThreadPoolMaxSize(threadPoolMaxSize)
+ .setRetryInterval(retryInterval)
+ .setRetryIntervalMultiplier(retryIntervalMultiplier)
+ .setMaxRetryInterval(maxRetryInterval)
+ .setReconnectAttempts(reconnectAttempts)
+ .setFailoverOnInitialConnection(failoverOnInitialConnection)
+ .setGroupID(groupId);
+
+ createConnectionFactory(true, configuration, jndiBindings);
+ }
+ }
+
+ public synchronized void createConnectionFactory(final String name,
+ final boolean ha,
+ final JMSFactoryType cfType,
+ final String discoveryGroupName,
+ final String clientID,
+ final long clientFailureCheckPeriod,
+ final long connectionTTL,
+ final long callTimeout,
+ final long callFailoverTimeout,
+ final boolean cacheLargeMessagesClient,
+ final int minLargeMessageSize,
+ final boolean compressLargeMessages,
+ final int consumerWindowSize,
+ final int consumerMaxRate,
+ final int confirmationWindowSize,
+ final int producerWindowSize,
+ final int producerMaxRate,
+ final boolean blockOnAcknowledge,
+ final boolean blockOnDurableSend,
+ final boolean blockOnNonDurableSend,
+ final boolean autoGroup,
+ final boolean preAcknowledge,
+ final String loadBalancingPolicyClassName,
+ final int transactionBatchSize,
+ final int dupsOKBatchSize,
+ final boolean useGlobalPools,
+ final int scheduledThreadPoolMaxSize,
+ final int threadPoolMaxSize,
+ final long retryInterval,
+ final double retryIntervalMultiplier,
+ final long maxRetryInterval,
+ final int reconnectAttempts,
+ final boolean failoverOnInitialConnection,
+ final String groupId,
+ final String... jndiBindings) throws Exception
+ {
+ checkInitialised();
+ HornetQConnectionFactory cf = connectionFactories.get(name);
+ if (cf == null)
+ {
+ ConnectionFactoryConfiguration configuration = new ConnectionFactoryConfigurationImpl()
+ .setName(name)
+ .setHA(ha)
+ .setBindings(jndiBindings)
+ .setDiscoveryGroupName(discoveryGroupName)
+ .setFactoryType(cfType)
+ .setClientID(clientID)
+ .setClientFailureCheckPeriod(clientFailureCheckPeriod)
+ .setConnectionTTL(connectionTTL)
+ .setCallTimeout(callTimeout)
+ .setCallFailoverTimeout(callFailoverTimeout)
+ .setCacheLargeMessagesClient(cacheLargeMessagesClient)
+ .setMinLargeMessageSize(minLargeMessageSize)
+ .setCompressLargeMessages(compressLargeMessages)
+ .setConsumerWindowSize(consumerWindowSize)
+ .setConsumerMaxRate(consumerMaxRate)
+ .setConfirmationWindowSize(confirmationWindowSize)
+ .setProducerWindowSize(producerWindowSize)
+ .setProducerMaxRate(producerMaxRate)
+ .setBlockOnAcknowledge(blockOnAcknowledge)
+ .setBlockOnDurableSend(blockOnDurableSend)
+ .setBlockOnNonDurableSend(blockOnNonDurableSend)
+ .setAutoGroup(autoGroup)
+ .setPreAcknowledge(preAcknowledge)
+ .setLoadBalancingPolicyClassName(loadBalancingPolicyClassName)
+ .setTransactionBatchSize(transactionBatchSize)
+ .setDupsOKBatchSize(dupsOKBatchSize)
+ .setUseGlobalPools(useGlobalPools)
+ .setScheduledThreadPoolMaxSize(scheduledThreadPoolMaxSize)
+ .setThreadPoolMaxSize(threadPoolMaxSize)
+ .setRetryInterval(retryInterval)
+ .setRetryIntervalMultiplier(retryIntervalMultiplier)
+ .setMaxRetryInterval(maxRetryInterval)
+ .setReconnectAttempts(reconnectAttempts)
+ .setFailoverOnInitialConnection(failoverOnInitialConnection);
+ createConnectionFactory(true, configuration, jndiBindings);
+ }
+ }
+
+ public synchronized void createConnectionFactory(final String name,
+ final boolean ha,
+ final JMSFactoryType cfType,
+ final String discoveryGroupName,
+ final String... jndiBindings) throws Exception
+ {
+ checkInitialised();
+ HornetQConnectionFactory cf = connectionFactories.get(name);
+ if (cf == null)
+ {
+ ConnectionFactoryConfiguration configuration = new ConnectionFactoryConfigurationImpl()
+ .setName(name)
+ .setHA(ha)
+ .setBindings(jndiBindings)
+ .setDiscoveryGroupName(discoveryGroupName);
+ createConnectionFactory(true, configuration, jndiBindings);
+ }
+ }
+
+ public synchronized HornetQConnectionFactory recreateCF(String name, ConnectionFactoryConfiguration cf) throws Exception
+ {
+ List<String> jndi = connectionFactoryJNDI.get(name);
+
+ if (jndi == null)
+ {
+ throw HornetQJMSServerBundle.BUNDLE.cfDoesntExist(name);
+ }
+
+ String[] usedJNDI = jndi.toArray(new String[jndi.size()]);
+
+ HornetQConnectionFactory realCF = internalCreateCFPOJO(cf);
+
+ if (cf.isPersisted())
+ {
+ storage.storeConnectionFactory(new PersistedConnectionFactory(cf));
+ storage.addJNDI(PersistedType.ConnectionFactory, cf.getName(), usedJNDI);
+ }
+
+ for (String jndiElement : usedJNDI)
+ {
+ this.bindToJndi(jndiElement, realCF);
+ }
+
+ return realCF;
+ }
+
+ public synchronized void createConnectionFactory(final boolean storeConfig,
+ final ConnectionFactoryConfiguration cfConfig,
+ final String... jndi) throws Exception
+ {
+ runAfterActive(new WrappedRunnable()
+ {
+
+ @Override
+ public String toString()
+ {
+ return "createConnectionFactory for " + cfConfig.getName();
+ }
+
+ @Override
+ public void runException() throws Exception
+ {
+ checkJNDI(jndi);
+
+ HornetQConnectionFactory cf = internalCreateCF(storeConfig, cfConfig);
+
+ ArrayList<String> bindings = new ArrayList<String>();
+
+ for (String jndiItem : jndi)
+ {
+ if (bindToJndi(jndiItem, cf))
+ {
+ bindings.add(jndiItem);
+ }
+ }
+
+ String[] usedJNDI = bindings.toArray(new String[bindings.size()]);
+ addToBindings(connectionFactoryJNDI, cfConfig.getName(), usedJNDI);
+
+ if (storeConfig)
+ {
+ storage.storeConnectionFactory(new PersistedConnectionFactory(cfConfig));
+ storage.addJNDI(PersistedType.ConnectionFactory, cfConfig.getName(), usedJNDI);
+ }
+
+ JMSServerManagerImpl.this.recoverJndiBindings(cfConfig.getName(), PersistedType.ConnectionFactory);
+ sendNotification(JMSNotificationType.CONNECTION_FACTORY_CREATED, cfConfig.getName());
+ }
+ });
+ }
+
+ private void sendNotification(JMSNotificationType type, String message)
+ {
+ TypedProperties prop = new TypedProperties();
+ prop.putSimpleStringProperty(JMSNotificationType.MESSAGE, SimpleString.toSimpleString(message));
+ Notification notif = new Notification(null, type, prop);
+ try
+ {
+ server.getManagementService().sendNotification(notif);
+ }
+ catch (Exception e)
+ {
+ HornetQJMSServerLogger.LOGGER.warn("Failed to send notification : " + notif);
+ }
+ }
+
+ public JMSStorageManager getJMSStorageManager()
+ {
+ return storage;
+ }
+
+ // used on tests only
+ public void replaceStorageManager(JMSStorageManager newStorage)
+ {
+ this.storage = newStorage;
+ }
+
+ private String[] getJNDIList(final Map<String, List<String>> map, final String name)
+ {
+ List<String> result = map.get(name);
+ if (result == null)
+ {
+ return new String[0];
+ }
+ else
+ {
+ String[] strings = new String[result.size()];
+ result.toArray(strings);
+ return strings;
+ }
+ }
+
+ private boolean internalCreateQueue(final String queueName, final String selectorString, final boolean durable) throws Exception
+ {
+ if (queues.get(queueName) != null)
+ {
+ return false;
+ }
+ else
+ {
+ HornetQQueue hqQueue = HornetQDestination.createQueue(queueName);
+
+ // Convert from JMS selector to core filter
+ String coreFilterString = null;
+
+ if (selectorString != null)
+ {
+ coreFilterString = SelectorTranslator.convertToHornetQFilterString(selectorString);
+ }
+
+ Queue queue = server.deployQueue(SimpleString.toSimpleString(hqQueue.getAddress()),
+ SimpleString.toSimpleString(hqQueue.getAddress()),
+ SimpleString.toSimpleString(coreFilterString),
+ durable,
+ false);
+
+ queues.put(queueName, hqQueue);
+
+ this.recoverJndiBindings(queueName, PersistedType.Queue);
+
+ jmsManagementService.registerQueue(hqQueue, queue);
+
+ return true;
+ }
+ }
+
+ /**
+ * Performs the internal creation without activating any storage.
+ * The storage load will call this method
+ *
+ * @param topicName
+ * @return
+ * @throws Exception
+ */
+ private boolean internalCreateTopic(final String topicName) throws Exception
+ {
+
+ if (topics.get(topicName) != null)
+ {
+ return false;
+ }
+ else
+ {
+ HornetQTopic hqTopic = HornetQDestination.createTopic(topicName);
+ // We create a dummy subscription on the topic, that never receives messages - this is so we can perform JMS
+ // checks when routing messages to a topic that
+ // does not exist - otherwise we would not be able to distinguish from a non existent topic and one with no
+ // subscriptions - core has no notion of a topic
+ server.deployQueue(SimpleString.toSimpleString(hqTopic.getAddress()),
+ SimpleString.toSimpleString(hqTopic.getAddress()),
+ SimpleString.toSimpleString(JMSServerManagerImpl.REJECT_FILTER),
+ true,
+ false);
+
+ topics.put(topicName, hqTopic);
+
+ this.recoverJndiBindings(topicName, PersistedType.Topic);
+
+ jmsManagementService.registerTopic(hqTopic);
+
+ return true;
+ }
+ }
+
+ /**
+ * @param cfConfig
+ * @throws Exception
+ */
+ private HornetQConnectionFactory internalCreateCF(final boolean persisted,
+ final ConnectionFactoryConfiguration cfConfig) throws Exception
+ {
+ checkInitialised();
+
+ HornetQConnectionFactory cf = connectionFactories.get(cfConfig.getName());
+
+ if (cf == null)
+ {
+ cf = internalCreateCFPOJO(cfConfig);
+ }
+
+ connectionFactories.put(cfConfig.getName(), cf);
+
+ jmsManagementService.registerConnectionFactory(cfConfig.getName(), cfConfig, cf);
+
+ return cf;
+ }
+
+ /**
+ * @param cfConfig
+ * @return
+ * @throws HornetQException
+ */
+ protected HornetQConnectionFactory internalCreateCFPOJO(final ConnectionFactoryConfiguration cfConfig) throws HornetQException
+ {
+ HornetQConnectionFactory cf;
+ if (cfConfig.getDiscoveryGroupName() != null)
+ {
+ DiscoveryGroupConfiguration groupConfig = server.getConfiguration()
+ .getDiscoveryGroupConfigurations()
+ .get(cfConfig.getDiscoveryGroupName());
+
+ if (groupConfig == null)
+ {
+ throw HornetQJMSServerBundle.BUNDLE.discoveryGroupDoesntExist(cfConfig.getDiscoveryGroupName());
+ }
+
+ if (cfConfig.isHA())
+ {
+ cf = HornetQJMSClient.createConnectionFactoryWithHA(groupConfig, cfConfig.getFactoryType());
+ }
+ else
+ {
+ cf = HornetQJMSClient.createConnectionFactoryWithoutHA(groupConfig, cfConfig.getFactoryType());
+ }
+ }
+ else
+ {
+ if (cfConfig.getConnectorNames() == null || cfConfig.getConnectorNames().size() == 0)
+ {
+ throw HornetQJMSServerBundle.BUNDLE.noConnectorNameOnCF();
+ }
+
+ TransportConfiguration[] configs = new TransportConfiguration[cfConfig.getConnectorNames().size()];
+
+ int count = 0;
+ for (String name : cfConfig.getConnectorNames())
+ {
+ TransportConfiguration connector = server.getConfiguration().getConnectorConfigurations().get(name);
+ if (connector == null)
+ {
+ throw HornetQJMSServerBundle.BUNDLE.noConnectorNameConfiguredOnCF(name);
+ }
+ correctInvalidNettyConnectorHost(connector);
+ configs[count++] = connector;
+ }
+
+ if (cfConfig.isHA())
+ {
+ cf = HornetQJMSClient.createConnectionFactoryWithHA(cfConfig.getFactoryType(), configs);
+ }
+ else
+ {
+ cf = HornetQJMSClient.createConnectionFactoryWithoutHA(cfConfig.getFactoryType(), configs);
+ }
+ }
+
+ cf.setClientID(cfConfig.getClientID());
+ cf.setClientFailureCheckPeriod(cfConfig.getClientFailureCheckPeriod());
+ cf.setConnectionTTL(cfConfig.getConnectionTTL());
+ cf.setCallTimeout(cfConfig.getCallTimeout());
+ cf.setCallFailoverTimeout(cfConfig.getCallFailoverTimeout());
+ cf.setCacheLargeMessagesClient(cfConfig.isCacheLargeMessagesClient());
+ cf.setMinLargeMessageSize(cfConfig.getMinLargeMessageSize());
+ cf.setConsumerWindowSize(cfConfig.getConsumerWindowSize());
+ cf.setConsumerMaxRate(cfConfig.getConsumerMaxRate());
+ cf.setConfirmationWindowSize(cfConfig.getConfirmationWindowSize());
+ cf.setProducerWindowSize(cfConfig.getProducerWindowSize());
+ cf.setProducerMaxRate(cfConfig.getProducerMaxRate());
+ cf.setBlockOnAcknowledge(cfConfig.isBlockOnAcknowledge());
+ cf.setBlockOnDurableSend(cfConfig.isBlockOnDurableSend());
+ cf.setBlockOnNonDurableSend(cfConfig.isBlockOnNonDurableSend());
+ cf.setAutoGroup(cfConfig.isAutoGroup());
+ cf.setPreAcknowledge(cfConfig.isPreAcknowledge());
+ cf.setConnectionLoadBalancingPolicyClassName(cfConfig.getLoadBalancingPolicyClassName());
+ cf.setTransactionBatchSize(cfConfig.getTransactionBatchSize());
+ cf.setDupsOKBatchSize(cfConfig.getDupsOKBatchSize());
+ cf.setUseGlobalPools(cfConfig.isUseGlobalPools());
+ cf.setScheduledThreadPoolMaxSize(cfConfig.getScheduledThreadPoolMaxSize());
+ cf.setThreadPoolMaxSize(cfConfig.getThreadPoolMaxSize());
+ cf.setRetryInterval(cfConfig.getRetryInterval());
+ cf.setRetryIntervalMultiplier(cfConfig.getRetryIntervalMultiplier());
+ cf.setMaxRetryInterval(cfConfig.getMaxRetryInterval());
+ cf.setReconnectAttempts(cfConfig.getReconnectAttempts());
+ cf.setFailoverOnInitialConnection(cfConfig.isFailoverOnInitialConnection());
+ cf.setCompressLargeMessage(cfConfig.isCompressLargeMessages());
+ cf.setGroupID(cfConfig.getGroupID());
+ return cf;
+ }
+
+ public synchronized boolean destroyConnectionFactory(final String name) throws Exception
+ {
+ final AtomicBoolean valueReturn = new AtomicBoolean(false);
+
+ // HORNETQ-911 - make this runAfterActive to prevent WARN messages on shutdown/undeployment when the backup was never activated
+ runAfterActive(new WrappedRunnable()
+ {
+
+ @Override
+ public String toString()
+ {
+ return "destroyConnectionFactory for " + name;
+ }
+
+ @Override
+ public void runException() throws Exception
+ {
+ shutdownConnectionFactory(name);
+
+ storage.deleteConnectionFactory(name);
+ valueReturn.set(true);
+ }
+ });
+
+ if (valueReturn.get())
+ {
+ sendNotification(JMSNotificationType.CONNECTION_FACTORY_DESTROYED, name);
+ }
+
+ return valueReturn.get();
+ }
+
+ /**
+ * @param name
+ * @throws Exception
+ */
+ protected boolean shutdownConnectionFactory(final String name) throws Exception
+ {
+ checkInitialised();
+ List<String> jndiBindings = connectionFactoryJNDI.get(name);
+
+ if (registry != null)
+ {
+ for (String jndiBinding : jndiBindings)
+ {
+ registry.unbind(jndiBinding);
+ }
+ }
+
+ connectionFactoryJNDI.remove(name);
+ connectionFactories.remove(name);
+
+ jmsManagementService.unregisterConnectionFactory(name);
+
+ return true;
+ }
+
+ public String[] listRemoteAddresses() throws Exception
+ {
+ checkInitialised();
+ return server.getHornetQServerControl().listRemoteAddresses();
+ }
+
+ public String[] listRemoteAddresses(final String ipAddress) throws Exception
+ {
+ checkInitialised();
+ return server.getHornetQServerControl().listRemoteAddresses(ipAddress);
+ }
+
+ public boolean closeConnectionsForAddress(final String ipAddress) throws Exception
+ {
+ checkInitialised();
+ return server.getHornetQServerControl().closeConnectionsForAddress(ipAddress);
+ }
+
+ public boolean closeConsumerConnectionsForAddress(final String address) throws Exception
+ {
+ checkInitialised();
+ return server.getHornetQServerControl().closeConsumerConnectionsForAddress(address);
+ }
+
+ public boolean closeConnectionsForUser(final String userName) throws Exception
+ {
+ checkInitialised();
+ return server.getHornetQServerControl().closeConnectionsForUser(userName);
+ }
+
+ public String[] listConnectionIDs() throws Exception
+ {
+ return server.getHornetQServerControl().listConnectionIDs();
+ }
+
+ public String[] listSessions(final String connectionID) throws Exception
+ {
+ checkInitialised();
+ return server.getHornetQServerControl().listSessions(connectionID);
+ }
+
+ public String listPreparedTransactionDetailsAsJSON() throws Exception
+ {
+ ResourceManager resourceManager = server.getResourceManager();
+ Map<Xid, Long> xids = resourceManager.getPreparedTransactionsWithCreationTime();
+ if (xids == null || xids.size() == 0)
+ {
+ return "";
+ }
+
+ ArrayList<Entry<Xid, Long>> xidsSortedByCreationTime = new ArrayList<Map.Entry<Xid, Long>>(xids.entrySet());
+ Collections.sort(xidsSortedByCreationTime, new Comparator<Entry<Xid, Long>>()
+ {
+ public int compare(final Entry<Xid, Long> entry1, final Entry<Xid, Long> entry2)
+ {
+ // sort by creation time, oldest first
+ return (int) (entry1.getValue() - entry2.getValue());
+ }
+ });
+
+ JSONArray txDetailListJson = new JSONArray();
+ for (Map.Entry<Xid, Long> entry : xidsSortedByCreationTime)
+ {
+ Xid xid = entry.getKey();
+ Transaction tx = resourceManager.getTransaction(xid);
+ if (tx == null)
+ {
+ continue;
+ }
+ TransactionDetail detail = new JMSTransactionDetail(xid, tx, entry.getValue());
+ txDetailListJson.put(detail.toJSON());
+ }
+ return txDetailListJson.toString();
+ }
+
+ public String listPreparedTransactionDetailsAsHTML() throws Exception
+ {
+ ResourceManager resourceManager = server.getResourceManager();
+ Map<Xid, Long> xids = resourceManager.getPreparedTransactionsWithCreationTime();
+ if (xids == null || xids.size() == 0)
+ {
+ return "<h3>*** Prepared Transaction Details ***</h3><p>No entry.</p>";
+ }
+
+ ArrayList<Entry<Xid, Long>> xidsSortedByCreationTime = new ArrayList<Map.Entry<Xid, Long>>(xids.entrySet());
+ Collections.sort(xidsSortedByCreationTime, new Comparator<Entry<Xid, Long>>()
+ {
+ public int compare(final Entry<Xid, Long> entry1, final Entry<Xid, Long> entry2)
+ {
+ // sort by creation time, oldest first
+ return (int) (entry1.getValue() - entry2.getValue());
+ }
+ });
+
+ StringBuilder html = new StringBuilder();
+ html.append("<h3>*** Prepared Transaction Details ***</h3>");
+
+ for (Map.Entry<Xid, Long> entry : xidsSortedByCreationTime)
+ {
+ Xid xid = entry.getKey();
+ Transaction tx = resourceManager.getTransaction(xid);
+ if (tx == null)
+ {
+ continue;
+ }
+ TransactionDetail detail = new JMSTransactionDetail(xid, tx, entry.getValue());
+ JSONObject txJson = detail.toJSON();
+
+ html.append("<table border=\"1\">");
+ html.append("<tr><th>creation_time</th>");
+ html.append("<td>" + txJson.get(TransactionDetail.KEY_CREATION_TIME) + "</td>");
+ html.append("<th>xid_as_base_64</th>");
+ html.append("<td colspan=\"3\">" + txJson.get(TransactionDetail.KEY_XID_AS_BASE64) + "</td></tr>");
+ html.append("<tr><th>xid_format_id</th>");
+ html.append("<td>" + txJson.get(TransactionDetail.KEY_XID_FORMAT_ID) + "</td>");
+ html.append("<th>xid_global_txid</th>");
+ html.append("<td>" + txJson.get(TransactionDetail.KEY_XID_GLOBAL_TXID) + "</td>");
+ html.append("<th>xid_branch_qual</th>");
+ html.append("<td>" + txJson.get(TransactionDetail.KEY_XID_BRANCH_QUAL) + "</td></tr>");
+
+ html.append("<tr><th colspan=\"6\">Message List</th></tr>");
+ html.append("<tr><td colspan=\"6\">");
+ html.append("<table border=\"1\" cellspacing=\"0\" cellpadding=\"0\">");
+
+ JSONArray msgs = txJson.getJSONArray(TransactionDetail.KEY_TX_RELATED_MESSAGES);
+ for (int i = 0; i < msgs.length(); i++)
+ {
+ JSONObject msgJson = msgs.getJSONObject(i);
+ JSONObject props = msgJson.getJSONObject(TransactionDetail.KEY_MSG_PROPERTIES);
+ StringBuilder propstr = new StringBuilder();
+ @SuppressWarnings("unchecked")
+ Iterator<String> propkeys = props.keys();
+ while (propkeys.hasNext())
+ {
+ String key = propkeys.next();
+ propstr.append(key);
+ propstr.append("=");
+ propstr.append(props.get(key));
+ propstr.append(", ");
+ }
+
+ html.append("<th>operation_type</th>");
+ html.append("<td>" + msgJson.get(TransactionDetail.KEY_MSG_OP_TYPE) + "</th>");
+ html.append("<th>message_type</th>");
+ html.append("<td>" + msgJson.get(TransactionDetail.KEY_MSG_TYPE) + "</td></tr>");
+ html.append("<tr><th>properties</th>");
+ html.append("<td colspan=\"3\">" + propstr.toString() + "</td></tr>");
+ }
+ html.append("</table></td></tr>");
+ html.append("</table><br/>");
+ }
+
+ return html.toString();
+ }
+
+ // Public --------------------------------------------------------
+
+ // Private -------------------------------------------------------
+
+ private synchronized void checkInitialised()
+ {
+ if (!active)
+ {
+ throw new IllegalStateException("Cannot access JMS Server, core server is not yet active");
+ }
+ }
+
+ private void addToBindings(Map<String, List<String>> map, String name, String... jndi)
+ {
+ List<String> list = map.get(name);
+ if (list == null)
+ {
+ list = new ArrayList<String>();
+ map.put(name, list);
+ }
+ for (String jndiItem : jndi)
+ {
+ list.add(jndiItem);
+ }
+ }
+
+ private void checkJNDI(final String... jndiNames) throws NamingException
+ {
+
+ for (String jndiName : jndiNames)
+ {
+ if (registry.lookup(jndiName) != null)
+ {
+ throw new NamingException(jndiName + " already has an object bound");
+ }
+ }
+ }
+
+ private boolean bindToJndi(final String jndiName, final Object objectToBind) throws NamingException
+ {
+ if (registry != null)
+ {
+ registry.unbind(jndiName);
+ registry.bind(jndiName, objectToBind);
+ }
+ return true;
+ }
+
+ private void deploy() throws Exception
+ {
+ if (config == null)
+ {
+ return;
+ }
+
+ if (config.getContext() != null)
+ {
+ setContext(config.getContext());
+ }
+
+ List<ConnectionFactoryConfiguration> connectionFactoryConfigurations = config.getConnectionFactoryConfigurations();
+ for (ConnectionFactoryConfiguration cfConfig : connectionFactoryConfigurations)
+ {
+ createConnectionFactory(false, cfConfig, cfConfig.getBindings());
+ }
+
+ List<JMSQueueConfiguration> queueConfigs = config.getQueueConfigurations();
+ for (JMSQueueConfiguration qConfig : queueConfigs)
+ {
+ String[] bindings = qConfig.getBindings();
+ createQueue(false, qConfig.getName(), qConfig.getSelector(), qConfig.isDurable(), bindings);
+ }
+
+ List<TopicConfiguration> topicConfigs = config.getTopicConfigurations();
+ for (TopicConfiguration tConfig : topicConfigs)
+ {
+ String[] bindings = tConfig.getBindings();
+ createTopic(false, tConfig.getName(), bindings);
+ }
+ }
+
+ /**
+ * @param param
+ */
+ private void unbindJNDI(Map<String, List<String>> param)
+ {
+ if (registry != null)
+ {
+ for (List<String> elementList : param.values())
+ {
+ for (String key : elementList)
+ {
+ try
+ {
+ registry.unbind(key);
+ }
+ catch (Exception e)
+ {
+ HornetQJMSServerLogger.LOGGER.jndiUnbindError(e, key);
+ }
+ }
+ }
+ }
+ }
+
+ /**
+ * @throws Exception
+ */
+ private void initJournal() throws Exception
+ {
+ this.coreConfig = server.getConfiguration();
+
+ createJournal();
+
+ storage.load();
+
+ List<PersistedConnectionFactory> cfs = storage.recoverConnectionFactories();
+
+ for (PersistedConnectionFactory cf : cfs)
+ {
+ internalCreateCF(true, cf.getConfig());
+ }
+
+ List<PersistedDestination> destinations = storage.recoverDestinations();
+
+ for (PersistedDestination destination : destinations)
+ {
+ if (destination.getType() == PersistedType.Queue)
+ {
+ internalCreateQueue(destination.getName(), destination.getSelector(), destination.isDurable());
+ }
+ else if (destination.getType() == PersistedType.Topic)
+ {
+ internalCreateTopic(destination.getName());
+ }
+ }
+ }
+
+ /**
+ * @throws Exception
+ */
+ private void createJournal() throws Exception
+ {
+ if (storage == null)
+ {
+ if (coreConfig.isPersistenceEnabled())
+ {
+ storage = new JMSJournalStorageManagerImpl(new TimeAndCounterIDGenerator(),
+ server.getConfiguration(),
+ server.getReplicationManager());
+ }
+ else
+ {
+ storage = new NullJMSStorageManagerImpl();
+ }
+ }
+ else
+ {
+ if (storage.isStarted())
+ {
+ storage.stop();
+ }
+ }
+
+ storage.start();
+ }
+
+ private synchronized boolean removeFromJNDI(final Map<String, ?> keys,
+ final Map<String, List<String>> jndiMap,
+ final String name) throws Exception
+ {
+ checkInitialised();
+ List<String> jndiBindings = jndiMap.remove(name);
+ if (jndiBindings == null || jndiBindings.size() == 0)
+ {
+ return false;
+ }
+ else
+ {
+ keys.remove(name);
+ }
+ if (registry != null)
+ {
+ Iterator<String> iter = jndiBindings.iterator();
+ while (iter.hasNext())
+ {
+ String jndiBinding = iter.next();
+ registry.unbind(jndiBinding);
+ iter.remove();
+ }
+ }
+ return true;
+ }
+
+ private synchronized boolean removeFromJNDI(final Map<String, List<String>> jndiMap,
+ final String name,
+ final String jndi) throws Exception
+ {
+ checkInitialised();
+ List<String> jndiBindings = jndiMap.get(name);
+ if (jndiBindings == null || jndiBindings.size() == 0)
+ {
+ return false;
+ }
+
+ if (jndiBindings.remove(jndi))
+ {
+ registry.unbind(jndi);
+ return true;
+ }
+ else
+ {
+ return false;
+ }
+ }
+
+ private boolean runAfterActive(WrappedRunnable runnable) throws Exception
+ {
+ if (active)
+ {
+ runnable.runException();
+ return true;
+ }
+ else
+ {
+ HornetQJMSServerLogger.LOGGER.serverCachingCommand(runnable);
+ cachedCommands.add(runnable);
+ return false;
+ }
+ }
+
+ private abstract class WrappedRunnable implements Runnable
+ {
+ public void run()
+ {
+ try
+ {
+ runException();
+ }
+ catch (Exception e)
+ {
+ HornetQJMSServerLogger.LOGGER.jmsServerError(e);
+ }
+ }
+
+ public abstract void runException() throws Exception;
+ }
+
+ private void correctInvalidNettyConnectorHost(TransportConfiguration transportConfiguration)
+ {
+ Map<String, Object> params = transportConfiguration.getParams();
+
+ if (transportConfiguration.getFactoryClassName().equals(NettyConnectorFactory.class.getCanonicalName()) &&
+ params.containsKey(TransportConstants.HOST_PROP_NAME) &&
+ params.get(TransportConstants.HOST_PROP_NAME).equals("0.0.0.0"))
+ {
+ try
+ {
+ String newHost = InetAddress.getLocalHost().getHostName();
+ HornetQJMSServerLogger.LOGGER.invalidHostForConnector(transportConfiguration.getName(), newHost);
+ params.put(TransportConstants.HOST_PROP_NAME, newHost);
+ }
+ catch (UnknownHostException e)
+ {
+ HornetQJMSServerLogger.LOGGER.failedToCorrectHost(e, transportConfiguration.getName());
+ }
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/activemq-6/blob/23e8edd9/activemq6-jms-server/src/main/java/org/apache/activemq6/jms/server/impl/StandaloneNamingServer.java
----------------------------------------------------------------------
diff --git a/activemq6-jms-server/src/main/java/org/apache/activemq6/jms/server/impl/StandaloneNamingServer.java b/activemq6-jms-server/src/main/java/org/apache/activemq6/jms/server/impl/StandaloneNamingServer.java
new file mode 100644
index 0000000..7255e1e
--- /dev/null
+++ b/activemq6-jms-server/src/main/java/org/apache/activemq6/jms/server/impl/StandaloneNamingServer.java
@@ -0,0 +1,167 @@
+/*
+ * Copyright 2005-2014 Red Hat, Inc.
+ * Red Hat 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 org.apache.activemq6.jms.server.impl;
+
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+import org.apache.activemq6.core.server.ActivateCallback;
+import org.apache.activemq6.core.server.HornetQComponent;
+import org.apache.activemq6.core.server.HornetQServer;
+import org.apache.activemq6.core.server.HornetQServerLogger;
+import org.jnp.server.Main;
+import org.jnp.server.NamingBeanImpl;
+
+/**
+ * This server class is only used in the standalone mode, its used to control the life cycle of the Naming Server to allow
+ * it to be activated and deactivated
+ *
+ * @author <a href="mailto:andy.taylor@jboss.org">Andy Taylor</a>
+ * 11/8/12
+ */
+public class StandaloneNamingServer implements HornetQComponent
+{
+ private Main jndiServer;
+
+ private HornetQServer server;
+
+ private NamingBeanImpl namingBean;
+
+ private int port = 1099;
+
+ private String bindAddress = "localhost";
+
+ private int rmiPort = 1098;
+
+ private String rmiBindAddress = "localhost";
+
+ private ExecutorService executor;
+
+ public StandaloneNamingServer(HornetQServer server)
+ {
+ this.server = server;
+ }
+
+ @Override
+ public void start() throws Exception
+ {
+ server.registerActivateCallback(new ServerActivateCallback());
+ }
+
+ @Override
+ public void stop() throws Exception
+ {
+ }
+
+ @Override
+ public boolean isStarted()
+ {
+ return false;
+ }
+
+ public void setPort(int port)
+ {
+ this.port = port;
+ }
+
+ public void setBindAddress(String bindAddress)
+ {
+ this.bindAddress = bindAddress;
+ }
+
+ public void setRmiPort(int rmiPort)
+ {
+ this.rmiPort = rmiPort;
+ }
+
+ public void setRmiBindAddress(String rmiBindAddress)
+ {
+ this.rmiBindAddress = rmiBindAddress;
+ }
+
+ private class ServerActivateCallback implements ActivateCallback
+ {
+ private boolean activated = false;
+
+ @Override
+ public synchronized void preActivate()
+ {
+ if (activated)
+ {
+ return;
+ }
+ try
+ {
+ jndiServer = new Main();
+ namingBean = new NamingBeanImpl();
+ jndiServer.setNamingInfo(namingBean);
+ executor = Executors.newCachedThreadPool();
+ jndiServer.setLookupExector(executor);
+ jndiServer.setPort(port);
+ jndiServer.setBindAddress(bindAddress);
+ jndiServer.setRmiPort(rmiPort);
+ jndiServer.setRmiBindAddress(rmiBindAddress);
+ namingBean.start();
+ jndiServer.start();
+ }
+ catch (Exception e)
+ {
+ HornetQServerLogger.LOGGER.unableToStartNamingServer(e);
+ }
+
+ activated = true;
+ }
+
+ @Override
+ public void activated()
+ {
+
+ }
+
+ @Override
+ public synchronized void deActivate()
+ {
+ if (!activated)
+ {
+ return;
+ }
+ if (jndiServer != null)
+ {
+ try
+ {
+ jndiServer.stop();
+ }
+ catch (Exception e)
+ {
+ HornetQServerLogger.LOGGER.unableToStopNamingServer(e);
+ }
+ }
+ if (namingBean != null)
+ {
+ namingBean.stop();
+ }
+ if (executor != null)
+ {
+ executor.shutdown();
+ }
+ activated = false;
+ }
+
+ @Override
+ public void activationComplete()
+ {
+
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/activemq-6/blob/23e8edd9/activemq6-jms-server/src/main/java/org/apache/activemq6/jms/server/management/JMSManagementService.java
----------------------------------------------------------------------
diff --git a/activemq6-jms-server/src/main/java/org/apache/activemq6/jms/server/management/JMSManagementService.java b/activemq6-jms-server/src/main/java/org/apache/activemq6/jms/server/management/JMSManagementService.java
new file mode 100644
index 0000000..a72fbbe
--- /dev/null
+++ b/activemq6-jms-server/src/main/java/org/apache/activemq6/jms/server/management/JMSManagementService.java
@@ -0,0 +1,47 @@
+/*
+ * Copyright 2005-2014 Red Hat, Inc.
+ * Red Hat 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 org.apache.activemq6.jms.server.management;
+
+import org.apache.activemq6.api.jms.management.JMSServerControl;
+import org.apache.activemq6.core.server.Queue;
+import org.apache.activemq6.jms.client.HornetQConnectionFactory;
+import org.apache.activemq6.jms.client.HornetQQueue;
+import org.apache.activemq6.jms.client.HornetQTopic;
+import org.apache.activemq6.jms.server.JMSServerManager;
+import org.apache.activemq6.jms.server.config.ConnectionFactoryConfiguration;
+
+/**
+ * @author <a href="mailto:jmesnil@redhat.com">Jeff Mesnil</a>
+ *
+ *
+ */
+public interface JMSManagementService
+{
+ JMSServerControl registerJMSServer(JMSServerManager server) throws Exception;
+
+ void unregisterJMSServer() throws Exception;
+
+ void registerQueue(HornetQQueue queue, Queue serverQueue) throws Exception;
+
+ void unregisterQueue(String name) throws Exception;
+
+ void registerTopic(HornetQTopic topic) throws Exception;
+
+ void unregisterTopic(String name) throws Exception;
+
+ void registerConnectionFactory(String name, ConnectionFactoryConfiguration config, HornetQConnectionFactory connectionFactory) throws Exception;
+
+ void unregisterConnectionFactory(String name) throws Exception;
+
+ void stop() throws Exception;
+}
http://git-wip-us.apache.org/repos/asf/activemq-6/blob/23e8edd9/activemq6-jms-server/src/main/java/org/apache/activemq6/jms/server/management/JMSNotificationType.java
----------------------------------------------------------------------
diff --git a/activemq6-jms-server/src/main/java/org/apache/activemq6/jms/server/management/JMSNotificationType.java b/activemq6-jms-server/src/main/java/org/apache/activemq6/jms/server/management/JMSNotificationType.java
new file mode 100644
index 0000000..3de5f3e
--- /dev/null
+++ b/activemq6-jms-server/src/main/java/org/apache/activemq6/jms/server/management/JMSNotificationType.java
@@ -0,0 +1,41 @@
+/*
+ * Copyright 2005-2014 Red Hat, Inc.
+ * Red Hat 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 org.apache.activemq6.jms.server.management;
+
+import org.apache.activemq6.api.core.SimpleString;
+import org.apache.activemq6.api.core.management.NotificationType;
+
+public enum JMSNotificationType implements NotificationType
+{
+ QUEUE_CREATED(0),
+ QUEUE_DESTROYED(1),
+ TOPIC_CREATED(2),
+ TOPIC_DESTROYED(3),
+ CONNECTION_FACTORY_CREATED(4),
+ CONNECTION_FACTORY_DESTROYED(5);
+
+ public static final SimpleString MESSAGE = new SimpleString("message");
+
+ private int type;
+
+ private JMSNotificationType(int type)
+ {
+ this.type = type;
+ }
+
+ @Override
+ public int getType()
+ {
+ return type;
+ }
+}
http://git-wip-us.apache.org/repos/asf/activemq-6/blob/23e8edd9/activemq6-jms-server/src/main/java/org/apache/activemq6/jms/server/management/impl/JMSManagementServiceImpl.java
----------------------------------------------------------------------
diff --git a/activemq6-jms-server/src/main/java/org/apache/activemq6/jms/server/management/impl/JMSManagementServiceImpl.java b/activemq6-jms-server/src/main/java/org/apache/activemq6/jms/server/management/impl/JMSManagementServiceImpl.java
new file mode 100644
index 0000000..ee29649
--- /dev/null
+++ b/activemq6-jms-server/src/main/java/org/apache/activemq6/jms/server/management/impl/JMSManagementServiceImpl.java
@@ -0,0 +1,163 @@
+/*
+ * Copyright 2005-2014 Red Hat, Inc.
+ * Red Hat 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 org.apache.activemq6.jms.server.management.impl;
+
+import javax.management.ObjectName;
+
+import org.apache.activemq6.api.core.management.AddressControl;
+import org.apache.activemq6.api.core.management.QueueControl;
+import org.apache.activemq6.api.core.management.ResourceNames;
+import org.apache.activemq6.api.jms.management.ConnectionFactoryControl;
+import org.apache.activemq6.api.jms.management.JMSQueueControl;
+import org.apache.activemq6.api.jms.management.JMSServerControl;
+import org.apache.activemq6.api.jms.management.TopicControl;
+import org.apache.activemq6.core.messagecounter.MessageCounter;
+import org.apache.activemq6.core.messagecounter.MessageCounterManager;
+import org.apache.activemq6.core.server.HornetQServer;
+import org.apache.activemq6.core.server.Queue;
+import org.apache.activemq6.core.server.management.ManagementService;
+import org.apache.activemq6.jms.client.HornetQConnectionFactory;
+import org.apache.activemq6.jms.client.HornetQQueue;
+import org.apache.activemq6.jms.client.HornetQTopic;
+import org.apache.activemq6.jms.management.impl.JMSConnectionFactoryControlImpl;
+import org.apache.activemq6.jms.management.impl.JMSQueueControlImpl;
+import org.apache.activemq6.jms.management.impl.JMSServerControlImpl;
+import org.apache.activemq6.jms.management.impl.JMSTopicControlImpl;
+import org.apache.activemq6.jms.server.JMSServerManager;
+import org.apache.activemq6.jms.server.config.ConnectionFactoryConfiguration;
+import org.apache.activemq6.jms.server.management.JMSManagementService;
+
+/*
+ * @author <a href="mailto:jmesnil@redhat.com">Jeff Mesnil</a>
+ *
+ */
+public class JMSManagementServiceImpl implements JMSManagementService
+{
+
+ // Constants -----------------------------------------------------
+
+ // Attributes ----------------------------------------------------
+
+ private final ManagementService managementService;
+
+ private final JMSServerManager jmsServerManager;
+
+ // Static --------------------------------------------------------
+
+ public JMSManagementServiceImpl(final ManagementService managementService, final HornetQServer server, final JMSServerManager jmsServerManager)
+ {
+ this.managementService = managementService;
+ this.jmsServerManager = jmsServerManager;
+ }
+
+ // Public --------------------------------------------------------
+
+ // JMSManagementRegistration implementation ----------------------
+
+ public synchronized JMSServerControl registerJMSServer(final JMSServerManager server) throws Exception
+ {
+ ObjectName objectName = managementService.getObjectNameBuilder().getJMSServerObjectName();
+ JMSServerControlImpl control = new JMSServerControlImpl(server);
+ managementService.registerInJMX(objectName, control);
+ managementService.registerInRegistry(ResourceNames.JMS_SERVER, control);
+ return control;
+ }
+
+ public synchronized void unregisterJMSServer() throws Exception
+ {
+ ObjectName objectName = managementService.getObjectNameBuilder().getJMSServerObjectName();
+ managementService.unregisterFromJMX(objectName);
+ managementService.unregisterFromRegistry(ResourceNames.JMS_SERVER);
+ }
+
+ public synchronized void registerQueue(final HornetQQueue queue, final Queue serverQueue) throws Exception
+ {
+ QueueControl coreQueueControl = (QueueControl)managementService.getResource(ResourceNames.CORE_QUEUE + queue.getAddress());
+ MessageCounterManager messageCounterManager = managementService.getMessageCounterManager();
+ MessageCounter counter = new MessageCounter(queue.getName(),
+ null,
+ serverQueue,
+ false,
+ coreQueueControl.isDurable(),
+ messageCounterManager.getMaxDayCount());
+ messageCounterManager.registerMessageCounter(queue.getName(), counter);
+ ObjectName objectName = managementService.getObjectNameBuilder().getJMSQueueObjectName(queue.getQueueName());
+ JMSQueueControlImpl control = new JMSQueueControlImpl(queue, coreQueueControl, jmsServerManager, counter);
+ managementService.registerInJMX(objectName, control);
+ managementService.registerInRegistry(ResourceNames.JMS_QUEUE + queue.getQueueName(), control);
+ }
+
+ public synchronized void unregisterQueue(final String name) throws Exception
+ {
+ ObjectName objectName = managementService.getObjectNameBuilder().getJMSQueueObjectName(name);
+ managementService.unregisterFromJMX(objectName);
+ managementService.unregisterFromRegistry(ResourceNames.JMS_QUEUE + name);
+ }
+
+ public synchronized void registerTopic(final HornetQTopic topic) throws Exception
+ {
+ ObjectName objectName = managementService.getObjectNameBuilder().getJMSTopicObjectName(topic.getTopicName());
+ AddressControl addressControl = (AddressControl)managementService.getResource(ResourceNames.CORE_ADDRESS + topic.getAddress());
+ JMSTopicControlImpl control = new JMSTopicControlImpl(topic, jmsServerManager, addressControl, managementService);
+ managementService.registerInJMX(objectName, control);
+ managementService.registerInRegistry(ResourceNames.JMS_TOPIC + topic.getTopicName(), control);
+ }
+
+ public synchronized void unregisterTopic(final String name) throws Exception
+ {
+ ObjectName objectName = managementService.getObjectNameBuilder().getJMSTopicObjectName(name);
+ managementService.unregisterFromJMX(objectName);
+ managementService.unregisterFromRegistry(ResourceNames.JMS_TOPIC + name);
+ }
+
+ public synchronized void registerConnectionFactory(final String name,
+ final ConnectionFactoryConfiguration cfConfig,
+ final HornetQConnectionFactory connectionFactory) throws Exception
+ {
+ ObjectName objectName = managementService.getObjectNameBuilder().getConnectionFactoryObjectName(name);
+ JMSConnectionFactoryControlImpl control = new JMSConnectionFactoryControlImpl(cfConfig, connectionFactory, jmsServerManager, name);
+ managementService.registerInJMX(objectName, control);
+ managementService.registerInRegistry(ResourceNames.JMS_CONNECTION_FACTORY + name, control);
+ }
+
+ public synchronized void unregisterConnectionFactory(final String name) throws Exception
+ {
+ ObjectName objectName = managementService.getObjectNameBuilder().getConnectionFactoryObjectName(name);
+ managementService.unregisterFromJMX(objectName);
+ managementService.unregisterFromRegistry(ResourceNames.JMS_CONNECTION_FACTORY + name);
+ }
+
+ public void stop() throws Exception
+ {
+ for (Object resource : managementService.getResources(ConnectionFactoryControl.class))
+ {
+ unregisterConnectionFactory(((ConnectionFactoryControl)resource).getName());
+ }
+ for (Object resource : managementService.getResources(JMSQueueControl.class))
+ {
+ unregisterQueue(((JMSQueueControl)resource).getName());
+ }
+ for (Object resource : managementService.getResources(TopicControl.class))
+ {
+ unregisterTopic(((TopicControl)resource).getName());
+ }
+ }
+
+ // Package protected ---------------------------------------------
+
+ // Protected -----------------------------------------------------
+
+ // Private -------------------------------------------------------
+
+ // Inner classes -------------------------------------------------
+}