You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by rg...@apache.org on 2012/03/10 20:22:37 UTC
svn commit: r1299257 [4/26] - in
/qpid/branches/rg-amqp-1-0-sandbox/qpid/java: broker-plugins/
broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/config/
broker-plugins/access-control/src/main/java/org/apache/qpid/server...
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/VirtualHostConfiguration.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/VirtualHostConfiguration.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/VirtualHostConfiguration.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/VirtualHostConfiguration.java Sat Mar 10 19:22:10 2012
@@ -20,17 +20,11 @@
*/
package org.apache.qpid.server.configuration;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
-
import org.apache.commons.configuration.CompositeConfiguration;
import org.apache.commons.configuration.Configuration;
import org.apache.commons.configuration.ConfigurationException;
import org.apache.commons.configuration.PropertiesConfiguration;
+
import org.apache.qpid.exchange.ExchangeDefaults;
import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.server.binding.Binding;
@@ -39,11 +33,18 @@ import org.apache.qpid.server.queue.AMQQ
import org.apache.qpid.server.registry.ApplicationRegistry;
import org.apache.qpid.server.store.MemoryMessageStore;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+
public class VirtualHostConfiguration extends ConfigurationPlugin
{
- private String _name;
- private Map<String, QueueConfiguration> _queues = new HashMap<String, QueueConfiguration>();
- private Map<String, ExchangeConfiguration> _exchanges = new HashMap<String, ExchangeConfiguration>();
+ private final String _name;
+ private final Map<String, QueueConfiguration> _queues = new HashMap<String, QueueConfiguration>();
+ private final Map<String, ExchangeConfiguration> _exchanges = new HashMap<String, ExchangeConfiguration>();
public VirtualHostConfiguration(String name, Configuration config) throws ConfigurationException
{
@@ -75,7 +76,7 @@ public class VirtualHostConfiguration ex
{
CompositeConfiguration mungedConf = new CompositeConfiguration();
mungedConf.addConfiguration(config.subset("exchanges.exchange(" + count++ + ")"));
- mungedConf.addConfiguration(_configuration.subset("exchanges"));
+ mungedConf.addConfiguration(getConfig().subset("exchanges"));
String exchName = (String) i.next();
_exchanges.put(exchName, new ExchangeConfiguration(exchName, mungedConf));
}
@@ -91,11 +92,6 @@ public class VirtualHostConfiguration ex
return getLongValue("housekeeping.checkPeriod", ApplicationRegistry.getInstance().getConfiguration().getHousekeepingCheckPeriod());
}
- public String getAuthenticationDatabase()
- {
- return getStringValue("security.authentication.name");
- }
-
public List getCustomExchanges()
{
return getListValue("custom-exchanges.class-name");
@@ -103,7 +99,7 @@ public class VirtualHostConfiguration ex
public Configuration getStoreConfiguration()
{
- return _configuration.subset("store");
+ return getConfig().subset("store");
}
public String getMessageStoreClass()
@@ -113,7 +109,7 @@ public class VirtualHostConfiguration ex
public void setMessageStoreClass(String storeClass)
{
- _configuration.setProperty("store.class", storeClass);
+ getConfig().setProperty("store.class", storeClass);
}
public List getExchanges()
@@ -252,16 +248,6 @@ public class VirtualHostConfiguration ex
return queueConfig;
}
- public long getMemoryUsageMaximum()
- {
- return getLongValue("queues.maximumMemoryUsage");
- }
-
- public long getMemoryUsageMinimum()
- {
- return getLongValue("queues.minimumMemoryUsage");
- }
-
public int getMaximumMessageAge()
{
return getIntValue("queues.maximumMessageAge");
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/management/ConfigurationManagementMBean.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/management/ConfigurationManagementMBean.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/management/ConfigurationManagementMBean.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/management/ConfigurationManagementMBean.java Sat Mar 10 19:22:10 2012
@@ -20,13 +20,12 @@
*/
package org.apache.qpid.server.configuration.management;
-import javax.management.NotCompliantMBeanException;
-
-import org.apache.commons.configuration.ConfigurationException;
import org.apache.qpid.management.common.mbeans.ConfigurationManagement;
import org.apache.qpid.server.management.AMQManagedObject;
import org.apache.qpid.server.registry.ApplicationRegistry;
+import javax.management.NotCompliantMBeanException;
+
public class ConfigurationManagementMBean extends AMQManagedObject implements ConfigurationManagement
{
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/plugins/ConfigurationPlugin.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/plugins/ConfigurationPlugin.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/plugins/ConfigurationPlugin.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/plugins/ConfigurationPlugin.java Sat Mar 10 19:22:10 2012
@@ -22,6 +22,7 @@ import org.apache.commons.configuration.
import org.apache.commons.configuration.ConfigurationException;
import org.apache.commons.configuration.ConversionException;
import org.apache.log4j.Logger;
+
import org.apache.qpid.server.configuration.ConfigurationManager;
import org.apache.qpid.server.registry.ApplicationRegistry;
import org.apache.qpid.server.registry.IApplicationRegistry;
@@ -42,7 +43,7 @@ public abstract class ConfigurationPlugi
private Map<String, ConfigurationPlugin>
_pluginConfiguration = new HashMap<String, ConfigurationPlugin>();
- protected Configuration _configuration;
+ private Configuration _config;
/**
* The Elements that this Plugin can process.
@@ -65,7 +66,7 @@ public abstract class ConfigurationPlugi
public Configuration getConfig()
{
- return _configuration;
+ return _config;
}
public <C extends ConfigurationPlugin> C getConfiguration(String plugin)
@@ -81,7 +82,7 @@ public abstract class ConfigurationPlugi
*/
public void setConfiguration(String path, Configuration configuration) throws ConfigurationException
{
- _configuration = configuration;
+ _config = configuration;
// Extract a list of elements for processing
Iterator<?> keys = configuration.getKeys();
@@ -215,7 +216,7 @@ public abstract class ConfigurationPlugi
protected boolean hasConfiguration()
{
- return _configuration != null;
+ return _config != null;
}
/// Getters
@@ -227,7 +228,7 @@ public abstract class ConfigurationPlugi
protected double getDoubleValue(String property, double defaultValue)
{
- return _configuration.getDouble(property, defaultValue);
+ return _config.getDouble(property, defaultValue);
}
protected long getLongValue(String property)
@@ -237,7 +238,7 @@ public abstract class ConfigurationPlugi
protected long getLongValue(String property, long defaultValue)
{
- return _configuration.getLong(property, defaultValue);
+ return _config.getLong(property, defaultValue);
}
protected int getIntValue(String property)
@@ -247,7 +248,7 @@ public abstract class ConfigurationPlugi
protected int getIntValue(String property, int defaultValue)
{
- return _configuration.getInt(property, defaultValue);
+ return _config.getInt(property, defaultValue);
}
protected String getStringValue(String property)
@@ -257,7 +258,7 @@ public abstract class ConfigurationPlugi
protected String getStringValue(String property, String defaultValue)
{
- return _configuration.getString(property, defaultValue);
+ return _config.getString(property, defaultValue);
}
protected boolean getBooleanValue(String property)
@@ -267,7 +268,7 @@ public abstract class ConfigurationPlugi
protected boolean getBooleanValue(String property, boolean defaultValue)
{
- return _configuration.getBoolean(property, defaultValue);
+ return _config.getBoolean(property, defaultValue);
}
protected List getListValue(String property)
@@ -277,14 +278,14 @@ public abstract class ConfigurationPlugi
protected List getListValue(String property, List defaultValue)
{
- return _configuration.getList(property, defaultValue);
+ return _config.getList(property, defaultValue);
}
/// Validation Helpers
protected boolean contains(String property)
{
- return _configuration.getProperty(property) != null;
+ return _config.getProperty(property) != null;
}
/**
@@ -323,7 +324,7 @@ public abstract class ConfigurationPlugi
throw new ConfigurationException(this.getClass().getSimpleName() +
": unable to configure invalid " +
property + ":" +
- _configuration.getString(property),
+ _config.getString(property),
last);
}
}
@@ -332,7 +333,7 @@ public abstract class ConfigurationPlugi
{
try
{
- _configuration.getLong(property);
+ _config.getLong(property);
return true;
}
catch (NoSuchElementException e)
@@ -345,7 +346,7 @@ public abstract class ConfigurationPlugi
{
try
{
- long value = _configuration.getLong(property);
+ long value = _config.getLong(property);
return value > 0;
}
catch (NoSuchElementException e)
@@ -359,7 +360,7 @@ public abstract class ConfigurationPlugi
{
try
{
- _configuration.getInt(property);
+ _config.getInt(property);
return true;
}
catch (NoSuchElementException e)
@@ -372,7 +373,7 @@ public abstract class ConfigurationPlugi
{
try
{
- _configuration.getBoolean(property);
+ _config.getBoolean(property);
return true;
}
catch (NoSuchElementException e)
@@ -453,7 +454,7 @@ public abstract class ConfigurationPlugi
protected void mergeConfiguration(ConfigurationPlugin configuration)
{
- _configuration = configuration.getConfig();
+ _config = configuration.getConfig();
}
public String toString()
@@ -478,6 +479,10 @@ public abstract class ConfigurationPlugi
return super.toString();
}
+ protected void setConfig(Configuration config)
+ {
+ _config = config;
+ }
}
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/plugins/ConfigurationPluginFactory.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/plugins/ConfigurationPluginFactory.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/plugins/ConfigurationPluginFactory.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/plugins/ConfigurationPluginFactory.java Sat Mar 10 19:22:10 2012
@@ -20,11 +20,11 @@
*/
package org.apache.qpid.server.configuration.plugins;
-import java.util.List;
-
import org.apache.commons.configuration.Configuration;
import org.apache.commons.configuration.ConfigurationException;
+import java.util.List;
+
public interface ConfigurationPluginFactory
{
/**
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/plugins/SlowConsumerDetectionConfiguration.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/plugins/SlowConsumerDetectionConfiguration.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/plugins/SlowConsumerDetectionConfiguration.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/plugins/SlowConsumerDetectionConfiguration.java Sat Mar 10 19:22:10 2012
@@ -22,9 +22,6 @@ package org.apache.qpid.server.configura
import org.apache.commons.configuration.Configuration;
import org.apache.commons.configuration.ConfigurationException;
-import org.apache.commons.configuration.ConversionException;
-import org.apache.qpid.server.configuration.plugins.ConfigurationPlugin;
-import org.apache.qpid.server.configuration.plugins.ConfigurationPluginFactory;
import java.util.Arrays;
import java.util.List;
@@ -48,7 +45,7 @@ public class SlowConsumerDetectionConfig
}
//Set Default time unit to seconds
- TimeUnit _timeUnit = TimeUnit.SECONDS;
+ private TimeUnit _timeUnit = TimeUnit.SECONDS;
public String[] getElementsProcessed()
{
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/plugins/SlowConsumerDetectionPolicyConfiguration.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/plugins/SlowConsumerDetectionPolicyConfiguration.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/plugins/SlowConsumerDetectionPolicyConfiguration.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/plugins/SlowConsumerDetectionPolicyConfiguration.java Sat Mar 10 19:22:10 2012
@@ -22,8 +22,6 @@ package org.apache.qpid.server.configura
import org.apache.commons.configuration.Configuration;
import org.apache.commons.configuration.ConfigurationException;
-import org.apache.qpid.server.configuration.plugins.ConfigurationPlugin;
-import org.apache.qpid.server.configuration.plugins.ConfigurationPluginFactory;
import java.util.Arrays;
import java.util.List;
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/plugins/SlowConsumerDetectionQueueConfiguration.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/plugins/SlowConsumerDetectionQueueConfiguration.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/plugins/SlowConsumerDetectionQueueConfiguration.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/plugins/SlowConsumerDetectionQueueConfiguration.java Sat Mar 10 19:22:10 2012
@@ -22,8 +22,7 @@ package org.apache.qpid.server.configura
import org.apache.commons.configuration.Configuration;
import org.apache.commons.configuration.ConfigurationException;
-import org.apache.qpid.server.configuration.plugins.ConfigurationPlugin;
-import org.apache.qpid.server.configuration.plugins.ConfigurationPluginFactory;
+
import org.apache.qpid.server.plugins.PluginManager;
import org.apache.qpid.server.registry.ApplicationRegistry;
import org.apache.qpid.slowconsumerdetection.policies.SlowConsumerPolicyPlugin;
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/connection/ConnectionRegistry.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/connection/ConnectionRegistry.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/connection/ConnectionRegistry.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/connection/ConnectionRegistry.java Sat Mar 10 19:22:10 2012
@@ -20,18 +20,18 @@
*/
package org.apache.qpid.server.connection;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.concurrent.CopyOnWriteArrayList;
-
import org.apache.log4j.Logger;
+
import org.apache.qpid.AMQException;
import org.apache.qpid.common.Closeable;
import org.apache.qpid.protocol.AMQConstant;
import org.apache.qpid.server.protocol.AMQConnectionModel;
-import org.apache.qpid.server.protocol.AMQProtocolEngine;
import org.apache.qpid.transport.TransportException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+
public class ConnectionRegistry implements IConnectionRegistry, Closeable
{
private List<AMQConnectionModel> _registry = new CopyOnWriteArrayList<AMQConnectionModel>();
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/connection/IConnectionRegistry.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/connection/IConnectionRegistry.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/connection/IConnectionRegistry.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/connection/IConnectionRegistry.java Sat Mar 10 19:22:10 2012
@@ -20,12 +20,12 @@
*/
package org.apache.qpid.server.connection;
-import java.util.List;
-
import org.apache.qpid.AMQException;
import org.apache.qpid.protocol.AMQConstant;
import org.apache.qpid.server.protocol.AMQConnectionModel;
+import java.util.List;
+
public interface IConnectionRegistry
{
public void initialise();
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/AbstractExchange.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/AbstractExchange.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/AbstractExchange.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/AbstractExchange.java Sat Mar 10 19:22:10 2012
@@ -41,7 +41,6 @@ import org.apache.qpid.server.queue.Queu
import org.apache.qpid.server.virtualhost.VirtualHost;
import javax.management.JMException;
-import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
@@ -62,20 +61,20 @@ public abstract class AbstractExchange i
private Exchange _alternateExchange;
- protected boolean _durable;
- protected int _ticket;
+ private boolean _durable;
+ private int _ticket;
private VirtualHost _virtualHost;
private final List<Exchange.Task> _closeTaskList = new CopyOnWriteArrayList<Exchange.Task>();
- protected AbstractExchangeMBean _exchangeMbean;
+ private AbstractExchangeMBean _exchangeMbean;
/**
* Whether the exchange is automatically deleted once all queues have detached from it
*/
- protected boolean _autoDelete;
+ private boolean _autoDelete;
//The logSubject for ths exchange
private LogSubject _logSubject;
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/AbstractExchangeMBean.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/AbstractExchangeMBean.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/AbstractExchangeMBean.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/AbstractExchangeMBean.java Sat Mar 10 19:22:10 2012
@@ -20,29 +20,31 @@
*/
package org.apache.qpid.server.exchange;
-import java.util.Collections;
-import java.util.Map;
-
import org.apache.qpid.AMQException;
-import org.apache.qpid.AMQSecurityException;
+import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.management.common.mbeans.ManagedExchange;
+import org.apache.qpid.server.logging.actors.CurrentActor;
+import org.apache.qpid.server.logging.actors.ManagementActor;
import org.apache.qpid.server.management.AMQManagedObject;
import org.apache.qpid.server.management.ManagedObject;
import org.apache.qpid.server.management.ManagedObjectRegistry;
+import org.apache.qpid.server.queue.AMQQueue;
import org.apache.qpid.server.registry.ApplicationRegistry;
import org.apache.qpid.server.virtualhost.VirtualHost;
-import org.apache.qpid.server.queue.AMQQueue;
-import org.apache.qpid.server.binding.BindingFactory;
-import org.apache.qpid.server.logging.actors.CurrentActor;
-import org.apache.qpid.server.logging.actors.ManagementActor;
-import org.apache.qpid.management.common.mbeans.ManagedExchange;
-import org.apache.qpid.framing.AMQShortString;
-import javax.management.openmbean.*;
+import javax.management.JMException;
import javax.management.MBeanException;
+import javax.management.MalformedObjectNameException;
import javax.management.NotCompliantMBeanException;
import javax.management.ObjectName;
-import javax.management.MalformedObjectNameException;
-import javax.management.JMException;
+import javax.management.openmbean.ArrayType;
+import javax.management.openmbean.CompositeType;
+import javax.management.openmbean.OpenDataException;
+import javax.management.openmbean.OpenType;
+import javax.management.openmbean.SimpleType;
+import javax.management.openmbean.TabularType;
+import java.util.Collections;
+
/**
* Abstract MBean class. This has some of the methods implemented from
@@ -52,9 +54,9 @@ import javax.management.JMException;
public abstract class AbstractExchangeMBean<T extends AbstractExchange> extends AMQManagedObject implements ManagedExchange
{
// open mbean data types for representing exchange bindings
- protected OpenType[] _bindingItemTypes;
- protected CompositeType _bindingDataType;
- protected TabularType _bindinglistDataType;
+ private OpenType[] _bindingItemTypes;
+ private CompositeType _bindingDataType;
+ private TabularType _bindinglistDataType;
private T _exchange;
@@ -105,17 +107,17 @@ public abstract class AbstractExchangeMB
public Integer getTicketNo()
{
- return _exchange._ticket;
+ return _exchange.getTicket();
}
public boolean isDurable()
{
- return _exchange._durable;
+ return _exchange.isDurable();
}
public boolean isAutoDelete()
{
- return _exchange._autoDelete;
+ return _exchange.isAutoDelete();
}
// Added exchangetype in the object name lets maangement apps to do any customization required
@@ -140,7 +142,7 @@ public abstract class AbstractExchangeMB
throw new JMException("Queue \"" + queueName + "\" is not registered with the virtualhost.");
}
- CurrentActor.set(new ManagementActor(_logActor.getRootMessageLogger()));
+ CurrentActor.set(new ManagementActor(getLogActor().getRootMessageLogger()));
try
{
vhost.getBindingFactory().addBinding(binding,queue,getExchange(),null);
@@ -156,7 +158,7 @@ public abstract class AbstractExchangeMB
/**
* Removes a queue binding from the exchange.
*
- * @see BindingFactory#removeBinding(String, AMQQueue, Exchange, Map)
+ * @see org.apache.qpid.server.binding.BindingFactory#removeBinding(String, AMQQueue, Exchange, java.util.Map)
*/
public void removeBinding(String queueName, String binding) throws JMException
{
@@ -167,7 +169,7 @@ public abstract class AbstractExchangeMB
throw new JMException("Queue \"" + queueName + "\" is not registered with the virtualhost.");
}
- CurrentActor.set(new ManagementActor(_logActor.getRootMessageLogger()));
+ CurrentActor.set(new ManagementActor(getLogActor().getRootMessageLogger()));
try
{
vhost.getBindingFactory().removeBinding(binding, queue, _exchange, Collections.<String, Object>emptyMap());
@@ -179,4 +181,35 @@ public abstract class AbstractExchangeMB
}
CurrentActor.remove();
}
+
+
+ protected OpenType[] getBindingItemTypes()
+ {
+ return _bindingItemTypes;
+ }
+
+ protected void setBindingItemTypes(OpenType[] bindingItemTypes)
+ {
+ _bindingItemTypes = bindingItemTypes;
+ }
+
+ protected CompositeType getBindingDataType()
+ {
+ return _bindingDataType;
+ }
+
+ protected void setBindingDataType(CompositeType bindingDataType)
+ {
+ _bindingDataType = bindingDataType;
+ }
+
+ protected TabularType getBindinglistDataType()
+ {
+ return _bindinglistDataType;
+ }
+
+ protected void setBindinglistDataType(TabularType bindinglistDataType)
+ {
+ _bindinglistDataType = bindinglistDataType;
+ }
}
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/DefaultExchangeFactory.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/DefaultExchangeFactory.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/DefaultExchangeFactory.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/DefaultExchangeFactory.java Sat Mar 10 19:22:10 2012
@@ -20,23 +20,22 @@
*/
package org.apache.qpid.server.exchange;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.Map;
-
import org.apache.log4j.Logger;
-import org.apache.qpid.AMQConnectionException;
+
import org.apache.qpid.AMQException;
import org.apache.qpid.AMQSecurityException;
import org.apache.qpid.AMQUnknownExchangeType;
import org.apache.qpid.framing.AMQShortString;
-import org.apache.qpid.protocol.AMQConstant;
import org.apache.qpid.qmf.ManagementExchange;
import org.apache.qpid.server.configuration.VirtualHostConfiguration;
import org.apache.qpid.server.registry.ApplicationRegistry;
import org.apache.qpid.server.virtualhost.VirtualHost;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+
public class DefaultExchangeFactory implements ExchangeFactory
{
private static final Logger _logger = Logger.getLogger(DefaultExchangeFactory.class);
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/DefaultExchangeRegistry.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/DefaultExchangeRegistry.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/DefaultExchangeRegistry.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/DefaultExchangeRegistry.java Sat Mar 10 19:22:10 2012
@@ -20,20 +20,20 @@
*/
package org.apache.qpid.server.exchange;
-import java.util.Collection;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
-
import org.apache.log4j.Logger;
+
import org.apache.qpid.AMQException;
import org.apache.qpid.AMQSecurityException;
import org.apache.qpid.exchange.ExchangeDefaults;
import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.protocol.AMQConstant;
-import org.apache.qpid.server.queue.IncomingMessage;
import org.apache.qpid.server.store.DurableConfigurationStore;
import org.apache.qpid.server.virtualhost.VirtualHost;
+import java.util.Collection;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
public class DefaultExchangeRegistry implements ExchangeRegistry
{
private static final Logger _log = Logger.getLogger(DefaultExchangeRegistry.class);
@@ -153,24 +153,4 @@ public class DefaultExchangeRegistry imp
}
}
-
- /**
- * Routes content through exchanges, delivering it to 1 or more queues.
- * @param payload
- * @throws AMQException if something goes wrong delivering data
- */
- public void routeContent(IncomingMessage payload) throws AMQException
- {
- final AMQShortString exchange = payload.getExchange();
- final Exchange exch = getExchange(exchange);
- // there is a small window of opportunity for the exchange to be deleted in between
- // the BasicPublish being received (where the exchange is validated) and the final
- // content body being received (which triggers this method)
- // TODO: check where the exchange is validated
- if (exch == null)
- {
- throw new AMQException("Exchange '" + exchange + "' does not exist");
- }
- payload.enqueue(exch.route(payload));
- }
}
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/DirectExchangeMBean.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/DirectExchangeMBean.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/DirectExchangeMBean.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/DirectExchangeMBean.java Sat Mar 10 19:22:10 2012
@@ -20,16 +20,20 @@
*/
package org.apache.qpid.server.exchange;
-import org.apache.qpid.management.common.mbeans.annotations.MBeanDescription;
import org.apache.qpid.management.common.mbeans.annotations.MBeanConstructor;
+import org.apache.qpid.management.common.mbeans.annotations.MBeanDescription;
import org.apache.qpid.server.binding.Binding;
import javax.management.JMException;
-import javax.management.openmbean.*;
-import java.util.List;
-import java.util.Map;
+import javax.management.openmbean.CompositeData;
+import javax.management.openmbean.CompositeDataSupport;
+import javax.management.openmbean.OpenDataException;
+import javax.management.openmbean.TabularData;
+import javax.management.openmbean.TabularDataSupport;
import java.util.ArrayList;
import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
/**
* MBean class implementing the management interfaces.
@@ -47,7 +51,7 @@ final class DirectExchangeMBean extends
public TabularData bindings() throws OpenDataException
{
- TabularDataSupport bindingList = new TabularDataSupport(_bindinglistDataType);
+ TabularDataSupport bindingList = new TabularDataSupport(getBindinglistDataType());
Map<String, List<String>> bindingMap = new HashMap<String, List<String>>();
@@ -67,7 +71,7 @@ final class DirectExchangeMBean extends
for(Map.Entry<String, List<String>> entry : bindingMap.entrySet())
{
Object[] bindingItemValues = {entry.getKey(), entry.getValue().toArray(new String[0])};
- CompositeData bindingData = new CompositeDataSupport(_bindingDataType,
+ CompositeData bindingData = new CompositeDataSupport(getBindingDataType(),
COMPOSITE_ITEM_NAMES.toArray(new String[COMPOSITE_ITEM_NAMES.size()]),
bindingItemValues);
bindingList.put(bindingData);
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/Exchange.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/Exchange.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/Exchange.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/Exchange.java Sat Mar 10 19:22:10 2012
@@ -25,18 +25,17 @@ import org.apache.qpid.AMQInternalExcept
import org.apache.qpid.AMQSecurityException;
import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.framing.FieldTable;
-
+import org.apache.qpid.server.binding.Binding;
+import org.apache.qpid.server.configuration.ExchangeConfig;
+import org.apache.qpid.server.message.InboundMessage;
import org.apache.qpid.server.queue.AMQQueue;
import org.apache.qpid.server.queue.BaseQueue;
import org.apache.qpid.server.virtualhost.VirtualHost;
-import org.apache.qpid.server.message.InboundMessage;
-import org.apache.qpid.server.binding.Binding;
-import org.apache.qpid.server.configuration.ExchangeConfig;
import javax.management.JMException;
-import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
+import java.util.Map;
public interface Exchange extends ExchangeReferrer, ExchangeConfig
{
@@ -113,6 +112,8 @@ public interface Exchange extends Exchan
boolean isBound(String bindingKey, AMQQueue queue);
+ public boolean isBound(String bindingKey, Map<String,Object> arguments, AMQQueue queue);
+
boolean isBound(String bindingKey);
void addCloseTask(Task task);
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/ExchangeFactory.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/ExchangeFactory.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/ExchangeFactory.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/ExchangeFactory.java Sat Mar 10 19:22:10 2012
@@ -20,12 +20,12 @@
*/
package org.apache.qpid.server.exchange;
-import java.util.Collection;
-
import org.apache.qpid.AMQException;
import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.server.configuration.VirtualHostConfiguration;
+import java.util.Collection;
+
public interface ExchangeFactory
{
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/ExchangeRegistry.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/ExchangeRegistry.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/ExchangeRegistry.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/ExchangeRegistry.java Sat Mar 10 19:22:10 2012
@@ -26,7 +26,7 @@ import org.apache.qpid.framing.AMQShortS
import java.util.Collection;
-public interface ExchangeRegistry extends MessageRouter
+public interface ExchangeRegistry
{
void registerExchange(Exchange exchange) throws AMQException;
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/ExchangeType.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/ExchangeType.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/ExchangeType.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/ExchangeType.java Sat Mar 10 19:22:10 2012
@@ -20,8 +20,8 @@
*/
package org.apache.qpid.server.exchange;
-import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.AMQException;
+import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.server.virtualhost.VirtualHost;
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/FanoutExchangeMBean.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/FanoutExchangeMBean.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/FanoutExchangeMBean.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/FanoutExchangeMBean.java Sat Mar 10 19:22:10 2012
@@ -20,12 +20,16 @@
*/
package org.apache.qpid.server.exchange;
-import org.apache.qpid.management.common.mbeans.annotations.MBeanDescription;
import org.apache.qpid.management.common.mbeans.annotations.MBeanConstructor;
+import org.apache.qpid.management.common.mbeans.annotations.MBeanDescription;
import org.apache.qpid.server.binding.Binding;
import javax.management.JMException;
-import javax.management.openmbean.*;
+import javax.management.openmbean.CompositeData;
+import javax.management.openmbean.CompositeDataSupport;
+import javax.management.openmbean.OpenDataException;
+import javax.management.openmbean.TabularData;
+import javax.management.openmbean.TabularDataSupport;
import java.util.ArrayList;
/**
@@ -46,7 +50,7 @@ final class FanoutExchangeMBean extends
public TabularData bindings() throws OpenDataException
{
- TabularDataSupport bindingList = new TabularDataSupport(_bindinglistDataType);
+ TabularDataSupport bindingList = new TabularDataSupport(getBindinglistDataType());
ArrayList<String> queueNames = new ArrayList<String>();
@@ -58,7 +62,7 @@ final class FanoutExchangeMBean extends
}
Object[] bindingItemValues = {BINDING_KEY_SUBSTITUTE, queueNames.toArray(new String[0])};
- CompositeData bindingData = new CompositeDataSupport(_bindingDataType,
+ CompositeData bindingData = new CompositeDataSupport(getBindingDataType(),
COMPOSITE_ITEM_NAMES.toArray(new String[COMPOSITE_ITEM_NAMES.size()]),
bindingItemValues);
bindingList.put(bindingData);
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/HeadersBinding.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/HeadersBinding.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/HeadersBinding.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/HeadersBinding.java Sat Mar 10 19:22:10 2012
@@ -20,17 +20,18 @@
*/
package org.apache.qpid.server.exchange;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.Set;
-
import org.apache.log4j.Logger;
+
import org.apache.qpid.framing.AMQTypedValue;
import org.apache.qpid.framing.FieldTable;
import org.apache.qpid.server.binding.Binding;
import org.apache.qpid.server.message.AMQMessageHeader;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
/**
* Defines binding and matching based on a set of headers.
*/
@@ -38,7 +39,7 @@ class HeadersBinding
{
private static final Logger _logger = Logger.getLogger(HeadersBinding.class);
- private final FieldTable _mappings;
+ private final Map<String,Object> _mappings;
private final Binding _binding;
private final Set<String> required = new HashSet<String>();
private final Map<String,Object> matches = new HashMap<String,Object>();
@@ -57,7 +58,7 @@ class HeadersBinding
_binding = binding;
if(_binding !=null)
{
- _mappings = FieldTable.convertToFieldTable(_binding.getArguments());
+ _mappings = _binding.getArguments();
initMappings();
}
else
@@ -68,37 +69,23 @@ class HeadersBinding
private void initMappings()
{
- _mappings.processOverElements(new FieldTable.FieldTableElementProcessor()
+ for(Map.Entry<String, Object> entry : _mappings.entrySet())
{
-
- public boolean processElement(String propertyName, AMQTypedValue value)
+ String propertyName = entry.getKey();
+ Object value = entry.getValue();
+ if (isSpecial(propertyName))
{
- if (isSpecial(propertyName))
- {
- processSpecial(propertyName, value.getValue());
- }
- else if (value.getValue() == null || value.getValue().equals(""))
- {
- required.add(propertyName);
- }
- else
- {
- matches.put(propertyName,value.getValue());
- }
-
- return true;
+ processSpecial(propertyName, value);
}
-
- public Object getResult()
+ else if (value == null || value.equals(""))
{
- return null;
+ required.add(propertyName);
}
- });
- }
-
- protected FieldTable getMappings()
- {
- return _mappings;
+ else
+ {
+ matches.put(propertyName,value);
+ }
+ }
}
public Binding getBinding()
@@ -206,8 +193,14 @@ class HeadersBinding
{
if(value instanceof String)
{
- if("any".equalsIgnoreCase((String) value)) return true;
- if("all".equalsIgnoreCase((String) value)) return false;
+ if("any".equalsIgnoreCase((String) value))
+ {
+ return true;
+ }
+ if("all".equalsIgnoreCase((String) value))
+ {
+ return false;
+ }
}
_logger.warn("Ignoring unrecognised match type: " + value);
return false;//default to all
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/HeadersExchange.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/HeadersExchange.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/HeadersExchange.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/HeadersExchange.java Sat Mar 10 19:22:10 2012
@@ -21,23 +21,24 @@
package org.apache.qpid.server.exchange;
import org.apache.log4j.Logger;
+
import org.apache.qpid.AMQException;
import org.apache.qpid.exchange.ExchangeDefaults;
import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.framing.FieldTable;
+import org.apache.qpid.server.binding.Binding;
+import org.apache.qpid.server.message.AMQMessageHeader;
+import org.apache.qpid.server.message.InboundMessage;
import org.apache.qpid.server.queue.AMQQueue;
import org.apache.qpid.server.queue.BaseQueue;
import org.apache.qpid.server.virtualhost.VirtualHost;
-import org.apache.qpid.server.message.InboundMessage;
-import org.apache.qpid.server.message.AMQMessageHeader;
-import org.apache.qpid.server.binding.Binding;
import javax.management.JMException;
import java.util.ArrayList;
import java.util.LinkedHashSet;
import java.util.Map;
-import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CopyOnWriteArraySet;
/**
@@ -145,6 +146,33 @@ public class HeadersExchange extends Abs
return new ArrayList<BaseQueue>(queues);
}
+
+ public boolean isBound(String bindingKey, Map<String, Object> arguments, AMQQueue queue)
+ {
+ CopyOnWriteArraySet<Binding> bindings;
+ if(bindingKey == null)
+ {
+ bindings = new CopyOnWriteArraySet<Binding>(getBindings());
+ }
+ else
+ {
+ bindings = _bindingsByKey.get(bindingKey);
+ }
+
+ if(bindings != null)
+ {
+ for(Binding binding : bindings)
+ {
+ if(queue == null || binding.getQueue().equals(queue))
+ {
+ return arguments == null ? binding.getArguments() == null : binding.getArguments().equals(arguments);
+ }
+ }
+ }
+
+ return false;
+ }
+
public boolean isBound(AMQShortString routingKey, FieldTable arguments, AMQQueue queue)
{
//fixme isBound here should take the arguements in to consideration.
@@ -250,10 +278,11 @@ public class HeadersExchange extends Abs
{
bindings.remove(binding);
}
-
+
+ boolean removedBinding = _bindingHeaderMatchers.remove(new HeadersBinding(binding));
if(_logger.isDebugEnabled())
{
- _logger.debug("Removing Binding: " + _bindingHeaderMatchers.remove(new HeadersBinding(binding)));
+ _logger.debug("Removing Binding: " + removedBinding);
}
}
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/HeadersExchangeMBean.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/HeadersExchangeMBean.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/HeadersExchangeMBean.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/HeadersExchangeMBean.java Sat Mar 10 19:22:10 2012
@@ -22,8 +22,8 @@ package org.apache.qpid.server.exchange;
import org.apache.qpid.AMQException;
import org.apache.qpid.framing.AMQShortString;
-import org.apache.qpid.management.common.mbeans.annotations.MBeanDescription;
import org.apache.qpid.management.common.mbeans.annotations.MBeanConstructor;
+import org.apache.qpid.management.common.mbeans.annotations.MBeanDescription;
import org.apache.qpid.server.binding.Binding;
import org.apache.qpid.server.logging.actors.CurrentActor;
import org.apache.qpid.server.logging.actors.ManagementActor;
@@ -32,11 +32,19 @@ import org.apache.qpid.server.virtualhos
import javax.management.JMException;
import javax.management.MBeanException;
-import javax.management.openmbean.*;
-
+import javax.management.openmbean.ArrayType;
+import javax.management.openmbean.CompositeData;
+import javax.management.openmbean.CompositeDataSupport;
+import javax.management.openmbean.CompositeType;
+import javax.management.openmbean.OpenDataException;
+import javax.management.openmbean.OpenType;
+import javax.management.openmbean.SimpleType;
+import javax.management.openmbean.TabularData;
+import javax.management.openmbean.TabularDataSupport;
+import javax.management.openmbean.TabularType;
+import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
-import java.util.ArrayList;
import java.util.Map;
/**
@@ -60,20 +68,20 @@ final class HeadersExchangeMBean extends
protected void init() throws OpenDataException
{
- _bindingItemTypes = new OpenType[3];
- _bindingItemTypes[0] = SimpleType.INTEGER;
- _bindingItemTypes[1] = SimpleType.STRING;
- _bindingItemTypes[2] = new ArrayType(1, SimpleType.STRING);
- _bindingDataType = new CompositeType("Exchange Binding", "Queue name and header bindings",
- HEADERS_COMPOSITE_ITEM_NAMES.toArray(new String[HEADERS_COMPOSITE_ITEM_NAMES.size()]),
- HEADERS_COMPOSITE_ITEM_DESC.toArray(new String[HEADERS_COMPOSITE_ITEM_DESC.size()]), _bindingItemTypes);
- _bindinglistDataType = new TabularType("Exchange Bindings", "List of exchange bindings for " + getName(),
- _bindingDataType, HEADERS_TABULAR_UNIQUE_INDEX.toArray(new String[HEADERS_TABULAR_UNIQUE_INDEX.size()]));
+ setBindingItemTypes(new OpenType[3]);
+ getBindingItemTypes()[0] = SimpleType.INTEGER;
+ getBindingItemTypes()[1] = SimpleType.STRING;
+ getBindingItemTypes()[2] = new ArrayType(1, SimpleType.STRING);
+ setBindingDataType(new CompositeType("Exchange Binding", "Queue name and header bindings",
+ HEADERS_COMPOSITE_ITEM_NAMES.toArray(new String[HEADERS_COMPOSITE_ITEM_NAMES.size()]),
+ HEADERS_COMPOSITE_ITEM_DESC.toArray(new String[HEADERS_COMPOSITE_ITEM_DESC.size()]), getBindingItemTypes()));
+ setBindinglistDataType(new TabularType("Exchange Bindings", "List of exchange bindings for " + getName(),
+ getBindingDataType(), HEADERS_TABULAR_UNIQUE_INDEX.toArray(new String[HEADERS_TABULAR_UNIQUE_INDEX.size()])));
}
public TabularData bindings() throws OpenDataException
{
- TabularDataSupport bindingList = new TabularDataSupport(_bindinglistDataType);
+ TabularDataSupport bindingList = new TabularDataSupport(getBindinglistDataType());
int count = 1;
for (Binding binding : getExchange().getBindings())
{
@@ -95,7 +103,7 @@ final class HeadersExchangeMBean extends
Object[] bindingItemValues = {count++, queueName, mappingList.toArray(new String[0])};
- CompositeData bindingData = new CompositeDataSupport(_bindingDataType,
+ CompositeData bindingData = new CompositeDataSupport(getBindingDataType(),
HEADERS_COMPOSITE_ITEM_NAMES.toArray(new String[HEADERS_COMPOSITE_ITEM_NAMES.size()]), bindingItemValues);
bindingList.put(bindingData);
}
@@ -113,7 +121,7 @@ final class HeadersExchangeMBean extends
throw new JMException("Queue \"" + queueName + "\" is not registered with the virtualhost.");
}
- CurrentActor.set(new ManagementActor(_logActor.getRootMessageLogger()));
+ CurrentActor.set(new ManagementActor(getLogActor().getRootMessageLogger()));
final Map<String,Object> arguments = new HashMap<String, Object>();
final String[] bindings = binding.split(",");
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/TopicExchange.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/TopicExchange.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/TopicExchange.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/TopicExchange.java Sat Mar 10 19:22:10 2012
@@ -20,29 +20,39 @@
*/
package org.apache.qpid.server.exchange;
+import java.lang.ref.WeakReference;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.WeakHashMap;
+import java.util.concurrent.ConcurrentHashMap;
+import javax.management.JMException;
import org.apache.log4j.Logger;
-
import org.apache.qpid.AMQException;
import org.apache.qpid.AMQInvalidArgumentException;
import org.apache.qpid.common.AMQPFilterTypes;
import org.apache.qpid.exchange.ExchangeDefaults;
+import org.apache.qpid.filter.SelectorParsingException;
+import org.apache.qpid.filter.selector.ParseException;
+import org.apache.qpid.filter.selector.TokenMgrError;
import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.framing.FieldTable;
-import org.apache.qpid.server.queue.AMQQueue;
-import org.apache.qpid.server.queue.BaseQueue;
-import org.apache.qpid.server.virtualhost.VirtualHost;
import org.apache.qpid.server.binding.Binding;
-import org.apache.qpid.server.exchange.topic.*;
+import org.apache.qpid.server.exchange.topic.TopicExchangeResult;
+import org.apache.qpid.server.exchange.topic.TopicMatcherResult;
+import org.apache.qpid.server.exchange.topic.TopicNormalizer;
+import org.apache.qpid.server.exchange.topic.TopicParser;
import org.apache.qpid.server.filter.JMSSelectorFilter;
+import org.apache.qpid.server.filter.MessageFilter;
import org.apache.qpid.server.message.InboundMessage;
-
-import javax.management.JMException;
-import java.sql.Array;
-import java.util.*;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.atomic.AtomicLong;
-
-import java.lang.ref.WeakReference;
+import org.apache.qpid.server.protocol.AMQSessionModel;
+import org.apache.qpid.server.queue.AMQQueue;
+import org.apache.qpid.server.queue.BaseQueue;
+import org.apache.qpid.server.queue.Filterable;
+import org.apache.qpid.server.virtualhost.VirtualHost;
public class TopicExchange extends AbstractExchange
{
@@ -113,24 +123,26 @@ public class TopicExchange extends Abstr
FieldTable oldArgs = _bindings.get(binding);
TopicExchangeResult result = _topicExchangeResults.get(routingKey);
- if(argumentsContainSelector(args))
+ if(argumentsContainFilter(args))
{
- if(argumentsContainSelector(oldArgs))
+ if(argumentsContainFilter(oldArgs))
{
- result.replaceQueueFilter(queue,createSelectorFilter(oldArgs), createSelectorFilter(args));
+ result.replaceQueueFilter(queue,
+ createMessageFilter(oldArgs, queue),
+ createMessageFilter(args, queue));
}
else
{
- result.addFilteredQueue(queue,createSelectorFilter(args));
+ result.addFilteredQueue(queue, createMessageFilter(args, queue));
result.removeUnfilteredQueue(queue);
}
}
else
{
- if(argumentsContainSelector(oldArgs))
+ if(argumentsContainFilter(oldArgs))
{
result.addUnfilteredQueue(queue);
- result.removeFilteredQueue(queue, createSelectorFilter(oldArgs));
+ result.removeFilteredQueue(queue, createMessageFilter(oldArgs, queue));
}
else
{
@@ -149,9 +161,9 @@ public class TopicExchange extends Abstr
if(result == null)
{
result = new TopicExchangeResult();
- if(argumentsContainSelector(args))
+ if(argumentsContainFilter(args))
{
- result.addFilteredQueue(queue, createSelectorFilter(args));
+ result.addFilteredQueue(queue, createMessageFilter(args, queue));
}
else
{
@@ -162,9 +174,9 @@ public class TopicExchange extends Abstr
}
else
{
- if(argumentsContainSelector(args))
+ if(argumentsContainFilter(args))
{
- result.addFilteredQueue(queue, createSelectorFilter(args));
+ result.addFilteredQueue(queue, createMessageFilter(args, queue));
}
else
{
@@ -178,26 +190,74 @@ public class TopicExchange extends Abstr
}
- private JMSSelectorFilter createSelectorFilter(final FieldTable args) throws AMQInvalidArgumentException
+ private MessageFilter createMessageFilter(final FieldTable args, AMQQueue queue) throws AMQInvalidArgumentException
{
+ if(argumentsContainNoLocal(args))
+ {
+ MessageFilter filter = new NoLocalFilter(queue);
+
+ if(argumentsContainJMSSelector(args))
+ {
+ filter = new CompoundFilter(filter, createJMSSelectorFilter(args));
+ }
+ return filter;
+ }
+ else
+ {
+ return createJMSSelectorFilter(args);
+ }
+
+ }
+
+ private MessageFilter createJMSSelectorFilter(FieldTable args) throws AMQInvalidArgumentException
+ {
final String selectorString = args.getString(AMQPFilterTypes.JMS_SELECTOR.getValue());
WeakReference<JMSSelectorFilter> selectorRef = _selectorCache.get(selectorString);
JMSSelectorFilter selector = null;
if(selectorRef == null || (selector = selectorRef.get())==null)
{
- selector = new JMSSelectorFilter(selectorString);
+ try
+ {
+ selector = new JMSSelectorFilter(selectorString);
+ }
+ catch (ParseException e)
+ {
+ throw new AMQInvalidArgumentException("Cannot parse JMS selector \"" + selectorString + "\"", e);
+ }
+ catch (SelectorParsingException e)
+ {
+ throw new AMQInvalidArgumentException("Cannot parse JMS selector \"" + selectorString + "\"", e);
+ }
+ catch (TokenMgrError e)
+ {
+ throw new AMQInvalidArgumentException("Cannot parse JMS selector \"" + selectorString + "\"", e);
+ }
_selectorCache.put(selectorString, new WeakReference<JMSSelectorFilter>(selector));
}
return selector;
}
- private static boolean argumentsContainSelector(final FieldTable args)
+ private static boolean argumentsContainFilter(final FieldTable args)
{
- return args != null && args.containsKey(AMQPFilterTypes.JMS_SELECTOR.getValue()) && args.getString(AMQPFilterTypes.JMS_SELECTOR.getValue()).trim().length() != 0;
+ return argumentsContainNoLocal(args) || argumentsContainJMSSelector(args);
}
+ private static boolean argumentsContainNoLocal(final FieldTable args)
+ {
+ return args != null
+ && args.containsKey(AMQPFilterTypes.NO_LOCAL.getValue())
+ && Boolean.TRUE.equals(args.get(AMQPFilterTypes.NO_LOCAL.getValue()));
+ }
+
+ private static boolean argumentsContainJMSSelector(final FieldTable args)
+ {
+ return args != null && (args.containsKey(AMQPFilterTypes.JMS_SELECTOR.getValue())
+ && args.getString(AMQPFilterTypes.JMS_SELECTOR.getValue()).trim().length() != 0);
+ }
+
+
public ArrayList<BaseQueue> doRoute(InboundMessage payload)
{
@@ -251,6 +311,28 @@ public class TopicExchange extends Abstr
}
}
+ public boolean isBound(String bindingKey, Map<String, Object> arguments, AMQQueue queue)
+ {
+ Binding binding = new Binding(null, bindingKey, queue, this, arguments);
+ if (arguments == null)
+ {
+ return _bindings.containsKey(binding);
+ }
+ else
+ {
+ FieldTable o = _bindings.get(binding);
+ if (o != null)
+ {
+ return arguments.equals(FieldTable.convertToMap(o));
+ }
+ else
+ {
+ return false;
+ }
+ }
+
+ }
+
public boolean isBound(AMQShortString routingKey, AMQQueue queue)
{
return isBound(routingKey, null, queue);
@@ -297,11 +379,11 @@ public class TopicExchange extends Abstr
result.removeBinding(binding);
- if(argumentsContainSelector(bindingArgs))
+ if(argumentsContainFilter(bindingArgs))
{
try
{
- result.removeFilteredQueue(binding.getQueue(), createSelectorFilter(bindingArgs));
+ result.removeFilteredQueue(binding.getQueue(), createMessageFilter(bindingArgs, binding.getQueue()));
}
catch (AMQInvalidArgumentException e)
{
@@ -378,4 +460,96 @@ public class TopicExchange extends Abstr
deregisterQueue(binding);
}
+ private static final class NoLocalFilter implements MessageFilter
+ {
+ private final AMQQueue _queue;
+
+ public NoLocalFilter(AMQQueue queue)
+ {
+ _queue = queue;
+ }
+
+ public boolean matches(Filterable message)
+ {
+ InboundMessage inbound = (InboundMessage) message;
+ final AMQSessionModel exclusiveOwningSession = _queue.getExclusiveOwningSession();
+ return exclusiveOwningSession == null || !exclusiveOwningSession.onSameConnection(inbound);
+
+ }
+
+ @Override
+ public boolean equals(Object o)
+ {
+ if (this == o)
+ {
+ return true;
+ }
+
+ if (o == null || getClass() != o.getClass())
+ {
+ return false;
+ }
+
+ NoLocalFilter that = (NoLocalFilter) o;
+
+ return _queue == null ? that._queue == null : _queue.equals(that._queue);
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return _queue != null ? _queue.hashCode() : 0;
+ }
+ }
+
+ private static final class CompoundFilter implements MessageFilter
+ {
+ private MessageFilter _noLocalFilter;
+ private MessageFilter _jmsSelectorFilter;
+
+ public CompoundFilter(MessageFilter filter, MessageFilter jmsSelectorFilter)
+ {
+ _noLocalFilter = filter;
+ _jmsSelectorFilter = jmsSelectorFilter;
+ }
+
+ public boolean matches(Filterable message)
+ {
+ return _noLocalFilter.matches(message) && _jmsSelectorFilter.matches(message);
+ }
+
+ @Override
+ public boolean equals(Object o)
+ {
+ if (this == o)
+ {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass())
+ {
+ return false;
+ }
+
+ CompoundFilter that = (CompoundFilter) o;
+
+ if (_jmsSelectorFilter != null ? !_jmsSelectorFilter.equals(that._jmsSelectorFilter) : that._jmsSelectorFilter != null)
+ {
+ return false;
+ }
+ if (_noLocalFilter != null ? !_noLocalFilter.equals(that._noLocalFilter) : that._noLocalFilter != null)
+ {
+ return false;
+ }
+
+ return true;
+ }
+
+ @Override
+ public int hashCode()
+ {
+ int result = _noLocalFilter != null ? _noLocalFilter.hashCode() : 0;
+ result = 31 * result + (_jmsSelectorFilter != null ? _jmsSelectorFilter.hashCode() : 0);
+ return result;
+ }
+ }
}
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/TopicExchangeMBean.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/TopicExchangeMBean.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/TopicExchangeMBean.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/TopicExchangeMBean.java Sat Mar 10 19:22:10 2012
@@ -20,16 +20,20 @@
*/
package org.apache.qpid.server.exchange;
-import org.apache.qpid.management.common.mbeans.annotations.MBeanDescription;
import org.apache.qpid.management.common.mbeans.annotations.MBeanConstructor;
+import org.apache.qpid.management.common.mbeans.annotations.MBeanDescription;
import org.apache.qpid.server.binding.Binding;
import javax.management.JMException;
-import javax.management.openmbean.*;
+import javax.management.openmbean.CompositeData;
+import javax.management.openmbean.CompositeDataSupport;
+import javax.management.openmbean.OpenDataException;
+import javax.management.openmbean.TabularData;
+import javax.management.openmbean.TabularDataSupport;
+import java.util.ArrayList;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.HashMap;
-import java.util.ArrayList;
/** TopicExchangeMBean class implements the management interface for the Topic exchanges. */
@MBeanDescription("Management Bean for Topic Exchange")
@@ -47,7 +51,7 @@ final class TopicExchangeMBean extends A
/** returns exchange bindings in tabular form */
public TabularData bindings() throws OpenDataException
{
- TabularDataSupport bindingList = new TabularDataSupport(_bindinglistDataType);
+ TabularDataSupport bindingList = new TabularDataSupport(getBindinglistDataType());
Map<String, List<String>> bindingData = new HashMap<String, List<String>>();
for (Binding binding : getExchange().getBindings())
{
@@ -65,7 +69,7 @@ final class TopicExchangeMBean extends A
{
Object[] bindingItemValues = {entry.getKey(), entry.getValue().toArray(new String[entry.getValue().size()]) };
CompositeData bindingCompositeData =
- new CompositeDataSupport(_bindingDataType,
+ new CompositeDataSupport(getBindingDataType(),
COMPOSITE_ITEM_NAMES.toArray(new String[COMPOSITE_ITEM_NAMES.size()]),
bindingItemValues);
bindingList.put(bindingCompositeData);
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicExchangeResult.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicExchangeResult.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicExchangeResult.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicExchangeResult.java Sat Mar 10 19:22:10 2012
@@ -20,10 +20,10 @@
*/
package org.apache.qpid.server.exchange.topic;
-import org.apache.qpid.server.queue.AMQQueue;
import org.apache.qpid.server.binding.Binding;
import org.apache.qpid.server.filter.MessageFilter;
import org.apache.qpid.server.message.InboundMessage;
+import org.apache.qpid.server.queue.AMQQueue;
import java.util.ArrayList;
import java.util.Collection;
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicMatcherDFAState.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicMatcherDFAState.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicMatcherDFAState.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicMatcherDFAState.java Sat Mar 10 19:22:10 2012
@@ -3,7 +3,17 @@ package org.apache.qpid.server.exchange.
import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.framing.AMQShortStringTokenizer;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.SortedSet;
+import java.util.TreeSet;
import java.util.concurrent.atomic.AtomicInteger;
/*
@@ -246,25 +256,25 @@ public class TopicMatcherDFAState
transitions.append("[ ");
transitions.append(entry.getKey());
transitions.append("\t ->\t ");
- transitions.append(entry.getValue()._id);
+ transitions.append(entry.getValue().getId());
transitions.append(" ]\n");
}
- return "[ State " + _id + " ]\n" + transitions + "\n";
+ return "[ State " + getId() + " ]\n" + transitions + "\n";
}
public String reachableStates()
{
- StringBuilder result = new StringBuilder("Start state: " + _id + "\n");
+ StringBuilder result = new StringBuilder("Start state: " + getId() + "\n");
SortedSet<TopicMatcherDFAState> reachableStates =
new TreeSet<TopicMatcherDFAState>(new Comparator<TopicMatcherDFAState>()
{
public int compare(final TopicMatcherDFAState o1, final TopicMatcherDFAState o2)
{
- return o1._id - o2._id;
+ return o1.getId() - o2.getId();
}
});
reachableStates.add(this);
@@ -292,4 +302,9 @@ public class TopicMatcherDFAState
return result.toString();
}
+
+ int getId()
+ {
+ return _id;
+ }
}
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicNormalizer.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicNormalizer.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicNormalizer.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicNormalizer.java Sat Mar 10 19:22:10 2012
@@ -22,10 +22,9 @@ package org.apache.qpid.server.exchange.
import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.framing.AMQShortStringTokenizer;
-import org.apache.qpid.server.exchange.TopicExchange;
-import java.util.List;
import java.util.ArrayList;
+import java.util.List;
public class TopicNormalizer
{
@@ -37,6 +36,10 @@ public class TopicNormalizer
private static final AMQShortString AMQP_STAR_TOKEN = new AMQShortString("*");
private static final AMQShortString AMQP_HASH_TOKEN = new AMQShortString("#");
+ private TopicNormalizer()
+ {
+ }
+
public static AMQShortString normalize(AMQShortString routingKey)
{
if(routingKey == null)
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicParser.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicParser.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicParser.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicParser.java Sat Mar 10 19:22:10 2012
@@ -3,9 +3,15 @@ package org.apache.qpid.server.exchange.
import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.framing.AMQShortStringTokenizer;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
import java.util.concurrent.atomic.AtomicReference;
-import java.io.IOException;
/*
*
@@ -52,14 +58,43 @@ public class TopicParser
}
+ public TopicWord getWord()
+ {
+ return _word;
+ }
+
+ public boolean isSelfTransition()
+ {
+ return _selfTransition;
+ }
+
+ public int getPosition()
+ {
+ return _position;
+ }
+
+ public boolean isEndState()
+ {
+ return _endState;
+ }
+
+ public boolean isFollowedByAnyLoop()
+ {
+ return _followedByAnyLoop;
+ }
+
+ public void setFollowedByAnyLoop(boolean followedByAnyLoop)
+ {
+ _followedByAnyLoop = followedByAnyLoop;
+ }
}
private static final Position ERROR_POSITION = new Position(Integer.MAX_VALUE,null, true, false);
private static class SimpleState
{
- Set<Position> _positions;
- Map<TopicWord, SimpleState> _nextState;
+ private Set<Position> _positions;
+ private Map<TopicWord, SimpleState> _nextState;
}
@@ -180,11 +215,11 @@ public class TopicParser
while(followedByWildcards && n<(positionCount+1))
{
- if(positions[n]._selfTransition)
+ if(positions[n].isSelfTransition())
{
break;
}
- else if(positions[n]._word!=TopicWord.ANY_WORD)
+ else if(positions[n].getWord() !=TopicWord.ANY_WORD)
{
followedByWildcards = false;
}
@@ -192,7 +227,7 @@ public class TopicParser
}
- positions[p]._followedByAnyLoop = followedByWildcards && (n!= positionCount+1);
+ positions[p].setFollowedByAnyLoop(followedByWildcards && (n!= positionCount+1));
}
@@ -221,7 +256,7 @@ public class TopicParser
for(Position p : simpleStates[i]._positions)
{
- if(p._endState)
+ if(p.isEndState())
{
endState = true;
break;
@@ -267,7 +302,7 @@ public class TopicParser
for(Position pos : state._positions)
{
- if(pos._selfTransition)
+ if(pos.isSelfTransition())
{
Set<Position> dest = transitions.get(TopicWord.ANY_WORD);
if(dest == null)
@@ -278,14 +313,14 @@ public class TopicParser
dest.add(pos);
}
- final int nextPos = pos._position + 1;
+ final int nextPos = pos.getPosition() + 1;
Position nextPosition = nextPos == positions.length ? ERROR_POSITION : positions[nextPos];
- Set<Position> dest = transitions.get(pos._word);
+ Set<Position> dest = transitions.get(pos.getWord());
if(dest == null)
{
dest = new HashSet<Position>();
- transitions.put(pos._word,dest);
+ transitions.put(pos.getWord(),dest);
}
dest.add(nextPosition);
@@ -312,7 +347,7 @@ public class TopicParser
Position loopingTerminal = null;
for(Position destPos : dest.getValue())
{
- if(destPos._selfTransition && destPos._endState)
+ if(destPos.isSelfTransition() && destPos.isEndState())
{
loopingTerminal = destPos;
break;
@@ -328,9 +363,9 @@ public class TopicParser
Position anyLoop = null;
for(Position destPos : dest.getValue())
{
- if(destPos._followedByAnyLoop)
+ if(destPos.isFollowedByAnyLoop())
{
- if(anyLoop == null || anyLoop._position<destPos._position)
+ if(anyLoop == null || anyLoop.getPosition() < destPos.getPosition())
{
anyLoop = destPos;
}
@@ -341,7 +376,7 @@ public class TopicParser
Collection<Position> removals = new ArrayList<Position>();
for(Position destPos : dest.getValue())
{
- if(destPos._position < anyLoop._position)
+ if(destPos.getPosition() < anyLoop.getPosition())
{
removals.add(destPos);
}
@@ -421,193 +456,4 @@ public class TopicParser
}
- public static void main(String[] args)
- {
-
- printMatches("#.b.*.*.*.*.*.h.#.j.*.*.*.*.*.*.q.#.r.*.*.*.*.*.*.*.*","a.b.c.d.e.f.g.h.i.j.k.l.m.n.o.p.q.r.s.t.u.v.w.x.y.z");
- printMatches(new String[]{
- "#.a.#",
- "#.b.#",
- "#.c.#",
- "#.d.#",
- "#.e.#",
- "#.f.#",
- "#.g.#",
- "#.h.#",
- "#.i.#",
- "#.j.#",
- "#.k.#",
- "#.l.#",
- "#.m.#",
- "#.n.#",
- "#.o.#",
- "#.p.#",
- "#.q.#"
-
- }, "a.b.c.d.e.f.g.h.i.j.k.l.m.n.o.p.q.r.s.t.u.v.w.x.y.z");
-/*
- printMatches(new String[]{
- "#.a.#",
- "#.b.#",
- "#.c.#",
- "#.d.#",
- "#.e.#",
- "#.f.#",
- "#.g.#",
- "#.h.#",
- "#.i.#",
- "#.j.#",
- "#.k.#",
- "#.l.#",
- "#.m.#",
- "#.n.#",
- "#.o.#",
- "#.p.#",
- "#.q.#",
- "#.r.#",
- "#.s.#",
- "#.t.#",
- "#.u.#",
- "#.v.#",
- "#.w.#",
- "#.x.#",
- "#.y.#",
- "#.z.#"
-
-
- },"a.b");
-
- printMatches("#.b.*.*.*.*.*.h.#.j.*.*.*.*.*.p.#.r.*.*.*.*.*","a.b.c.d.e.f.g.h.i.j.k.l.m.n.o.p.q.r.s.t.u.v.w.x.y.z");
- printMatches("#.b.*.*.*.*.*.h.#.j.*.*.*.*.*.p.#.r.*.*.*.*.*.*.*.*","a.b.c.d.e.f.g.h.i.j.k.l.m.n.o.p.q.r.s.t.u.v.w.x.y.z");
- printMatches("a.#.b.#","a.b.b.b.b.b.b.b.c");
-
-*/
-
- printMatches("","");
- printMatches("a","a");
- printMatches("a","");
- printMatches("","a");
- printMatches("a.b","a.b");
- printMatches("a","a.b");
- printMatches("a.b","a");
- printMatches("*","a");
- printMatches("*.b","a.b");
- printMatches("*.*","a.b");
- printMatches("a.*","a.b");
- printMatches("a.*.#","a.b");
- printMatches("a.#.b","a.b");
-
- printMatches("#.b","a");
- printMatches("#.b","a.b");
- printMatches("#.a.b","a.b");
-
-
- printMatches("#","");
- printMatches("#","a");
- printMatches("#","a.b");
- printMatches("#.#","a.b");
- printMatches("#.*","a.b");
-
- printMatches("#.a.b","a.b");
- printMatches("a.b.#","a.b");
- printMatches("a.#","a.b");
- printMatches("#.*.#","a.b");
- printMatches("#.*.b.#","a.b");
- printMatches("#.a.*.#","a.b");
- printMatches("#.a.#.b.#","a.b");
- printMatches("#.*.#.*.#","a.b");
- printMatches("*.#.*.#","a.b");
- printMatches("#.*.#.*","a.b");
-
-
- printMatches(new String[]{"a.#.b.#","a.*.#.b.#"},"a.b.b.b.b.b.b.b.c");
-
-
- printMatches(new String[]{"a.b", "a.c"},"a.b");
- printMatches(new String[]{"a.#", "a.c", "#.b"},"a.b");
- printMatches(new String[]{"a.#", "a.c", "#.b", "#", "*.*"},"a.b");
-
- printMatches(new String[]{"a.b.c.d.e.#", "a.b.c.d.#", "a.b.c.d.*", "a.b.c.#", "#.e", "a.*.c.d.e","#.c.*.#.*.*"},"a.b.c.d.e");
- printMatches(new String[]{"a.b.c.d.e.#", "a.b.c.d.#", "a.b.c.d.*", "a.b.c.#", "#.e", "a.*.c.d.e","#.c.*.#.*.*"},"a.b.c.d.f.g");
-
-
-
-
- }
-
- private static void printMatches(final String[] bindingKeys, final String routingKey)
- {
- TopicMatcherDFAState sm = null;
- Map<TopicMatcherResult, String> resultMap = new HashMap<TopicMatcherResult, String>();
-
- TopicParser parser = new TopicParser();
-
- long start = System.currentTimeMillis();
- for(int i = 0; i < bindingKeys.length; i++)
- {
- System.out.println((System.currentTimeMillis() - start) + ":\t" + bindingKeys[i]);
- TopicMatcherResult r = new TopicMatcherResult(){};
- resultMap.put(r, bindingKeys[i]);
- AMQShortString bindingKeyShortString = new AMQShortString(bindingKeys[i]);
-
- System.err.println("=====================================================");
- System.err.println("Adding binding key: " + bindingKeyShortString);
- System.err.println("-----------------------------------------------------");
-
-
- if(i==0)
- {
- sm = parser.createStateMachine(bindingKeyShortString, r);
- }
- else
- {
- sm = sm.mergeStateMachines(parser.createStateMachine(bindingKeyShortString, r));
- }
- System.err.println(sm.reachableStates());
- System.err.println("=====================================================");
- try
- {
- System.in.read();
- }
- catch (IOException e)
- {
- e.printStackTrace(); //To change body of catch statement use File | Settings | File Templates.
- }
- }
- AMQShortString routingKeyShortString = new AMQShortString(routingKey);
-
- Collection<TopicMatcherResult> results = sm.parse(parser._dictionary, routingKeyShortString);
- Collection<String> resultStrings = new ArrayList<String>();
-
- for(TopicMatcherResult result : results)
- {
- resultStrings.add(resultMap.get(result));
- }
-
- final ArrayList<String> nonMatches = new ArrayList<String>(Arrays.asList(bindingKeys));
- nonMatches.removeAll(resultStrings);
- System.out.println("\""+routingKeyShortString+"\" matched with " + resultStrings + " DID NOT MATCH with " + nonMatches);
-
-
- }
-
- private static void printMatches(String bindingKey, String routingKey)
- {
- printMatches(new String[] { bindingKey }, routingKey);
- }
-
-
- private static boolean matches(String bindingKey, String routingKey)
- {
- AMQShortString bindingKeyShortString = new AMQShortString(bindingKey);
- AMQShortString routingKeyShortString = new AMQShortString(routingKey);
- TopicParser parser = new TopicParser();
-
- final TopicMatcherResult result = new TopicMatcherResult(){};
-
- TopicMatcherDFAState sm = parser.createStateMachine(bindingKeyShortString, result);
- return !sm.parse(parser._dictionary,routingKeyShortString).isEmpty();
-
- }
-
}
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicWord.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicWord.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicWord.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicWord.java Sat Mar 10 19:22:10 2012
@@ -2,10 +2,6 @@ package org.apache.qpid.server.exchange.
import org.apache.qpid.framing.AMQShortString;
-import java.util.Map;
-import java.util.HashMap;
-import java.util.concurrent.ConcurrentHashMap;
-
/*
*
* Licensed to the Apache Software Foundation (ASF) under one
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org