From 5ec5909ac0784635cfd43546ea900a22ed402db3 Mon Sep 17 00:00:00 2001 From: Rupert Smith Date: Wed, 28 Nov 2007 15:52:11 +0000 Subject: Merged revisions 598371-598586 via svnmerge from https://svn.apache.org/repos/asf/incubator/qpid/branches/M2.1 ........ r598371 | ritchiem | 2007-11-26 18:17:12 +0000 (Mon, 26 Nov 2007) | 1 line Qpid-559 : OSGi Patch Provided by Aidan Skinner ........ r598572 | ritchiem | 2007-11-27 09:44:28 +0000 (Tue, 27 Nov 2007) | 1 line Add Target to ignores list. ........ r598586 | rupertlssmith | 2007-11-27 10:51:29 +0000 (Tue, 27 Nov 2007) | 1 line Now outputs test results in xml format to the surefire reports directory. ........ git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/branches/M2.1.1@599036 13f79535-47bb-0310-9956-ffa450edef68 --- java/broker/pom.xml | 10 + .../server/exchange/DefaultExchangeFactory.java | 72 +++++-- .../qpid/server/exchange/DestNameExchange.java | 31 +++ .../qpid/server/exchange/DestWildExchange.java | 33 +++ .../qpid/server/exchange/ExchangeFactory.java | 8 + .../apache/qpid/server/exchange/ExchangeType.java | 35 ++++ .../qpid/server/exchange/FanoutExchange.java | 31 +++ .../qpid/server/exchange/HeadersExchange.java | 32 +++ .../org/apache/qpid/server/plugins/Activator.java | 44 ++++ .../apache/qpid/server/plugins/PluginManager.java | 147 ++++++++++++++ .../qpid/server/protocol/ExchangeInitialiser.java | 12 +- .../ConfigurationFileApplicationRegistry.java | 10 + .../qpid/server/registry/IApplicationRegistry.java | 4 + .../qpid/server/util/NullApplicationRegistry.java | 11 +- .../qpid/server/virtualhost/VirtualHost.java | 1 + java/plugins/pom.xml | 100 ++++++++++ .../java/org/apache/qpid/extras/Activator.java | 28 +++ .../exchanges/diagnostic/DiagnosticExchange.java | 221 +++++++++++++++++++++ .../diagnostic/DiagnosticExchangeType.java | 57 ++++++ .../extras/exchanges/example/TestExchange.java | 97 +++++++++ .../extras/exchanges/example/TestExchangeType.java | 36 ++++ java/pom.xml | 1 + java/systests/pom.xml | 13 +- .../org/apache/qpid/server/plugins/PluginTest.java | 35 ++++ .../qpid/server/util/TestApplicationRegistry.java | 7 + 25 files changed, 1048 insertions(+), 28 deletions(-) create mode 100644 java/broker/src/main/java/org/apache/qpid/server/exchange/ExchangeType.java create mode 100644 java/broker/src/main/java/org/apache/qpid/server/plugins/Activator.java create mode 100644 java/broker/src/main/java/org/apache/qpid/server/plugins/PluginManager.java create mode 100644 java/plugins/pom.xml create mode 100644 java/plugins/src/main/java/org/apache/qpid/extras/Activator.java create mode 100644 java/plugins/src/main/java/org/apache/qpid/extras/exchanges/diagnostic/DiagnosticExchange.java create mode 100644 java/plugins/src/main/java/org/apache/qpid/extras/exchanges/diagnostic/DiagnosticExchangeType.java create mode 100644 java/plugins/src/main/java/org/apache/qpid/extras/exchanges/example/TestExchange.java create mode 100644 java/plugins/src/main/java/org/apache/qpid/extras/exchanges/example/TestExchangeType.java create mode 100644 java/systests/src/main/java/org/apache/qpid/server/plugins/PluginTest.java diff --git a/java/broker/pom.xml b/java/broker/pom.xml index feb503e39f..80272f184e 100644 --- a/java/broker/pom.xml +++ b/java/broker/pom.xml @@ -89,6 +89,16 @@ test + + org.apache.felix + org.osgi.core + 1.0.0 + + + org.apache.felix + org.apache.felix.framework + 1.0.0 + diff --git a/java/broker/src/main/java/org/apache/qpid/server/exchange/DefaultExchangeFactory.java b/java/broker/src/main/java/org/apache/qpid/server/exchange/DefaultExchangeFactory.java index c349b44d6d..8ede553464 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/exchange/DefaultExchangeFactory.java +++ b/java/broker/src/main/java/org/apache/qpid/server/exchange/DefaultExchangeFactory.java @@ -20,56 +20,88 @@ */ package org.apache.qpid.server.exchange; +import java.util.Collection; import java.util.HashMap; import java.util.Map; import org.apache.log4j.Logger; +import org.apache.commons.configuration.Configuration; + import org.apache.qpid.AMQException; import org.apache.qpid.AMQUnknownExchangeType; -import org.apache.qpid.exchange.ExchangeDefaults; import org.apache.qpid.framing.AMQShortString; +import org.apache.qpid.server.registry.ApplicationRegistry; import org.apache.qpid.server.virtualhost.VirtualHost; public class DefaultExchangeFactory implements ExchangeFactory { private static final Logger _logger = Logger.getLogger(DefaultExchangeFactory.class); - private Map> _exchangeClassMap = new HashMap>(); + private Map> _exchangeClassMap = new HashMap>(); private final VirtualHost _host; public DefaultExchangeFactory(VirtualHost host) { _host = host; - _exchangeClassMap.put(ExchangeDefaults.DIRECT_EXCHANGE_CLASS, org.apache.qpid.server.exchange.DestNameExchange.class); - _exchangeClassMap.put(ExchangeDefaults.TOPIC_EXCHANGE_CLASS, org.apache.qpid.server.exchange.DestWildExchange.class); - _exchangeClassMap.put(ExchangeDefaults.HEADERS_EXCHANGE_CLASS, org.apache.qpid.server.exchange.HeadersExchange.class); - _exchangeClassMap.put(ExchangeDefaults.FANOUT_EXCHANGE_CLASS, org.apache.qpid.server.exchange.FanoutExchange.class); + registerExchangeType(DestNameExchange.TYPE); + registerExchangeType(DestWildExchange.TYPE); + registerExchangeType(HeadersExchange.TYPE); + registerExchangeType(FanoutExchange.TYPE); + } + public void registerExchangeType(ExchangeType type) + { + _exchangeClassMap.put(type.getName(), type); + } + + public Collection> getRegisteredTypes() + { + return _exchangeClassMap.values(); } public Exchange createExchange(AMQShortString exchange, AMQShortString type, boolean durable, boolean autoDelete, int ticket) throws AMQException { - Class exchClass = _exchangeClassMap.get(type); - if (exchClass == null) + ExchangeType exchType = _exchangeClassMap.get(type); + if (exchType == null) { throw new AMQUnknownExchangeType("Unknown exchange type: " + type); } - try - { - Exchange e = exchClass.newInstance(); - e.initialise(_host, exchange, durable, ticket, autoDelete); - return e; - } - catch (InstantiationException e) - { - throw new AMQException("Unable to create exchange: " + e, e); - } - catch (IllegalAccessException e) + Exchange e = exchType.newInstance(_host, exchange, durable, ticket, autoDelete); + return e; + } + + public void initialise(Configuration hostConfig) + { + for(Object className : hostConfig.getList("custom-exchanges.class-name")) { - throw new AMQException("Unable to create exchange: " + e, e); + try + { + ExchangeType exchangeType = ApplicationRegistry.getInstance().getPluginManager().getExchanges().get(String.valueOf(className)); + if (exchangeType == null) + { + _logger.error("No such custom exchange class found: \""+String.valueOf(className)+"\""); + return; + } + Class exchangeTypeClass = exchangeType.getClass(); + ExchangeType type = exchangeTypeClass.newInstance(); + registerExchangeType(type); + } + catch (ClassCastException classCastEx) + { + _logger.error("No custom exchange class: \""+String.valueOf(className)+"\" cannot be registered as it does not extend class \""+ExchangeType.class+"\""); + } + catch (IllegalAccessException e) + { + _logger.error("Cannot create custom exchange class: \""+String.valueOf(className)+"\"",e); + } + catch (InstantiationException e) + { + _logger.error("Cannot create custom exchange class: \""+String.valueOf(className)+"\"",e); + } } + } } diff --git a/java/broker/src/main/java/org/apache/qpid/server/exchange/DestNameExchange.java b/java/broker/src/main/java/org/apache/qpid/server/exchange/DestNameExchange.java index 5edffc19ed..85d478839c 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/exchange/DestNameExchange.java +++ b/java/broker/src/main/java/org/apache/qpid/server/exchange/DestNameExchange.java @@ -42,6 +42,7 @@ import org.apache.qpid.server.management.MBeanConstructor; import org.apache.qpid.server.management.MBeanDescription; import org.apache.qpid.server.queue.AMQMessage; import org.apache.qpid.server.queue.AMQQueue; +import org.apache.qpid.server.virtualhost.VirtualHost; public class DestNameExchange extends AbstractExchange { @@ -52,6 +53,36 @@ public class DestNameExchange extends AbstractExchange */ private final Index _index = new Index(); + public static final ExchangeType TYPE = new ExchangeType() + { + + public AMQShortString getName() + { + return ExchangeDefaults.DIRECT_EXCHANGE_CLASS; + } + + public Class getExchangeClass() + { + return DestNameExchange.class; + } + + public DestNameExchange newInstance(VirtualHost host, + AMQShortString name, + boolean durable, + int ticket, + boolean autoDelete) throws AMQException + { + DestNameExchange exch = new DestNameExchange(); + exch.initialise(host,name,durable,ticket,autoDelete); + return exch; + } + + public AMQShortString getDefaultExchangeName() + { + return ExchangeDefaults.DIRECT_EXCHANGE_NAME; + } + }; + /** * MBean class implementing the management interfaces. */ diff --git a/java/broker/src/main/java/org/apache/qpid/server/exchange/DestWildExchange.java b/java/broker/src/main/java/org/apache/qpid/server/exchange/DestWildExchange.java index b55dbcc792..c6f4dcf143 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/exchange/DestWildExchange.java +++ b/java/broker/src/main/java/org/apache/qpid/server/exchange/DestWildExchange.java @@ -30,6 +30,7 @@ import org.apache.qpid.server.management.MBeanConstructor; import org.apache.qpid.server.management.MBeanDescription; import org.apache.qpid.server.queue.AMQMessage; import org.apache.qpid.server.queue.AMQQueue; +import org.apache.qpid.server.virtualhost.VirtualHost; import javax.management.JMException; import javax.management.MBeanException; @@ -48,6 +49,38 @@ import java.util.concurrent.CopyOnWriteArrayList; public class DestWildExchange extends AbstractExchange { + + public static final ExchangeType TYPE = new ExchangeType() + { + + public AMQShortString getName() + { + return ExchangeDefaults.TOPIC_EXCHANGE_CLASS; + } + + public Class getExchangeClass() + { + return DestWildExchange.class; + } + + public DestWildExchange newInstance(VirtualHost host, + AMQShortString name, + boolean durable, + int ticket, + boolean autoDelete) throws AMQException + { + DestWildExchange exch = new DestWildExchange(); + exch.initialise(host, name, durable, ticket, autoDelete); + return exch; + } + + public AMQShortString getDefaultExchangeName() + { + return ExchangeDefaults.TOPIC_EXCHANGE_NAME; + } + }; + + private static final Logger _logger = Logger.getLogger(DestWildExchange.class); private ConcurrentHashMap> _routingKey2queues = diff --git a/java/broker/src/main/java/org/apache/qpid/server/exchange/ExchangeFactory.java b/java/broker/src/main/java/org/apache/qpid/server/exchange/ExchangeFactory.java index e07fd0b8fc..0bcfec7181 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/exchange/ExchangeFactory.java +++ b/java/broker/src/main/java/org/apache/qpid/server/exchange/ExchangeFactory.java @@ -20,6 +20,10 @@ */ package org.apache.qpid.server.exchange; +import java.util.Collection; + +import org.apache.commons.configuration.Configuration; + import org.apache.qpid.AMQException; import org.apache.qpid.framing.AMQShortString; @@ -29,4 +33,8 @@ public interface ExchangeFactory Exchange createExchange(AMQShortString exchange, AMQShortString type, boolean durable, boolean autoDelete, int ticket) throws AMQException; + + void initialise(Configuration hostConfig); + + Collection> getRegisteredTypes(); } diff --git a/java/broker/src/main/java/org/apache/qpid/server/exchange/ExchangeType.java b/java/broker/src/main/java/org/apache/qpid/server/exchange/ExchangeType.java new file mode 100644 index 0000000000..0b55caa2f1 --- /dev/null +++ b/java/broker/src/main/java/org/apache/qpid/server/exchange/ExchangeType.java @@ -0,0 +1,35 @@ +/* + * + * 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 org.apache.qpid.server.exchange; + +import org.apache.qpid.framing.AMQShortString; +import org.apache.qpid.AMQException; +import org.apache.qpid.server.virtualhost.VirtualHost; + + +public interface ExchangeType +{ + public AMQShortString getName(); + public Class getExchangeClass(); + public T newInstance(VirtualHost host, AMQShortString name, + boolean durable, int ticket, boolean autoDelete) throws AMQException; + public AMQShortString getDefaultExchangeName(); +} diff --git a/java/broker/src/main/java/org/apache/qpid/server/exchange/FanoutExchange.java b/java/broker/src/main/java/org/apache/qpid/server/exchange/FanoutExchange.java index aa13f1d8ee..f7af44b232 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/exchange/FanoutExchange.java +++ b/java/broker/src/main/java/org/apache/qpid/server/exchange/FanoutExchange.java @@ -30,6 +30,7 @@ import org.apache.qpid.server.management.MBeanConstructor; import org.apache.qpid.server.management.MBeanDescription; import org.apache.qpid.server.queue.AMQMessage; import org.apache.qpid.server.queue.AMQQueue; +import org.apache.qpid.server.virtualhost.VirtualHost; import javax.management.JMException; import javax.management.MBeanException; @@ -115,6 +116,36 @@ public class FanoutExchange extends AbstractExchange } } + public static final ExchangeType TYPE = new ExchangeType() + { + + public AMQShortString getName() + { + return ExchangeDefaults.FANOUT_EXCHANGE_CLASS; + } + + public Class getExchangeClass() + { + return FanoutExchange.class; + } + + public FanoutExchange newInstance(VirtualHost host, + AMQShortString name, + boolean durable, + int ticket, + boolean autoDelete) throws AMQException + { + FanoutExchange exch = new FanoutExchange(); + exch.initialise(host, name, durable, ticket, autoDelete); + return exch; + } + + public AMQShortString getDefaultExchangeName() + { + return ExchangeDefaults.DIRECT_EXCHANGE_NAME; + } + }; + public Map> getBindings() { return null; diff --git a/java/broker/src/main/java/org/apache/qpid/server/exchange/HeadersExchange.java b/java/broker/src/main/java/org/apache/qpid/server/exchange/HeadersExchange.java index 3544e9d1f8..e97ea78252 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/exchange/HeadersExchange.java +++ b/java/broker/src/main/java/org/apache/qpid/server/exchange/HeadersExchange.java @@ -32,6 +32,7 @@ import org.apache.qpid.server.management.MBeanConstructor; import org.apache.qpid.server.management.MBeanDescription; import org.apache.qpid.server.queue.AMQMessage; import org.apache.qpid.server.queue.AMQQueue; +import org.apache.qpid.server.virtualhost.VirtualHost; import javax.management.JMException; import javax.management.openmbean.ArrayType; @@ -81,6 +82,37 @@ public class HeadersExchange extends AbstractExchange { private static final Logger _logger = Logger.getLogger(HeadersExchange.class); + + + public static final ExchangeType TYPE = new ExchangeType() + { + + public AMQShortString getName() + { + return ExchangeDefaults.HEADERS_EXCHANGE_CLASS; + } + + public Class getExchangeClass() + { + return HeadersExchange.class; + } + + public HeadersExchange newInstance(VirtualHost host, AMQShortString name, boolean durable, int ticket, + boolean autoDelete) throws AMQException + { + HeadersExchange exch = new HeadersExchange(); + exch.initialise(host, name, durable, ticket, autoDelete); + return exch; + } + + public AMQShortString getDefaultExchangeName() + { + + return ExchangeDefaults.HEADERS_EXCHANGE_NAME; + } + }; + + private final List _bindings = new CopyOnWriteArrayList(); /** diff --git a/java/broker/src/main/java/org/apache/qpid/server/plugins/Activator.java b/java/broker/src/main/java/org/apache/qpid/server/plugins/Activator.java new file mode 100644 index 0000000000..b0ebf197f9 --- /dev/null +++ b/java/broker/src/main/java/org/apache/qpid/server/plugins/Activator.java @@ -0,0 +1,44 @@ +/* + * 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 org.apache.qpid.server.plugins; + +import org.osgi.framework.BundleActivator; +import org.osgi.framework.BundleContext; + +public class Activator implements BundleActivator +{ + + BundleContext _context = null; + + public void start(BundleContext ctx) throws Exception + { + _context = ctx; + } + + public void stop(BundleContext ctx) throws Exception + { + start(null); + } + + public BundleContext getContext() + { + return _context; + } +} diff --git a/java/broker/src/main/java/org/apache/qpid/server/plugins/PluginManager.java b/java/broker/src/main/java/org/apache/qpid/server/plugins/PluginManager.java new file mode 100644 index 0000000000..be22a90d0b --- /dev/null +++ b/java/broker/src/main/java/org/apache/qpid/server/plugins/PluginManager.java @@ -0,0 +1,147 @@ +/* + * 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 org.apache.qpid.server.plugins; + +import java.io.File; +import java.util.ArrayList; +import java.util.Collection; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import org.apache.felix.framework.Felix; +import org.apache.felix.framework.cache.BundleCache; +import org.apache.felix.framework.util.FelixConstants; +import org.apache.felix.framework.util.StringMap; +import org.apache.qpid.server.exchange.Exchange; +import org.apache.qpid.server.exchange.ExchangeType; +import org.osgi.framework.BundleActivator; +import org.osgi.framework.BundleException; +import org.osgi.util.tracker.ServiceTracker; + +/** + * + * @author aidan + * + * Provides access to pluggable elements, such as exchanges + */ + +public class PluginManager +{ + + private Felix _felix = null; + private ServiceTracker _exchangeTracker = null; + private Activator _activator = null; + private boolean _empty; + + public PluginManager(String plugindir) throws Exception + { + StringMap configMap = new StringMap(false); + + // Tell felix it's being embedded + configMap.put(FelixConstants.EMBEDDED_EXECUTION_PROP, "true"); + // Add the bundle provided service interface package and the core OSGi + // packages to be exported from the class path via the system bundle. + configMap.put(FelixConstants.FRAMEWORK_SYSTEMPACKAGES, "org.osgi.framework; version=1.3.0," + + "org.osgi.service.packageadmin; version=1.2.0," + + "org.osgi.service.startlevel; version=1.0.0," + + "org.osgi.service.url; version=1.0.0," + + "org.apache.qpid.framing; version=0.2.1," + + "org.apache.qpid.server.exchange; version=0.2.1," + + "org.apache.qpid.server.management; version=0.2.1,"+ + "org.apache.qpid.protocol; version=0.2.1,"+ + "org.apache.qpid.server.virtualhost; version=0.2.1," + + "org.apache.qpid; version=0.2.1," + + "org.apache.qpid.server.queue; version=0.2.1," + + "javax.management.openmbean; version=1.0.0,"+ + "javax.management; version=1.0.0,"+ + "uk.co.thebadgerset.junit.extensions.util; version=0.6.1," + ); + + if (plugindir == null) + { + _empty = true; + return; + } + + // Set the list of bundles to load + File dir = new File(plugindir); + if (!dir.exists()) + { + _empty = true; + return; + } + StringBuffer pluginJars = new StringBuffer(); + + if (dir.isDirectory()) + { + for (String child : dir.list()) + { + if (child.endsWith("jar")) + { + pluginJars.append(String.format(" file:%s%s%s", plugindir,File.separator,child)); + } + } + } + if (pluginJars.length() == 0) + { + _empty = true; + return; + } + + configMap.put(FelixConstants.AUTO_START_PROP + ".1", pluginJars.toString()); + configMap.put(BundleCache.CACHE_PROFILE_DIR_PROP, plugindir); + + List activators = new ArrayList(); + _activator = new Activator(); + activators.add(_activator); + + _felix = new Felix(configMap, activators); + try + { + _felix.start(); + _exchangeTracker = new ServiceTracker(_activator.getContext(), ExchangeType.class.getName(), null); + _exchangeTracker.open(); + } + catch (BundleException e) + { + throw new Exception("Could not create bundle"); + } + } + + public Map> getExchanges() + { + if (_empty) + { + return null; + } + Map>exchanges = new HashMap>(); + for (Object service : _exchangeTracker.getServices()) + { + if (service instanceof ExchangeType) + { + exchanges.put(service.getClass().getName(), (ExchangeType) service); + } + } + + return exchanges; + } + +} diff --git a/java/broker/src/main/java/org/apache/qpid/server/protocol/ExchangeInitialiser.java b/java/broker/src/main/java/org/apache/qpid/server/protocol/ExchangeInitialiser.java index 29d55ce763..2abcecb6de 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/protocol/ExchangeInitialiser.java +++ b/java/broker/src/main/java/org/apache/qpid/server/protocol/ExchangeInitialiser.java @@ -23,18 +23,20 @@ package org.apache.qpid.server.protocol; import org.apache.qpid.AMQException; import org.apache.qpid.exchange.ExchangeDefaults; import org.apache.qpid.framing.AMQShortString; +import org.apache.qpid.server.exchange.Exchange; import org.apache.qpid.server.exchange.ExchangeFactory; import org.apache.qpid.server.exchange.ExchangeRegistry; +import org.apache.qpid.server.exchange.ExchangeType; public class ExchangeInitialiser { public void initialise(ExchangeFactory factory, ExchangeRegistry registry) throws AMQException{ + for (ExchangeType type : factory.getRegisteredTypes()) + { + define (registry, factory, type.getDefaultExchangeName(), type.getName()); + } + define(registry, factory, ExchangeDefaults.DEFAULT_EXCHANGE_NAME, ExchangeDefaults.DIRECT_EXCHANGE_CLASS); - define(registry, factory, ExchangeDefaults.DIRECT_EXCHANGE_NAME, ExchangeDefaults.DIRECT_EXCHANGE_CLASS); - define(registry, factory, ExchangeDefaults.TOPIC_EXCHANGE_NAME, ExchangeDefaults.TOPIC_EXCHANGE_CLASS); - define(registry, factory, ExchangeDefaults.HEADERS_EXCHANGE_NAME, ExchangeDefaults.HEADERS_EXCHANGE_CLASS); - define(registry, factory, ExchangeDefaults.FANOUT_EXCHANGE_NAME, ExchangeDefaults.FANOUT_EXCHANGE_CLASS); - registry.setDefaultExchange(registry.getExchange(ExchangeDefaults.DEFAULT_EXCHANGE_NAME)); } diff --git a/java/broker/src/main/java/org/apache/qpid/server/registry/ConfigurationFileApplicationRegistry.java b/java/broker/src/main/java/org/apache/qpid/server/registry/ConfigurationFileApplicationRegistry.java index 1cca259a8d..42c32dcf00 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/registry/ConfigurationFileApplicationRegistry.java +++ b/java/broker/src/main/java/org/apache/qpid/server/registry/ConfigurationFileApplicationRegistry.java @@ -34,6 +34,7 @@ import org.apache.qpid.server.management.JMXManagedObjectRegistry; import org.apache.qpid.server.management.ManagedObjectRegistry; import org.apache.qpid.server.management.ManagementConfiguration; import org.apache.qpid.server.management.NoopManagedObjectRegistry; +import org.apache.qpid.server.plugins.PluginManager; import org.apache.qpid.server.security.auth.manager.AuthenticationManager; import org.apache.qpid.server.security.auth.database.ConfigurationFilePrincipalDatabaseManager; import org.apache.qpid.server.security.auth.database.PrincipalDatabaseManager; @@ -60,6 +61,8 @@ public class ConfigurationFileApplicationRegistry extends ApplicationRegistry private final Map _virtualHosts = new ConcurrentHashMap(); + private PluginManager _pluginManager; + public ConfigurationFileApplicationRegistry(File configurationURL) throws ConfigurationException { @@ -117,6 +120,8 @@ public class ConfigurationFileApplicationRegistry extends ApplicationRegistry _managedObjectRegistry.start(); + _pluginManager = new PluginManager(_configuration.getString("plugin-directory")); + initialiseVirtualHosts(); } @@ -173,4 +178,9 @@ public class ConfigurationFileApplicationRegistry extends ApplicationRegistry { return getConfiguration().getList("virtualhosts.virtualhost.name"); } + + public PluginManager getPluginManager() + { + return _pluginManager; + } } diff --git a/java/broker/src/main/java/org/apache/qpid/server/registry/IApplicationRegistry.java b/java/broker/src/main/java/org/apache/qpid/server/registry/IApplicationRegistry.java index 5a48431288..6aac21a161 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/registry/IApplicationRegistry.java +++ b/java/broker/src/main/java/org/apache/qpid/server/registry/IApplicationRegistry.java @@ -24,6 +24,7 @@ import java.util.Collection; import org.apache.commons.configuration.Configuration; import org.apache.qpid.server.management.ManagedObjectRegistry; +import org.apache.qpid.server.plugins.PluginManager; import org.apache.qpid.server.security.auth.manager.AuthenticationManager; import org.apache.qpid.server.security.auth.database.PrincipalDatabaseManager; import org.apache.qpid.server.security.access.AccessManager; @@ -68,4 +69,7 @@ public interface IApplicationRegistry VirtualHostRegistry getVirtualHostRegistry(); AccessManager getAccessManager(); + + PluginManager getPluginManager(); + } diff --git a/java/broker/src/main/java/org/apache/qpid/server/util/NullApplicationRegistry.java b/java/broker/src/main/java/org/apache/qpid/server/util/NullApplicationRegistry.java index 150b98b424..e9fa642175 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/util/NullApplicationRegistry.java +++ b/java/broker/src/main/java/org/apache/qpid/server/util/NullApplicationRegistry.java @@ -29,6 +29,7 @@ import org.apache.commons.configuration.Configuration; import org.apache.commons.configuration.MapConfiguration; import org.apache.qpid.server.management.ManagedObjectRegistry; import org.apache.qpid.server.management.NoopManagedObjectRegistry; +import org.apache.qpid.server.plugins.PluginManager; import org.apache.qpid.server.registry.ApplicationRegistry; import org.apache.qpid.server.security.auth.manager.AuthenticationManager; import org.apache.qpid.server.security.auth.manager.PrincipalDatabaseAuthenticationManager; @@ -51,6 +52,8 @@ public class NullApplicationRegistry extends ApplicationRegistry private PrincipalDatabaseManager _databaseManager; + private PluginManager _pluginManager; + public NullApplicationRegistry() { @@ -76,7 +79,7 @@ public class NullApplicationRegistry extends ApplicationRegistry VirtualHost dummyHost = new VirtualHost("test", getConfiguration()); _virtualHostRegistry.registerVirtualHost(dummyHost); _virtualHostRegistry.setDefaultVirtualHostName("test"); - + _pluginManager = new PluginManager(""); _configuration.addProperty("heartbeat.delay", 10 * 60); // 10 minutes } @@ -117,6 +120,12 @@ public class NullApplicationRegistry extends ApplicationRegistry { return _accessManager; } + + public PluginManager getPluginManager() + { + return _pluginManager; + } } + diff --git a/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java b/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java index b95772b680..8d6a26fdbc 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java +++ b/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java @@ -139,6 +139,7 @@ public class VirtualHost implements Accessable _queueRegistry = new DefaultQueueRegistry(this); _exchangeFactory = new DefaultExchangeFactory(this); + _exchangeFactory.initialise(hostConfig); _exchangeRegistry = new DefaultExchangeRegistry(this); if (store != null) diff --git a/java/plugins/pom.xml b/java/plugins/pom.xml new file mode 100644 index 0000000000..51d22ab01e --- /dev/null +++ b/java/plugins/pom.xml @@ -0,0 +1,100 @@ + + + + + . + 1.5 + -Xlint:fallthrough,finally + 2.0.1 + + + + org.apache + apache + 4 + + + 4.0.0 + bundle + Qpid Plugins + A simple plugin for qpid. + org.apache.qpid.extras + example-plugin + 1.0-incubating-M2.1-SNAPSHOT + + + org.apache.felix + org.osgi.core + 1.0.0 + + + org.apache.qpid + qpid-broker + 1.0-incubating-M2.1-SNAPSHOT + + + uk.co.thebadgerset + junit-toolkit + 0.6.1 + + + + + + + + org.apache.felix + maven-bundle-plugin + 1.0.0 + true + + + + org.apache.qpid.extras.* + + + org.apache.qpid.extras.Activator + + + + + + + org.apache.maven.plugins + maven-compiler-plugin + ${compiler.version} + + ${java.source.version} + ${java.source.version} + + + + + + src/main/java + + **/*.java + **/log4j.properties + + + + + + diff --git a/java/plugins/src/main/java/org/apache/qpid/extras/Activator.java b/java/plugins/src/main/java/org/apache/qpid/extras/Activator.java new file mode 100644 index 0000000000..a9292de7cd --- /dev/null +++ b/java/plugins/src/main/java/org/apache/qpid/extras/Activator.java @@ -0,0 +1,28 @@ +package org.apache.qpid.extras; + +import org.apache.qpid.extras.exchanges.diagnostic.DiagnosticExchangeType; +import org.apache.qpid.extras.exchanges.example.TestExchangeType; +import org.apache.qpid.server.exchange.ExchangeType; +import org.osgi.framework.BundleActivator; +import org.osgi.framework.BundleContext; + +/** + * + * @author aidan + * + * Dummy class, used by PluginTest + */ + +public class Activator implements BundleActivator +{ + + public void start(BundleContext ctx) throws Exception + { + ctx.registerService(ExchangeType.class.getName(), new TestExchangeType(), null); + ctx.registerService(ExchangeType.class.getName(), new DiagnosticExchangeType(), null); + } + + public void stop(BundleContext ctx) throws Exception + { + } +} diff --git a/java/plugins/src/main/java/org/apache/qpid/extras/exchanges/diagnostic/DiagnosticExchange.java b/java/plugins/src/main/java/org/apache/qpid/extras/exchanges/diagnostic/DiagnosticExchange.java new file mode 100644 index 0000000000..e7d899983e --- /dev/null +++ b/java/plugins/src/main/java/org/apache/qpid/extras/exchanges/diagnostic/DiagnosticExchange.java @@ -0,0 +1,221 @@ +/* + * + * 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 org.apache.qpid.extras.exchanges.diagnostic; + +import java.lang.instrument.Instrumentation; +import java.util.List; +import java.util.Map; + +import javax.management.JMException; +import javax.management.openmbean.OpenDataException; +import javax.management.openmbean.TabularData; + +import org.apache.qpid.AMQException; +import org.apache.qpid.framing.AMQShortString; +import org.apache.qpid.framing.BasicContentHeaderProperties; +import org.apache.qpid.framing.FieldTable; +import org.apache.qpid.server.exchange.AbstractExchange; +import org.apache.qpid.server.management.MBeanConstructor; +import org.apache.qpid.server.management.MBeanDescription; +import org.apache.qpid.server.queue.AMQMessage; +import org.apache.qpid.server.queue.AMQQueue; + +import uk.co.thebadgerset.junit.extensions.util.SizeOf; + +/** + * + * This is a special diagnostic exchange type which doesn't actually do anything + * with messages. When it receives a message, it writes information about the + * current memory usage to the "memory" property of the message and places it on the + * diagnosticqueue for retrieval + * + * @author Aidan Skinner + * + */ + +public class DiagnosticExchange extends AbstractExchange +{ + + public static final AMQShortString DIAGNOSTIC_EXCHANGE_CLASS = new AMQShortString("x-diagnostic"); + public static final AMQShortString DIAGNOSTIC_EXCHANGE_NAME = new AMQShortString("diagnostic"); + + /** + * the logger. + */ + //private static final Logger _logger = Logger.getLogger(DiagnosticExchange.class); + + /** + * MBean class implementing the management interfaces. + */ + @MBeanDescription("Management Bean for Diagnostic Exchange") + private final class DiagnosticExchangeMBean extends ExchangeMBean + { + + /** + * Usual constructor. + * + * @throws JMException + */ + @MBeanConstructor("Creates an MBean for AMQ Diagnostic exchange") + public DiagnosticExchangeMBean() throws JMException + { + super(); + _exchangeType = "diagnostic"; + init(); + } + + /** + * Returns nothing, there can be no tabular data for this... + * + * @throws OpenDataException + * @returns null + * @todo ... or can there? Could this actually return all the + * information in one easy to read table? + */ + public TabularData bindings() throws OpenDataException + { + return null; + } + + /** + * This exchange type doesn't support queues, so this method does + * nothing. + * + * @param queueName + * the queue you'll fail to create + * @param binding + * the binding you'll fail to create + * @throws JMException + * an exception that will never be thrown + */ + public void createNewBinding(String queueName, String binding) throws JMException + { + // No Op + } + + } // End of MBean class + + /** + * Creates a new MBean instance + * + * @return the newly created MBean + * @throws AMQException + * if something goes wrong + */ + protected ExchangeMBean createMBean() throws AMQException + { + try + { + return new DiagnosticExchange.DiagnosticExchangeMBean(); + } + catch (JMException ex) + { + // _logger.error("Exception occured in creating the direct exchange mbean", ex); + throw new AMQException(null, "Exception occured in creating the direct exchange mbean", ex); + } + } + + public AMQShortString getType() + { + return DIAGNOSTIC_EXCHANGE_CLASS; + } + + /** + * Does nothing. + * + * @param routingKey + * pointless + * @param queue + * pointless + * @param args + * pointless + * @throws AMQException + * never + */ + public void registerQueue(AMQShortString routingKey, AMQQueue queue, FieldTable args) throws AMQException + { + // No op + } + + /** + * Does nothing. + * + * @param routingKey + * pointless + * @param queue + * pointless + * @param args + * pointless + * @throws AMQException + * never + */ + public void deregisterQueue(AMQShortString routingKey, AMQQueue queue, FieldTable args) throws AMQException + { + // No op + } + + public boolean isBound(AMQShortString routingKey, AMQQueue queue) + { + return false; + } + + public boolean isBound(AMQShortString routingKey) + { + return false; + } + + public boolean isBound(AMQQueue queue) + { + return false; + } + + public boolean hasBindings() + { + return false; + } + + public void route(AMQMessage payload) throws AMQException + { + + Long value = new Long(SizeOf.getUsedMemory()); + AMQShortString key = new AMQShortString("memory"); + + FieldTable headers = ((BasicContentHeaderProperties)payload.getContentHeaderBody().properties).getHeaders(); + headers.put(key, value); + ((BasicContentHeaderProperties)payload.getContentHeaderBody().properties).setHeaders(headers); + AMQQueue q = getQueueRegistry().getQueue(new AMQShortString("diagnosticqueue")); + + payload.enqueue(q); + + } + + @Override + public Map> getBindings() { + // TODO Auto-generated method stub + return null; + } + + public boolean isBound(AMQShortString routingKey, FieldTable arguments, + AMQQueue queue) { + // TODO Auto-generated method stub + return false; + } +} diff --git a/java/plugins/src/main/java/org/apache/qpid/extras/exchanges/diagnostic/DiagnosticExchangeType.java b/java/plugins/src/main/java/org/apache/qpid/extras/exchanges/diagnostic/DiagnosticExchangeType.java new file mode 100644 index 0000000000..d96b4dc99e --- /dev/null +++ b/java/plugins/src/main/java/org/apache/qpid/extras/exchanges/diagnostic/DiagnosticExchangeType.java @@ -0,0 +1,57 @@ +/* + * + * 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 org.apache.qpid.extras.exchanges.diagnostic; + +import org.apache.qpid.AMQException; +import org.apache.qpid.framing.AMQShortString; +import org.apache.qpid.server.exchange.ExchangeType; +import org.apache.qpid.server.virtualhost.VirtualHost; + +/** + * Exchange type class for getting hold of the exchange. + */ +public final class DiagnosticExchangeType implements ExchangeType +{ + + public AMQShortString getName() + { + return DiagnosticExchange.DIAGNOSTIC_EXCHANGE_CLASS; + } + + public Class getExchangeClass() + { + return DiagnosticExchange.class; + } + + public DiagnosticExchange newInstance(VirtualHost host, AMQShortString name, boolean durable, int ticket, boolean autoDelete) + throws AMQException + { + DiagnosticExchange exch = new DiagnosticExchange(); + exch.initialise(host, name, durable, ticket, autoDelete); + return exch; + } + + public AMQShortString getDefaultExchangeName() + { + return DiagnosticExchange.DIAGNOSTIC_EXCHANGE_NAME; + } +} diff --git a/java/plugins/src/main/java/org/apache/qpid/extras/exchanges/example/TestExchange.java b/java/plugins/src/main/java/org/apache/qpid/extras/exchanges/example/TestExchange.java new file mode 100644 index 0000000000..f9904a76b9 --- /dev/null +++ b/java/plugins/src/main/java/org/apache/qpid/extras/exchanges/example/TestExchange.java @@ -0,0 +1,97 @@ +package org.apache.qpid.extras.exchanges.example; + +import java.util.List; +import java.util.Map; + +import org.apache.qpid.AMQException; +import org.apache.qpid.framing.AMQShortString; +import org.apache.qpid.framing.FieldTable; +import org.apache.qpid.server.exchange.Exchange; +import org.apache.qpid.server.queue.AMQMessage; +import org.apache.qpid.server.queue.AMQQueue; +import org.apache.qpid.server.virtualhost.VirtualHost; + +public class TestExchange implements Exchange +{ + + public void close() throws AMQException + { + } + + public void deregisterQueue(AMQShortString routingKey, AMQQueue queue, FieldTable args) throws AMQException + { + } + + public Map> getBindings() + { + return null; + } + + public AMQShortString getName() + { + return null; + } + + public AMQShortString getType() + { + return null; + } + + public boolean hasBindings() + { + return false; + } + + public void initialise(VirtualHost host, AMQShortString name, boolean durable, boolean autoDelete) + throws AMQException + { + } + + public boolean isAutoDelete() + { + return false; + } + + public boolean isBound(AMQShortString routingKey, FieldTable arguments, AMQQueue queue) + { + return false; + } + + public boolean isBound(AMQShortString routingKey, AMQQueue queue) + { + return false; + } + + public boolean isBound(AMQShortString routingKey) + { + return false; + } + + public boolean isBound(AMQQueue queue) + { + return false; + } + + public boolean isDurable() + { + return false; + } + + public void registerQueue(AMQShortString routingKey, AMQQueue queue, FieldTable args) throws AMQException + { + } + + public void route(AMQMessage message) throws AMQException + { + } + + public int getTicket() + { + return 0; + } + + public void initialise(VirtualHost arg0, AMQShortString arg1, boolean arg2, int arg3, boolean arg4) + throws AMQException + { + } +} diff --git a/java/plugins/src/main/java/org/apache/qpid/extras/exchanges/example/TestExchangeType.java b/java/plugins/src/main/java/org/apache/qpid/extras/exchanges/example/TestExchangeType.java new file mode 100644 index 0000000000..96ef8b3470 --- /dev/null +++ b/java/plugins/src/main/java/org/apache/qpid/extras/exchanges/example/TestExchangeType.java @@ -0,0 +1,36 @@ +package org.apache.qpid.extras.exchanges.example; + +import org.apache.qpid.AMQException; +import org.apache.qpid.framing.AMQShortString; +import org.apache.qpid.server.exchange.Exchange; +import org.apache.qpid.server.exchange.ExchangeType; +import org.apache.qpid.server.virtualhost.VirtualHost; + +public class TestExchangeType implements ExchangeType +{ + + public Class getExchangeClass() + { + return TestExchange.class; + } + + public AMQShortString getName() + { + return null; + } + + public Exchange newInstance(VirtualHost host, AMQShortString name, boolean durable, + int token, boolean autoDelete) + throws AMQException + { + TestExchange ex = new TestExchange(); + ex.initialise(host, name, durable, token, autoDelete); + return ex; + } + + public AMQShortString getDefaultExchangeName() + { + return new AMQShortString("test.exchange"); + } + +} diff --git a/java/pom.xml b/java/pom.xml index b8bc0e3578..a3b3843064 100644 --- a/java/pom.xml +++ b/java/pom.xml @@ -150,6 +150,7 @@ under the License. common broker client + plugins systests perftests integrationtests diff --git a/java/systests/pom.xml b/java/systests/pom.xml index d9d3e684ca..8173396dae 100644 --- a/java/systests/pom.xml +++ b/java/systests/pom.xml @@ -82,6 +82,14 @@ **/*Test.class + + + + example.plugin.target + ${basedir}/${topDirectoryLocation}/plugins/target + + + **/testcases/ImmediateMessageTest.class **/testcases/MandatoryMessageTest.class @@ -107,8 +115,8 @@ - - + + @@ -161,3 +169,4 @@ + diff --git a/java/systests/src/main/java/org/apache/qpid/server/plugins/PluginTest.java b/java/systests/src/main/java/org/apache/qpid/server/plugins/PluginTest.java new file mode 100644 index 0000000000..1b082beee4 --- /dev/null +++ b/java/systests/src/main/java/org/apache/qpid/server/plugins/PluginTest.java @@ -0,0 +1,35 @@ +package org.apache.qpid.server.plugins; + +import java.util.Collection; +import java.util.Map; + +import org.apache.qpid.server.exchange.Exchange; +import org.apache.qpid.server.exchange.ExchangeType; + +import junit.framework.TestCase; + +public class PluginTest extends TestCase +{ + + private static final String TEST_EXCHANGE_CLASS = "org.apache.qpid.extras.exchanges.example.TestExchangeType"; + private static final String PLUGIN_DIRECTORY = System.getProperty("example.plugin.target"); + + public void testLoadExchanges() throws Exception + { + PluginManager manager = new PluginManager(PLUGIN_DIRECTORY); + Map> exchanges = manager.getExchanges(); + assertNotNull("No exchanges found in "+PLUGIN_DIRECTORY, exchanges); + assertEquals("Wrong number of exchanges found in "+PLUGIN_DIRECTORY, + 2, exchanges.size()); + assertNotNull("Wrong exchange found in "+PLUGIN_DIRECTORY, + exchanges.get(TEST_EXCHANGE_CLASS)); + } + + public void testNoExchanges() throws Exception + { + PluginManager manager = new PluginManager("/path/to/nowhere"); + Map> exchanges = manager.getExchanges(); + assertNull("Exchanges found", exchanges); + } + +} diff --git a/java/systests/src/main/java/org/apache/qpid/server/util/TestApplicationRegistry.java b/java/systests/src/main/java/org/apache/qpid/server/util/TestApplicationRegistry.java index bd7ed60d1d..0218109369 100644 --- a/java/systests/src/main/java/org/apache/qpid/server/util/TestApplicationRegistry.java +++ b/java/systests/src/main/java/org/apache/qpid/server/util/TestApplicationRegistry.java @@ -23,6 +23,7 @@ package org.apache.qpid.server.util; import org.apache.qpid.server.exchange.ExchangeFactory; import org.apache.qpid.server.exchange.ExchangeRegistry; import org.apache.qpid.server.management.ManagedObjectRegistry; +import org.apache.qpid.server.plugins.PluginManager; import org.apache.qpid.server.queue.QueueRegistry; import org.apache.qpid.server.registry.ApplicationRegistry; import org.apache.qpid.server.registry.IApplicationRegistry; @@ -145,5 +146,11 @@ public class TestApplicationRegistry extends ApplicationRegistry { return _messageStore; } + + public PluginManager getPluginManager() + { + return null; + } } + -- cgit v1.2.1