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 2009/10/25 23:59:05 UTC
svn commit: r829675 [3/11] - in /qpid/trunk/qpid/java: ./
broker-plugins/src/main/java/org/apache/qpid/extras/exchanges/diagnostic/
broker-plugins/src/main/java/org/apache/qpid/extras/exchanges/example/
broker/ broker/bin/ broker/src/main/java/org/apac...
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/FanoutExchange.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/FanoutExchange.java?rev=829675&r1=829674&r2=829675&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/FanoutExchange.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/FanoutExchange.java Sun Oct 25 22:58:57 2009
@@ -28,9 +28,9 @@
import org.apache.qpid.exchange.ExchangeDefaults;
import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.framing.FieldTable;
-import org.apache.qpid.server.queue.IncomingMessage;
import org.apache.qpid.server.queue.AMQQueue;
import org.apache.qpid.server.virtualhost.VirtualHost;
+import org.apache.qpid.server.message.InboundMessage;
import org.apache.qpid.server.logging.actors.CurrentActor;
import org.apache.qpid.server.logging.actors.ManagementActor;
@@ -62,7 +62,7 @@
private final class FanoutExchangeMBean extends ExchangeMBean
{
private static final String BINDING_KEY_SUBSTITUTE = "*";
-
+
@MBeanConstructor("Creates an MBean for AMQ fanout exchange")
public FanoutExchangeMBean() throws JMException
{
@@ -75,7 +75,7 @@
{
_bindingList = new TabularDataSupport(_bindinglistDataType);
-
+
if(_queues.isEmpty())
{
return _bindingList;
@@ -88,7 +88,7 @@
String queueName = queue.getName().toString();
queueNames.add(queueName);
}
-
+
Object[] bindingItemValues = {BINDING_KEY_SUBSTITUTE, queueNames.toArray(new String[0])};
CompositeData bindingData = new CompositeDataSupport(_bindingDataType, COMPOSITE_ITEM_NAMES, bindingItemValues);
_bindingList.put(bindingData);
@@ -121,17 +121,14 @@
} // End of MBean class
- protected ExchangeMBean createMBean() throws AMQException
+ protected ExchangeMBean createMBean() throws JMException
{
- try
- {
- return new FanoutExchange.FanoutExchangeMBean();
- }
- catch (JMException ex)
- {
- _logger.error("Exception occured in creating the direct exchange mbean", ex);
- throw new AMQException("Exception occured in creating the direct exchange mbean", ex);
- }
+ return new FanoutExchange.FanoutExchangeMBean();
+ }
+
+ public Logger getLogger()
+ {
+ return _logger;
}
public static final ExchangeType<FanoutExchange> TYPE = new ExchangeType<FanoutExchange>()
@@ -199,16 +196,16 @@
}
}
- public void route(IncomingMessage payload) throws AMQException
+ public ArrayList<AMQQueue> route(InboundMessage payload)
{
-
+
if (_logger.isDebugEnabled())
{
_logger.debug("Publishing message to queue " + _queues);
}
- payload.enqueue(new ArrayList(_queues));
+ return new ArrayList(_queues);
}
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/HeadersBinding.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/HeadersBinding.java?rev=829675&r1=829674&r2=829675&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/HeadersBinding.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/HeadersBinding.java Sun Oct 25 22:58:57 2009
@@ -7,9 +7,9 @@
* 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
@@ -28,6 +28,7 @@
import org.apache.log4j.Logger;
import org.apache.qpid.framing.AMQTypedValue;
import org.apache.qpid.framing.FieldTable;
+import org.apache.qpid.server.message.AMQMessageHeader;
/**
* Defines binding and matching based on a set of headers.
@@ -87,7 +88,7 @@
* Creates a binding for a set of mappings. Those mappings whose value is
* null or the empty string are assumed only to be required headers, with
* no constraint on the value. Those with a non-null value are assumed to
- * define a required match of value.
+ * define a required match of value.
* @param mappings the defined mappings this binding should use
*/
@@ -139,7 +140,7 @@
* @return true if the headers define any required keys and match any required
* values
*/
- public boolean matches(FieldTable headers)
+ public boolean matches(AMQMessageHeader headers)
{
if(headers == null)
{
@@ -151,13 +152,13 @@
}
}
- private boolean and(FieldTable headers)
+ private boolean and(AMQMessageHeader headers)
{
- if(headers.keys().containsAll(required))
+ if(headers.containsHeaders(required))
{
for(Map.Entry<String, Object> e : matches.entrySet())
{
- if(!e.getValue().equals(headers.getObject(e.getKey())))
+ if(!e.getValue().equals(headers.getHeader(e.getKey())))
{
return false;
}
@@ -171,17 +172,50 @@
}
- private boolean or(final FieldTable headers)
+ private boolean or(final AMQMessageHeader headers)
{
- if(required.isEmpty() || !(Boolean) headers.processOverElements(new RequiredOrProcessor()))
+ if(required.isEmpty())
{
- return ((!matches.isEmpty()) && (Boolean) headers.processOverElements(new MatchesOrProcessor()))
- || (required.isEmpty() && matches.isEmpty());
+ return matches.isEmpty() || passesMatchesOr(headers);
}
else
{
- return true;
+ if(!passesRequiredOr(headers))
+ {
+ return !matches.isEmpty() && passesMatchesOr(headers);
+ }
+ else
+ {
+ return true;
+ }
+
+ }
+ }
+
+ private boolean passesMatchesOr(AMQMessageHeader headers)
+ {
+ for(Map.Entry<String,Object> entry : matches.entrySet())
+ {
+ if(headers.containsHeader(entry.getKey())
+ && ((entry.getValue() == null && headers.getHeader(entry.getKey()) == null)
+ || (entry.getValue().equals(headers.getHeader(entry.getKey())))))
+ {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private boolean passesRequiredOr(AMQMessageHeader headers)
+ {
+ for(String name : required)
+ {
+ if(headers.containsHeader(name))
+ {
+ return true;
+ }
}
+ return false;
}
private void processSpecial(String key, Object value)
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/HeadersExchange.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/HeadersExchange.java?rev=829675&r1=829674&r2=829675&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/HeadersExchange.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/HeadersExchange.java Sun Oct 25 22:58:57 2009
@@ -31,9 +31,10 @@
import org.apache.qpid.framing.BasicContentHeaderProperties;
import org.apache.qpid.framing.ContentHeaderBody;
import org.apache.qpid.framing.FieldTable;
-import org.apache.qpid.server.queue.IncomingMessage;
import org.apache.qpid.server.queue.AMQQueue;
import org.apache.qpid.server.virtualhost.VirtualHost;
+import org.apache.qpid.server.message.InboundMessage;
+import org.apache.qpid.server.message.AMQMessageHeader;
import javax.management.JMException;
import javax.management.openmbean.ArrayType;
@@ -50,8 +51,8 @@
import java.util.Iterator;
import java.util.List;
import java.util.Map;
-import java.util.Collection;
import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.ConcurrentHashMap;
/**
* An exchange that binds queues based on a set of required headers and header values
@@ -82,6 +83,7 @@
*/
public class HeadersExchange extends AbstractExchange
{
+
private static final Logger _logger = Logger.getLogger(HeadersExchange.class);
public static final ExchangeType<HeadersExchange> TYPE = new ExchangeType<HeadersExchange>()
@@ -101,6 +103,7 @@
boolean autoDelete) throws AMQException
{
HeadersExchange exch = new HeadersExchange();
+
exch.initialise(host, name, durable, ticket, autoDelete);
return exch;
}
@@ -114,6 +117,7 @@
private final List<Registration> _bindings = new CopyOnWriteArrayList<Registration>();
+ private Map<AMQShortString, Registration> _bindingByKey = new ConcurrentHashMap<AMQShortString, Registration>();
/**
* HeadersExchangeMBean class implements the management interface for the
@@ -208,7 +212,7 @@
{
throw new JMException("Format for headers binding should be \"<attribute1>=<value1>,<attribute2>=<value2>\" ");
}
-
+
if(keyAndValue.length ==1)
{
//no value was given, only a key. Use an empty value
@@ -221,7 +225,7 @@
}
}
- _bindings.add(new Registration(new HeadersBinding(bindingMap), queue));
+ _bindings.add(new Registration(new HeadersBinding(bindingMap), queue, new AMQShortString(binding)));
}
} // End of MBean class
@@ -234,44 +238,48 @@
public void registerQueue(AMQShortString routingKey, AMQQueue queue, FieldTable args) throws AMQException
{
_logger.debug("Exchange " + getName() + ": Binding " + queue.getName() + " with " + args);
- _bindings.add(new Registration(new HeadersBinding(args), queue));
+
+ Registration registration = new Registration(new HeadersBinding(args), queue, routingKey);
+ _bindings.add(registration);
+
}
public void deregisterQueue(AMQShortString routingKey, AMQQueue queue, FieldTable args) throws AMQException
{
_logger.debug("Exchange " + getName() + ": Unbinding " + queue.getName());
- if(!_bindings.remove(new Registration(new HeadersBinding(args), queue)))
+
+ if(!_bindings.remove(new Registration(args == null ? null : new HeadersBinding(args), queue, routingKey)))
{
throw new AMQException(AMQConstant.NOT_FOUND, "Queue " + queue + " was not registered with exchange " + this.getName()
- + " with headers args " + args);
+ + " with headers args " + args);
}
}
- public void route(IncomingMessage payload) throws AMQException
+ public ArrayList<AMQQueue> route(InboundMessage payload)
{
- FieldTable headers = getHeaders(payload.getContentHeaderBody());
+ AMQMessageHeader header = payload.getMessageHeader();
if (_logger.isDebugEnabled())
{
- _logger.debug("Exchange " + getName() + ": routing message with headers " + headers);
+ _logger.debug("Exchange " + getName() + ": routing message with headers " + header);
}
boolean routed = false;
ArrayList<AMQQueue> queues = new ArrayList<AMQQueue>();
for (Registration e : _bindings)
{
- if (e.binding.matches(headers))
+ if (e.binding.matches(header))
{
if (_logger.isDebugEnabled())
{
_logger.debug("Exchange " + getName() + ": delivering message with headers " +
- headers + " to " + e.queue.getName());
+ header + " to " + e.queue.getName());
}
queues.add(e.queue);
routed = true;
}
}
- payload.enqueue(queues);
+ return queues;
}
public boolean isBound(AMQShortString routingKey, FieldTable arguments, AMQQueue queue)
@@ -314,17 +322,9 @@
return ((BasicContentHeaderProperties) contentHeaderFrame.properties).getHeaders();
}
- protected ExchangeMBean createMBean() throws AMQException
+ protected ExchangeMBean createMBean() throws JMException
{
- try
- {
- return new HeadersExchangeMBean();
- }
- catch (JMException ex)
- {
- _logger.error("Exception occured in creating the HeadersExchangeMBean", ex);
- throw new AMQException("Exception occured in creating the HeadersExchangeMBean", ex);
- }
+ return new HeadersExchangeMBean();
}
public Map<AMQShortString, List<AMQQueue>> getBindings()
@@ -332,25 +332,38 @@
return null;
}
+ public Logger getLogger()
+ {
+ return _logger;
+ }
+
+
private static class Registration
{
private final HeadersBinding binding;
private final AMQQueue queue;
+ private final AMQShortString routingKey;
- Registration(HeadersBinding binding, AMQQueue queue)
+ Registration(HeadersBinding binding, AMQQueue queue, AMQShortString routingKey)
{
this.binding = binding;
this.queue = queue;
+ this.routingKey = routingKey;
}
public int hashCode()
{
- return queue.hashCode();
+ int queueHash = queue.hashCode();
+ int routingHash = routingKey == null ? 0 : routingKey.hashCode();
+ return queueHash + routingHash;
}
public boolean equals(Object o)
{
- return o instanceof Registration && ((Registration) o).queue.equals(queue);
+ return o instanceof Registration
+ && ((Registration) o).queue.equals(queue)
+ && (routingKey == null ? ((Registration)o).routingKey == null
+ : routingKey.equals(((Registration)o).routingKey));
}
}
}
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/Index.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/Index.java?rev=829675&r1=829674&r2=829675&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/Index.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/Index.java Sun Oct 25 22:58:57 2009
@@ -39,6 +39,9 @@
{
private ConcurrentMap<AMQShortString, ArrayList<AMQQueue>> _index
= new ConcurrentHashMap<AMQShortString, ArrayList<AMQQueue>>();
+ private ConcurrentMap<String, ArrayList<AMQQueue>> _stringIndex
+ = new ConcurrentHashMap<String, ArrayList<AMQQueue>>();
+
synchronized boolean add(AMQShortString key, AMQQueue queue)
{
@@ -51,8 +54,10 @@
{
queues = new ArrayList<AMQQueue>(queues);
}
+
//next call is atomic, so there is no race to create the list
_index.put(key, queues);
+ _stringIndex.put(key.toString(), queues);
if(queues.contains(queue))
{
@@ -64,6 +69,8 @@
}
}
+
+
synchronized boolean remove(AMQShortString key, AMQQueue queue)
{
ArrayList<AMQQueue> queues = _index.get(key);
@@ -76,10 +83,12 @@
if (queues.size() == 0)
{
_index.remove(key);
+ _stringIndex.remove(key.toString());
}
else
{
_index.put(key, queues);
+ _stringIndex.put(key.toString(), queues);
}
}
return removed;
@@ -92,6 +101,12 @@
return _index.get(key);
}
+ ArrayList<AMQQueue> get(String key)
+ {
+ return _stringIndex.get(key);
+ }
+
+
Map<AMQShortString, List<AMQQueue>> getBindingsMap()
{
return new HashMap<AMQShortString, List<AMQQueue>>(_index);
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/MessageRouter.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/MessageRouter.java?rev=829675&r1=829674&r2=829675&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/MessageRouter.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/MessageRouter.java Sun Oct 25 22:58:57 2009
@@ -7,9 +7,9 @@
* 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
@@ -21,7 +21,6 @@
package org.apache.qpid.server.exchange;
import org.apache.qpid.AMQException;
-import org.apache.qpid.server.queue.AMQMessage;
import org.apache.qpid.server.queue.IncomingMessage;
/**
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/TopicExchange.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/TopicExchange.java?rev=829675&r1=829674&r2=829675&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/TopicExchange.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/TopicExchange.java Sun Oct 25 22:58:57 2009
@@ -30,13 +30,13 @@
import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.framing.FieldTable;
import org.apache.qpid.framing.AMQShortStringTokenizer;
-import org.apache.qpid.server.queue.IncomingMessage;
import org.apache.qpid.server.queue.AMQQueue;
import org.apache.qpid.server.virtualhost.VirtualHost;
import org.apache.qpid.server.exchange.topic.TopicParser;
import org.apache.qpid.server.exchange.topic.TopicMatcherResult;
import org.apache.qpid.server.filter.MessageFilter;
import org.apache.qpid.server.filter.JMSSelectorFilter;
+import org.apache.qpid.server.message.InboundMessage;
import org.apache.qpid.server.logging.actors.CurrentActor;
import org.apache.qpid.server.logging.actors.ManagementActor;
@@ -111,7 +111,7 @@
private final Map<Binding, FieldTable> _bindings = new HashMap<Binding, FieldTable>();
- private final Map<String, WeakReference<JMSSelectorFilter<RuntimeException>>> _selectorCache = new WeakHashMap<String, WeakReference<JMSSelectorFilter<RuntimeException>>>();
+ private final Map<String, WeakReference<JMSSelectorFilter>> _selectorCache = new WeakHashMap<String, WeakReference<JMSSelectorFilter>>();
public static class Binding
{
@@ -162,7 +162,7 @@
private final class TopicExchangeResult implements TopicMatcherResult
{
private final Map<AMQQueue, Integer> _unfilteredQueues = new ConcurrentHashMap<AMQQueue, Integer>();
- private final ConcurrentHashMap<AMQQueue, Map<MessageFilter<RuntimeException>,Integer>> _filteredQueues = new ConcurrentHashMap<AMQQueue, Map<MessageFilter<RuntimeException>, Integer>>();
+ private final ConcurrentHashMap<AMQQueue, Map<MessageFilter,Integer>> _filteredQueues = new ConcurrentHashMap<AMQQueue, Map<MessageFilter, Integer>>();
public void addUnfilteredQueue(AMQQueue queue)
{
@@ -192,12 +192,12 @@
}
- public void addFilteredQueue(AMQQueue queue, MessageFilter<RuntimeException> filter)
+ public void addFilteredQueue(AMQQueue queue, MessageFilter filter)
{
- Map<MessageFilter<RuntimeException>,Integer> filters = _filteredQueues.get(queue);
+ Map<MessageFilter,Integer> filters = _filteredQueues.get(queue);
if(filters == null)
{
- filters = new ConcurrentHashMap<MessageFilter<RuntimeException>,Integer>();
+ filters = new ConcurrentHashMap<MessageFilter,Integer>();
_filteredQueues.put(queue, filters);
}
Integer instances = filters.get(filter);
@@ -212,9 +212,9 @@
}
- public void removeFilteredQueue(AMQQueue queue, MessageFilter<RuntimeException> filter)
+ public void removeFilteredQueue(AMQQueue queue, MessageFilter filter)
{
- Map<MessageFilter<RuntimeException>,Integer> filters = _filteredQueues.get(queue);
+ Map<MessageFilter,Integer> filters = _filteredQueues.get(queue);
if(filters != null)
{
Integer instances = filters.get(filter);
@@ -228,7 +228,7 @@
_filteredQueues.remove(queue);
}
}
- else
+ else
{
filters.put(filter, instances - 1);
}
@@ -239,11 +239,11 @@
}
public void replaceQueueFilter(AMQQueue queue,
- MessageFilter<RuntimeException> oldFilter,
- MessageFilter<RuntimeException> newFilter)
+ MessageFilter oldFilter,
+ MessageFilter newFilter)
{
- Map<MessageFilter<RuntimeException>,Integer> filters = _filteredQueues.get(queue);
- Map<MessageFilter<RuntimeException>,Integer> newFilters = new ConcurrentHashMap<MessageFilter<RuntimeException>,Integer>(filters);
+ Map<MessageFilter,Integer> filters = _filteredQueues.get(queue);
+ Map<MessageFilter,Integer> newFilters = new ConcurrentHashMap<MessageFilter,Integer>(filters);
Integer oldFilterInstances = filters.get(oldFilter);
if(oldFilterInstances == 1)
{
@@ -265,7 +265,7 @@
_filteredQueues.put(queue,newFilters);
}
- public Collection<AMQQueue> processMessage(IncomingMessage msg, Collection<AMQQueue> queues)
+ public Collection<AMQQueue> processMessage(InboundMessage msg, Collection<AMQQueue> queues)
{
if(queues == null)
{
@@ -286,11 +286,11 @@
queues.addAll(_unfilteredQueues.keySet());
if(!_filteredQueues.isEmpty())
{
- for(Map.Entry<AMQQueue, Map<MessageFilter<RuntimeException>, Integer>> entry : _filteredQueues.entrySet())
+ for(Map.Entry<AMQQueue, Map<MessageFilter, Integer>> entry : _filteredQueues.entrySet())
{
if(!queues.contains(entry.getKey()))
{
- for(MessageFilter<RuntimeException> filter : entry.getValue().keySet())
+ for(MessageFilter filter : entry.getValue().keySet())
{
if(filter.matches(msg))
{
@@ -443,10 +443,10 @@
{
result.addUnfilteredQueue(queue);
}
- _parser.addBinding(routingKey, result);
+ _parser.addBinding(routingKey, result);
_topicExchangeResults.put(routingKey,result);
}
- else
+ else
{
if(argumentsContainSelector(args))
{
@@ -463,18 +463,18 @@
}
- private JMSSelectorFilter<RuntimeException> createSelectorFilter(final FieldTable args)
+ private JMSSelectorFilter createSelectorFilter(final FieldTable args)
throws AMQException
{
final String selectorString = args.getString(AMQPFilterTypes.JMS_SELECTOR.getValue());
- WeakReference<JMSSelectorFilter<RuntimeException>> selectorRef = _selectorCache.get(selectorString);
+ WeakReference<JMSSelectorFilter> selectorRef = _selectorCache.get(selectorString);
JMSSelectorFilter selector = null;
if(selectorRef == null || (selector = selectorRef.get())==null)
{
- selector = new JMSSelectorFilter<RuntimeException>(selectorString);
- _selectorCache.put(selectorString, new WeakReference<JMSSelectorFilter<RuntimeException>>(selector));
+ selector = new JMSSelectorFilter(selectorString);
+ _selectorCache.put(selectorString, new WeakReference<JMSSelectorFilter>(selector));
}
return selector;
}
@@ -490,7 +490,7 @@
{
routingKey = AMQShortString.EMPTY_STRING;
}
-
+
AMQShortStringTokenizer routingTokens = routingKey.tokenize(TOPIC_SEPARATOR);
List<AMQShortString> subscriptionList = new ArrayList<AMQShortString>();
@@ -535,10 +535,12 @@
return normalizedString;
}
- public void route(IncomingMessage payload) throws AMQException
+ public ArrayList<AMQQueue> route(InboundMessage payload)
{
- final AMQShortString routingKey = payload.getRoutingKey();
+ final AMQShortString routingKey = payload.getRoutingKey() == null
+ ? AMQShortString.EMPTY_STRING
+ : new AMQShortString(payload.getRoutingKey());
// The copy here is unfortunate, but not too bad relevant to the amount of
// things created and copied in getMatchedQueues
@@ -550,7 +552,7 @@
_logger.info("Message routing key: " + payload.getRoutingKey() + " No routes.");
}
- payload.enqueue(queues);
+ return queues;
}
@@ -572,7 +574,7 @@
{
return false;
}
-
+
}
}
@@ -640,20 +642,17 @@
}
- protected ExchangeMBean createMBean() throws AMQException
+ protected ExchangeMBean createMBean() throws JMException
{
- try
- {
- return new TopicExchangeMBean();
- }
- catch (JMException ex)
- {
- _logger.error("Exception occured in creating the topic exchenge mbean", ex);
- throw new AMQException("Exception occured in creating the topic exchenge mbean", ex);
- }
+ return new TopicExchangeMBean();
+ }
+
+ public Logger getLogger()
+ {
+ return _logger;
}
- private Collection<AMQQueue> getMatchedQueues(IncomingMessage message, AMQShortString routingKey)
+ private Collection<AMQQueue> getMatchedQueues(InboundMessage message, AMQShortString routingKey)
{
Collection<TopicMatcherResult> results = _parser.parse(routingKey);
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/ArithmeticExpression.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/ArithmeticExpression.java?rev=829675&r1=829674&r2=829675&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/ArithmeticExpression.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/ArithmeticExpression.java Sun Oct 25 22:58:57 2009
@@ -26,7 +26,7 @@
/**
* An expression which performs an operation on two expression values
*/
-public abstract class ArithmeticExpression<E extends Exception> extends BinaryExpression<E>
+public abstract class ArithmeticExpression extends BinaryExpression
{
protected static final int INTEGER = 1;
@@ -248,7 +248,7 @@
}
}
- public Object evaluate(Filterable<E> message) throws E
+ public Object evaluate(Filterable message)
{
Object lvalue = left.evaluate(message);
if (lvalue == null)
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/BinaryExpression.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/BinaryExpression.java?rev=829675&r1=829674&r2=829675&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/BinaryExpression.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/BinaryExpression.java Sun Oct 25 22:58:57 2009
@@ -23,23 +23,23 @@
/**
* An expression which performs an operation on two expression values.
*/
-public abstract class BinaryExpression<E extends Exception> implements Expression<E>
+public abstract class BinaryExpression implements Expression
{
- protected Expression<E> left;
- protected Expression<E> right;
+ protected Expression left;
+ protected Expression right;
- public BinaryExpression(Expression<E> left, Expression<E> right)
+ public BinaryExpression(Expression left, Expression right)
{
this.left = left;
this.right = right;
}
- public Expression<E> getLeft()
+ public Expression getLeft()
{
return left;
}
- public Expression<E> getRight()
+ public Expression getRight()
{
return right;
}
@@ -90,7 +90,7 @@
/**
* @param expression
*/
- public void setRight(Expression<E> expression)
+ public void setRight(Expression expression)
{
right = expression;
}
@@ -98,7 +98,7 @@
/**
* @param expression
*/
- public void setLeft(Expression<E> expression)
+ public void setLeft(Expression expression)
{
left = expression;
}
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/BooleanExpression.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/BooleanExpression.java?rev=829675&r1=829674&r2=829675&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/BooleanExpression.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/BooleanExpression.java Sun Oct 25 22:58:57 2009
@@ -20,22 +20,19 @@
// Based on like named file from r450141 of the Apache ActiveMQ project <http://www.activemq.org/site/home.html>
//
-import org.apache.qpid.AMQException;
-import org.apache.qpid.server.queue.AMQMessage;
import org.apache.qpid.server.queue.Filterable;
/**
* A BooleanExpression is an expression that always
* produces a Boolean result.
*/
-public interface BooleanExpression<E extends Exception> extends Expression<E>
+public interface BooleanExpression extends Expression
{
/**
* @param message
* @return true if the expression evaluates to Boolean.TRUE.
- * @throws E
*/
- public boolean matches(Filterable<E> message) throws E;
+ public boolean matches(Filterable message);
}
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/ComparisonExpression.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/ComparisonExpression.java?rev=829675&r1=829674&r2=829675&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/ComparisonExpression.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/ComparisonExpression.java Sun Oct 25 22:58:57 2009
@@ -27,22 +27,20 @@
import java.util.List;
import java.util.regex.Pattern;
-import org.apache.qpid.AMQException;
-import org.apache.qpid.server.queue.AMQMessage;
import org.apache.qpid.server.queue.Filterable;
/**
* A filter performing a comparison of two objects
*/
-public abstract class ComparisonExpression<E extends Exception> extends BinaryExpression<E> implements BooleanExpression<E>
+public abstract class ComparisonExpression extends BinaryExpression implements BooleanExpression
{
- public static<E extends Exception> BooleanExpression<E> createBetween(Expression<E> value, Expression left, Expression<E> right)
+ public static BooleanExpression createBetween(Expression value, Expression left, Expression right)
{
return LogicExpression.createAND(createGreaterThanEqual(value, left), createLessThanEqual(value, right));
}
- public static<E extends Exception> BooleanExpression<E> createNotBetween(Expression<E> value, Expression<E> left, Expression<E> right)
+ public static BooleanExpression createNotBetween(Expression value, Expression left, Expression right)
{
return LogicExpression.createOR(createLessThan(value, left), createGreaterThan(value, right));
}
@@ -73,7 +71,7 @@
REGEXP_CONTROL_CHARS.add(new Character('!'));
}
- static class LikeExpression<E extends Exception> extends UnaryExpression<E> implements BooleanExpression<E>
+ static class LikeExpression extends UnaryExpression implements BooleanExpression
{
Pattern likePattern;
@@ -81,7 +79,7 @@
/**
* @param right
*/
- public LikeExpression(Expression<E> right, String like, int escape)
+ public LikeExpression(Expression right, String like, int escape)
{
super(right);
@@ -138,7 +136,7 @@
/**
* org.apache.activemq.filter.Expression#evaluate(MessageEvaluationContext)
*/
- public Object evaluate(Filterable<E> message) throws E
+ public Object evaluate(Filterable message)
{
Object rv = this.getRight().evaluate(message);
@@ -158,7 +156,7 @@
return likePattern.matcher((String) rv).matches() ? Boolean.TRUE : Boolean.FALSE;
}
- public boolean matches(Filterable<E> message) throws E
+ public boolean matches(Filterable message)
{
Object object = evaluate(message);
@@ -236,7 +234,7 @@
return doCreateEqual(left, right);
}
- private static<E extends Exception> BooleanExpression<E> doCreateEqual(Expression<E> left, Expression<E> right)
+ private static BooleanExpression doCreateEqual(Expression left, Expression right)
{
return new EqualExpression(left, right);
}
@@ -388,7 +386,7 @@
super(left, right);
}
- public Object evaluate(Filterable<E> message) throws E
+ public Object evaluate(Filterable message)
{
Comparable lv = (Comparable) left.evaluate(message);
if (lv == null)
@@ -550,21 +548,21 @@
protected abstract boolean asBoolean(int answer);
- public boolean matches(Filterable<E> message) throws E
+ public boolean matches(Filterable message)
{
Object object = evaluate(message);
return (object != null) && (object == Boolean.TRUE);
}
- private static class EqualExpression<E extends Exception> extends ComparisonExpression<E>
+ private static class EqualExpression extends ComparisonExpression
{
- public EqualExpression(final Expression<E> left, final Expression<E> right)
+ public EqualExpression(final Expression left, final Expression right)
{
super(left, right);
}
- public Object evaluate(Filterable<E> message) throws E
+ public Object evaluate(Filterable message)
{
Object lv = left.evaluate(message);
Object rv = right.evaluate(message);
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/ConstantExpression.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/ConstantExpression.java?rev=829675&r1=829674&r2=829675&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/ConstantExpression.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/ConstantExpression.java Sun Oct 25 22:58:57 2009
@@ -25,24 +25,22 @@
import java.math.BigDecimal;
-import org.apache.qpid.AMQException;
-import org.apache.qpid.server.queue.AMQMessage;
import org.apache.qpid.server.queue.Filterable;
/**
* Represents a constant expression
*/
-public class ConstantExpression<E extends Exception> implements Expression<E>
+public class ConstantExpression implements Expression
{
- static class BooleanConstantExpression<E extends Exception> extends ConstantExpression<E> implements BooleanExpression<E>
+ static class BooleanConstantExpression extends ConstantExpression implements BooleanExpression
{
public BooleanConstantExpression(Object value)
{
super(value);
}
- public boolean matches(Filterable<E> message) throws E
+ public boolean matches(Filterable message)
{
Object object = evaluate(message);
@@ -121,7 +119,7 @@
this.value = value;
}
- public Object evaluate(Filterable<E> message) throws E
+ public Object evaluate(Filterable message)
{
return value;
}
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/Expression.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/Expression.java?rev=829675&r1=829674&r2=829675&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/Expression.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/Expression.java Sun Oct 25 22:58:57 2009
@@ -20,19 +20,17 @@
// Based on like named file from r450141 of the Apache ActiveMQ project <http://www.activemq.org/site/home.html>
//
-import org.apache.qpid.AMQException;
-import org.apache.qpid.server.queue.AMQMessage;
import org.apache.qpid.server.queue.Filterable;
/**
* Represents an expression
*/
-public interface Expression<E extends Exception>
+public interface Expression
{
/**
* @return the value of this expression
*/
- public Object evaluate(Filterable<E> message) throws E;
+ public Object evaluate(Filterable message);
}
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/FilterManager.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/FilterManager.java?rev=829675&r1=829674&r2=829675&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/FilterManager.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/FilterManager.java Sun Oct 25 22:58:57 2009
@@ -14,26 +14,24 @@
* "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.
+ * under the License.
+ *
*
- *
*/
package org.apache.qpid.server.filter;
//
// Based on like named file from r450141 of the Apache ActiveMQ project <http://www.activemq.org/site/home.html>
//
-import org.apache.qpid.server.queue.AMQMessage;
import org.apache.qpid.server.queue.Filterable;
-import org.apache.qpid.AMQException;
-public interface FilterManager<E extends Exception>
+public interface FilterManager
{
- void add(MessageFilter<E> filter);
+ void add(MessageFilter filter);
- void remove(MessageFilter<E> filter);
+ void remove(MessageFilter filter);
- boolean allAllow(Filterable<E> msg);
+ boolean allAllow(Filterable msg);
boolean hasFilters();
}
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/JMSSelectorFilter.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/JMSSelectorFilter.java?rev=829675&r1=829674&r2=829675&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/JMSSelectorFilter.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/JMSSelectorFilter.java Sun Oct 25 22:58:57 2009
@@ -26,12 +26,12 @@
import org.apache.qpid.server.queue.Filterable;
-public class JMSSelectorFilter<E extends Exception> implements MessageFilter<E>
+public class JMSSelectorFilter implements MessageFilter
{
private final static Logger _logger = org.apache.log4j.Logger.getLogger(JMSSelectorFilter.class);
private String _selector;
- private BooleanExpression<E> _matcher;
+ private BooleanExpression _matcher;
public JMSSelectorFilter(String selector) throws AMQException
{
@@ -39,7 +39,7 @@
_matcher = new SelectorParser().parse(selector);
}
- public boolean matches(Filterable<E> message) throws E
+ public boolean matches(Filterable message)
{
boolean match = _matcher.matches(message);
if(_logger.isDebugEnabled())
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/LogicExpression.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/LogicExpression.java?rev=829675&r1=829674&r2=829675&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/LogicExpression.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/LogicExpression.java Sun Oct 25 22:58:57 2009
@@ -20,22 +20,20 @@
// Based on like named file from r450141 of the Apache ActiveMQ project <http://www.activemq.org/site/home.html>
//
-import org.apache.qpid.AMQException;
-import org.apache.qpid.server.queue.AMQMessage;
import org.apache.qpid.server.queue.Filterable;
/**
* A filter performing a comparison of two objects
*/
-public abstract class LogicExpression<E extends Exception> extends BinaryExpression<E> implements BooleanExpression<E>
+public abstract class LogicExpression extends BinaryExpression implements BooleanExpression
{
- public static<E extends Exception> BooleanExpression createOR(BooleanExpression<E> lvalue, BooleanExpression<E> rvalue)
+ public static BooleanExpression createOR(BooleanExpression lvalue, BooleanExpression rvalue)
{
return new OrExpression(lvalue, rvalue);
}
- public static<E extends Exception> BooleanExpression createAND(BooleanExpression<E> lvalue, BooleanExpression<E> rvalue)
+ public static BooleanExpression createAND(BooleanExpression lvalue, BooleanExpression rvalue)
{
return new AndExpression(lvalue, rvalue);
}
@@ -49,23 +47,23 @@
super(left, right);
}
- public abstract Object evaluate(Filterable<E> message) throws E;
+ public abstract Object evaluate(Filterable message);
- public boolean matches(Filterable<E> message) throws E
+ public boolean matches(Filterable message)
{
Object object = evaluate(message);
return (object != null) && (object == Boolean.TRUE);
}
- private static class OrExpression<E extends Exception> extends LogicExpression<E>
+ private static class OrExpression extends LogicExpression
{
- public OrExpression(final BooleanExpression<E> lvalue, final BooleanExpression<E> rvalue)
+ public OrExpression(final BooleanExpression lvalue, final BooleanExpression rvalue)
{
super(lvalue, rvalue);
}
- public Object evaluate(Filterable<E> message) throws E
+ public Object evaluate(Filterable message)
{
Boolean lv = (Boolean) left.evaluate(message);
@@ -86,14 +84,14 @@
}
}
- private static class AndExpression<E extends Exception> extends LogicExpression<E>
+ private static class AndExpression extends LogicExpression
{
- public AndExpression(final BooleanExpression<E> lvalue, final BooleanExpression<E> rvalue)
+ public AndExpression(final BooleanExpression lvalue, final BooleanExpression rvalue)
{
super(lvalue, rvalue);
}
- public Object evaluate(Filterable<E> message) throws E
+ public Object evaluate(Filterable message)
{
Boolean lv = (Boolean) left.evaluate(message);
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/MessageFilter.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/MessageFilter.java?rev=829675&r1=829674&r2=829675&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/MessageFilter.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/MessageFilter.java Sun Oct 25 22:58:57 2009
@@ -14,17 +14,15 @@
* "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.
+ * under the License.
+ *
*
- *
*/
package org.apache.qpid.server.filter;
-import org.apache.qpid.AMQException;
-import org.apache.qpid.server.queue.AMQMessage;
import org.apache.qpid.server.queue.Filterable;
-public interface MessageFilter<E extends Exception>
+public interface MessageFilter
{
- boolean matches(Filterable<E> message) throws E;
+ boolean matches(Filterable message);
}
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/PropertyExpression.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/PropertyExpression.java?rev=829675&r1=829674&r2=829675&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/PropertyExpression.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/PropertyExpression.java Sun Oct 25 22:58:57 2009
@@ -27,7 +27,6 @@
import org.apache.log4j.Logger;
-import org.apache.qpid.AMQException;
import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.framing.CommonContentHeaderProperties;
import org.apache.qpid.server.queue.Filterable;
@@ -35,7 +34,7 @@
/**
* Represents a property expression
*/
-public class PropertyExpression<E extends Exception> implements Expression<E>
+public class PropertyExpression implements Expression
{
// Constants - defined the same as JMS
private static final int NON_PERSISTENT = 1;
@@ -44,12 +43,12 @@
private static final Logger _logger = org.apache.log4j.Logger.getLogger(PropertyExpression.class);
- private static final HashMap<String, Expression<? extends Exception>> JMS_PROPERTY_EXPRESSIONS = new HashMap<String, Expression<? extends Exception>>();
+ private static final HashMap<String, Expression> JMS_PROPERTY_EXPRESSIONS = new HashMap<String, Expression>();
{
- JMS_PROPERTY_EXPRESSIONS.put("JMSDestination", new Expression<E>()
+ JMS_PROPERTY_EXPRESSIONS.put("JMSDestination", new Expression()
{
- public Object evaluate(Filterable<E> message)
+ public Object evaluate(Filterable message)
{
//TODO
return null;
@@ -73,9 +72,9 @@
JMS_PROPERTY_EXPRESSIONS.put("JMSExpiration", new ExpirationExpression());
- JMS_PROPERTY_EXPRESSIONS.put("JMSRedelivered", new Expression<E>()
+ JMS_PROPERTY_EXPRESSIONS.put("JMSRedelivered", new Expression()
{
- public Object evaluate(Filterable message) throws E
+ public Object evaluate(Filterable message)
{
return message.isRedelivered();
}
@@ -83,7 +82,7 @@
}
private final String name;
- private final Expression<E> jmsPropertyExpression;
+ private final Expression jmsPropertyExpression;
public boolean outerTest()
{
@@ -96,10 +95,10 @@
- jmsPropertyExpression = (Expression<E>) JMS_PROPERTY_EXPRESSIONS.get(name);
+ jmsPropertyExpression = (Expression) JMS_PROPERTY_EXPRESSIONS.get(name);
}
- public Object evaluate(Filterable<E> message) throws E
+ public Object evaluate(Filterable message)
{
if (jmsPropertyExpression != null)
@@ -108,17 +107,7 @@
}
else
{
-
- CommonContentHeaderProperties _properties =
- (CommonContentHeaderProperties) message.getContentHeaderBody().properties;
-
- if (_logger.isDebugEnabled())
- {
- _logger.debug("Looking up property:" + name);
- _logger.debug("Properties are:" + _properties.getHeaders().keySet());
- }
-
- return _properties.getHeaders().getObject(name);
+ return message.getMessageHeader().getHeader(name);
}
}
@@ -158,39 +147,30 @@
}
- private static class ReplyToExpression<E extends Exception> implements Expression<E>
+ private static class ReplyToExpression implements Expression
{
- public Object evaluate(Filterable<E> message) throws E
+ public Object evaluate(Filterable message)
{
-
- CommonContentHeaderProperties _properties =
- (CommonContentHeaderProperties)
- message.getContentHeaderBody().properties;
- AMQShortString replyTo = _properties.getReplyTo();
-
- return (replyTo == null) ? null : replyTo.toString();
-
+ String replyTo = message.getMessageHeader().getReplyTo();
+ return replyTo;
}
}
- private static class TypeExpression<E extends Exception> implements Expression<E>
+ private static class TypeExpression implements Expression
{
- public Object evaluate(Filterable<E> message) throws E
+ public Object evaluate(Filterable message)
{
- CommonContentHeaderProperties _properties =
- (CommonContentHeaderProperties)
- message.getContentHeaderBody().properties;
- AMQShortString type = _properties.getType();
- return (type == null) ? null : type.toString();
+ String type = message.getMessageHeader().getType();
+ return type;
}
}
- private static class DeliveryModeExpression<E extends Exception> implements Expression<E>
+ private static class DeliveryModeExpression implements Expression
{
- public Object evaluate(Filterable<E> message) throws E
+ public Object evaluate(Filterable message)
{
int mode = message.isPersistent() ? PERSISTENT : NON_PERSISTENT;
if (_logger.isDebugEnabled())
@@ -202,68 +182,53 @@
}
}
- private static class PriorityExpression<E extends Exception> implements Expression<E>
+ private static class PriorityExpression implements Expression
{
- public Object evaluate(Filterable<E> message) throws E
+ public Object evaluate(Filterable message)
{
- CommonContentHeaderProperties _properties =
- (CommonContentHeaderProperties)
- message.getContentHeaderBody().properties;
-
- return (int) _properties.getPriority();
+ byte priority = message.getMessageHeader().getPriority();
+ return (int) priority;
}
}
- private static class MessageIDExpression<E extends Exception> implements Expression<E>
+ private static class MessageIDExpression implements Expression
{
- public Object evaluate(Filterable<E> message) throws E
+ public Object evaluate(Filterable message)
{
- CommonContentHeaderProperties _properties =
- (CommonContentHeaderProperties)
- message.getContentHeaderBody().properties;
- AMQShortString messageId = _properties.getMessageId();
+ String messageId = message.getMessageHeader().getMessageId();
- return (messageId == null) ? null : messageId;
+ return messageId;
}
}
- private static class TimestampExpression<E extends Exception> implements Expression<E>
+ private static class TimestampExpression implements Expression
{
- public Object evaluate(Filterable<E> message) throws E
+ public Object evaluate(Filterable message)
{
- CommonContentHeaderProperties _properties =
- (CommonContentHeaderProperties)
- message.getContentHeaderBody().properties;
-
- return _properties.getTimestamp();
+ long timestamp = message.getMessageHeader().getTimestamp();
+ return timestamp;
}
}
- private static class CorrelationIdExpression<E extends Exception> implements Expression<E>
+ private static class CorrelationIdExpression implements Expression
{
- public Object evaluate(Filterable<E> message) throws E
+ public Object evaluate(Filterable message)
{
- CommonContentHeaderProperties _properties =
- (CommonContentHeaderProperties)
- message.getContentHeaderBody().properties;
- AMQShortString correlationId = _properties.getCorrelationId();
- return (correlationId == null) ? null : correlationId.toString();
+ String correlationId = message.getMessageHeader().getCorrelationId();
+
+ return correlationId;
}
}
- private static class ExpirationExpression<E extends Exception> implements Expression<E>
+ private static class ExpirationExpression implements Expression
{
- public Object evaluate(Filterable<E> message) throws E
+ public Object evaluate(Filterable message)
{
-
- CommonContentHeaderProperties _properties =
- (CommonContentHeaderProperties)
- message.getContentHeaderBody().properties;
-
- return _properties.getExpiration();
+ long expiration = message.getMessageHeader().getExpiration();
+ return expiration;
}
}
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/SimpleFilterManager.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/SimpleFilterManager.java?rev=829675&r1=829674&r2=829675&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/SimpleFilterManager.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/SimpleFilterManager.java Sun Oct 25 22:58:57 2009
@@ -26,46 +26,37 @@
import org.apache.qpid.AMQException;
import org.apache.qpid.server.queue.Filterable;
-public class SimpleFilterManager implements FilterManager<AMQException>
+public class SimpleFilterManager implements FilterManager
{
private final Logger _logger = Logger.getLogger(SimpleFilterManager.class);
- private final ConcurrentLinkedQueue<MessageFilter<AMQException>> _filters;
+ private final ConcurrentLinkedQueue<MessageFilter> _filters;
private String _toString = "";
public SimpleFilterManager()
{
_logger.debug("Creating SimpleFilterManager");
- _filters = new ConcurrentLinkedQueue<MessageFilter<AMQException>>();
+ _filters = new ConcurrentLinkedQueue<MessageFilter>();
}
- public void add(MessageFilter<AMQException> filter)
+ public void add(MessageFilter filter)
{
_filters.add(filter);
updateStringValue();
}
- public void remove(MessageFilter<AMQException> filter)
+ public void remove(MessageFilter filter)
{
_filters.remove(filter);
updateStringValue();
}
- public boolean allAllow(Filterable<AMQException> msg)
+ public boolean allAllow(Filterable msg)
{
- for (MessageFilter<AMQException> filter : _filters)
+ for (MessageFilter filter : _filters)
{
- try
+ if (!filter.matches(msg))
{
- if (!filter.matches(msg))
- {
- return false;
- }
- }
- catch (AMQException e)
- {
- //fixme
- e.printStackTrace();
return false;
}
}
@@ -87,7 +78,7 @@
private void updateStringValue()
{
StringBuilder toString = new StringBuilder();
- for (MessageFilter<AMQException> filter : _filters)
+ for (MessageFilter filter : _filters)
{
toString.append(filter.toString());
toString.append(",");
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/UnaryExpression.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/UnaryExpression.java?rev=829675&r1=829674&r2=829675&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/UnaryExpression.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/UnaryExpression.java Sun Oct 25 22:58:57 2009
@@ -35,18 +35,18 @@
/**
* An expression which performs an operation on two expression values
*/
-public abstract class UnaryExpression<E extends Exception> implements Expression<E>
+public abstract class UnaryExpression implements Expression
{
private static final BigDecimal BD_LONG_MIN_VALUE = BigDecimal.valueOf(Long.MIN_VALUE);
- protected Expression<E> right;
+ protected Expression right;
- public static<E extends Exception> Expression<E> createNegate(Expression<E> left)
+ public static Expression createNegate(Expression left)
{
return new NegativeExpression(left);
}
- public static<E extends Exception> BooleanExpression createInExpression(PropertyExpression<E> right, List elements, final boolean not)
+ public static BooleanExpression createInExpression(PropertyExpression right, List elements, final boolean not)
{
// Use a HashSet if there are many elements.
@@ -69,14 +69,14 @@
return new InExpression(right, inList, not);
}
- abstract static class BooleanUnaryExpression<E extends Exception> extends UnaryExpression<E> implements BooleanExpression<E>
+ abstract static class BooleanUnaryExpression extends UnaryExpression implements BooleanExpression
{
- public BooleanUnaryExpression(Expression<E> left)
+ public BooleanUnaryExpression(Expression left)
{
super(left);
}
- public boolean matches(Filterable<E> message) throws E
+ public boolean matches(Filterable message)
{
Object object = evaluate(message);
@@ -85,7 +85,7 @@
}
;
- public static<E extends Exception> BooleanExpression<E> createNOT(BooleanExpression<E> left)
+ public static<E extends Exception> BooleanExpression createNOT(BooleanExpression left)
{
return new NotExpression(left);
}
@@ -100,7 +100,7 @@
return new XQueryExpression(xpath);
}
- public static<E extends Exception> BooleanExpression createBooleanCast(Expression<E> left)
+ public static<E extends Exception> BooleanExpression createBooleanCast(Expression left)
{
return new BooleanCastExpression(left);
}
@@ -151,7 +151,7 @@
this.right = left;
}
- public Expression<E> getRight()
+ public Expression getRight()
{
return right;
}
@@ -204,14 +204,14 @@
*/
public abstract String getExpressionSymbol();
- private static class NegativeExpression<E extends Exception> extends UnaryExpression<E>
+ private static class NegativeExpression extends UnaryExpression
{
- public NegativeExpression(final Expression<E> left)
+ public NegativeExpression(final Expression left)
{
super(left);
}
- public Object evaluate(Filterable<E> message) throws E
+ public Object evaluate(Filterable message)
{
Object rvalue = right.evaluate(message);
if (rvalue == null)
@@ -233,19 +233,19 @@
}
}
- private static class InExpression<E extends Exception> extends BooleanUnaryExpression<E>
+ private static class InExpression extends BooleanUnaryExpression
{
private final Collection _inList;
private final boolean _not;
- public InExpression(final PropertyExpression<E> right, final Collection inList, final boolean not)
+ public InExpression(final PropertyExpression right, final Collection inList, final boolean not)
{
super(right);
_inList = inList;
_not = not;
}
- public Object evaluate(Filterable<E> message) throws E
+ public Object evaluate(Filterable message)
{
Object rvalue = right.evaluate(message);
@@ -309,14 +309,14 @@
}
}
- private static class NotExpression<E extends Exception> extends BooleanUnaryExpression<E>
+ private static class NotExpression extends BooleanUnaryExpression
{
- public NotExpression(final BooleanExpression<E> left)
+ public NotExpression(final BooleanExpression left)
{
super(left);
}
- public Object evaluate(Filterable<E> message) throws E
+ public Object evaluate(Filterable message)
{
Boolean lvalue = (Boolean) right.evaluate(message);
if (lvalue == null)
@@ -333,14 +333,14 @@
}
}
- private static class BooleanCastExpression<E extends Exception> extends BooleanUnaryExpression<E>
+ private static class BooleanCastExpression extends BooleanUnaryExpression
{
- public BooleanCastExpression(final Expression<E> left)
+ public BooleanCastExpression(final Expression left)
{
super(left);
}
- public Object evaluate(Filterable<E> message) throws E
+ public Object evaluate(Filterable message)
{
Object rvalue = right.evaluate(message);
if (rvalue == null)
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/XPathExpression.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/XPathExpression.java?rev=829675&r1=829674&r2=829675&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/XPathExpression.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/XPathExpression.java Sun Oct 25 22:58:57 2009
@@ -22,7 +22,6 @@
import org.apache.log4j.Logger;
import org.apache.qpid.AMQException;
-import org.apache.qpid.server.queue.AMQMessage;
import org.apache.qpid.server.queue.Filterable;
import java.lang.reflect.Constructor;
@@ -71,7 +70,7 @@
private final XPathEvaluator evaluator;
static public interface XPathEvaluator {
- public boolean evaluate(Filterable message) throws AMQException;
+ public boolean evaluate(Filterable message);
}
XPathExpression(String xpath) {
@@ -93,7 +92,7 @@
}
}
- public Object evaluate(Filterable message) throws AMQException {
+ public Object evaluate(Filterable message) {
// try {
//FIXME this is flow to disk work
// if( message.isDropped() )
@@ -118,7 +117,7 @@
* @return true if the expression evaluates to Boolean.TRUE.
* @throws AMQException
*/
- public boolean matches(Filterable message) throws AMQException
+ public boolean matches(Filterable message)
{
Object object = evaluate(message);
return object!=null && object==Boolean.TRUE;
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/XQueryExpression.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/XQueryExpression.java?rev=829675&r1=829674&r2=829675&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/XQueryExpression.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/XQueryExpression.java Sun Oct 25 22:58:57 2009
@@ -18,7 +18,6 @@
package org.apache.qpid.server.filter;
import org.apache.qpid.AMQException;
-import org.apache.qpid.server.queue.AMQMessage;
import org.apache.qpid.server.queue.Filterable;
//
@@ -36,23 +35,23 @@
this.xpath = xpath;
}
- public Object evaluate(Filterable message) throws AMQException {
+ public Object evaluate(Filterable message) {
return Boolean.FALSE;
}
public String toString() {
return "XQUERY "+ConstantExpression.encodeString(xpath);
}
-
+
/**
* @param message
* @return true if the expression evaluates to Boolean.TRUE.
* @throws AMQException
*/
- public boolean matches(Filterable message) throws AMQException
+ public boolean matches(Filterable message)
{
Object object = evaluate(message);
- return object!=null && object==Boolean.TRUE;
+ return object!=null && object==Boolean.TRUE;
}
}
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/XalanXPathEvaluator.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/XalanXPathEvaluator.java?rev=829675&r1=829674&r2=829675&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/XalanXPathEvaluator.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/XalanXPathEvaluator.java Sun Oct 25 22:58:57 2009
@@ -27,8 +27,6 @@
import javax.xml.parsers.DocumentBuilder;
import javax.xml.parsers.DocumentBuilderFactory;
-import org.apache.qpid.AMQException;
-import org.apache.qpid.server.queue.AMQMessage;
import org.apache.qpid.server.queue.Filterable;
import org.apache.xpath.CachedXPathAPI;
import org.w3c.dom.Document;
@@ -36,14 +34,14 @@
import org.xml.sax.InputSource;
public class XalanXPathEvaluator implements XPathExpression.XPathEvaluator {
-
+
private final String xpath;
public XalanXPathEvaluator(String xpath) {
this.xpath = xpath;
}
-
- public boolean evaluate(Filterable m) throws AMQException
+
+ public boolean evaluate(Filterable m)
{
// TODO - we would have to check the content type and then evaluate the content
// here... is this really a feature we wish to implement? - RobG
@@ -65,18 +63,18 @@
private boolean evaluate(byte[] data) {
try {
-
+
InputSource inputSource = new InputSource(new ByteArrayInputStream(data));
-
+
DocumentBuilderFactory factory = DocumentBuilderFactory.newInstance();
factory.setNamespaceAware(true);
DocumentBuilder dbuilder = factory.newDocumentBuilder();
Document doc = dbuilder.parse(inputSource);
-
+
CachedXPathAPI cachedXPathAPI = new CachedXPathAPI();
NodeIterator iterator = cachedXPathAPI.selectNodeIterator(doc,xpath);
return iterator.nextNode()!=null;
-
+
} catch (Throwable e) {
return false;
}
@@ -85,12 +83,12 @@
private boolean evaluate(String text) {
try {
InputSource inputSource = new InputSource(new StringReader(text));
-
+
DocumentBuilderFactory factory = DocumentBuilderFactory.newInstance();
factory.setNamespaceAware(true);
DocumentBuilder dbuilder = factory.newDocumentBuilder();
Document doc = dbuilder.parse(inputSource);
-
+
// We should associated the cachedXPathAPI object with the message being evaluated
// since that should speedup subsequent xpath expressions.
CachedXPathAPI cachedXPathAPI = new CachedXPathAPI();
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/flow/AbstractFlowCreditManager.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/flow/AbstractFlowCreditManager.java?rev=829675&r1=829674&r2=829675&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/flow/AbstractFlowCreditManager.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/flow/AbstractFlowCreditManager.java Sun Oct 25 22:58:57 2009
@@ -31,19 +31,28 @@
public final void addStateListener(FlowCreditManagerListener listener)
{
- _listeners.add(listener);
+ synchronized(_listeners)
+ {
+ _listeners.add(listener);
+ }
}
public final boolean removeListener(FlowCreditManagerListener listener)
{
- return _listeners.remove(listener);
+ synchronized(_listeners)
+ {
+ return _listeners.remove(listener);
+ }
}
private void notifyListeners(final boolean suspended)
{
- for(FlowCreditManagerListener listener : _listeners)
+ synchronized(_listeners)
{
- listener.creditStateChanged(!suspended);
+ for(FlowCreditManagerListener listener : _listeners)
+ {
+ listener.creditStateChanged(!suspended);
+ }
}
}
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/flow/BytesOnlyCreditManager.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/flow/BytesOnlyCreditManager.java?rev=829675&r1=829674&r2=829675&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/flow/BytesOnlyCreditManager.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/flow/BytesOnlyCreditManager.java Sun Oct 25 22:58:57 2009
@@ -1,11 +1,8 @@
package org.apache.qpid.server.flow;
-import org.apache.qpid.server.queue.AMQMessage;
+import org.apache.qpid.server.message.ServerMessage;
import java.util.concurrent.atomic.AtomicLong;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.Set;
-import java.util.HashSet;
/*
*
@@ -36,7 +33,17 @@
_bytesCredit = new AtomicLong(initialCredit);
}
- public void addCredit(long messageCredit, long bytesCredit)
+ public long getMessageCredit()
+ {
+ return -1L;
+ }
+
+ public long getBytesCredit()
+ {
+ return _bytesCredit.get();
+ }
+
+ public void restoreCredit(long messageCredit, long bytesCredit)
{
_bytesCredit.addAndGet(bytesCredit);
setSuspended(false);
@@ -52,7 +59,7 @@
return _bytesCredit.get() > 0L;
}
- public boolean useCreditForMessage(AMQMessage msg)
+ public boolean useCreditForMessage(ServerMessage msg)
{
final long msgSize = msg.getSize();
if(hasCredit())
@@ -74,4 +81,9 @@
}
}
+
+ public void setBytesCredit(long bytesCredit)
+ {
+ _bytesCredit.set( bytesCredit );
+ }
}
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/flow/FlowCreditManager.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/flow/FlowCreditManager.java?rev=829675&r1=829674&r2=829675&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/flow/FlowCreditManager.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/flow/FlowCreditManager.java Sun Oct 25 22:58:57 2009
@@ -1,6 +1,6 @@
package org.apache.qpid.server.flow;
-import org.apache.qpid.server.queue.AMQMessage;
+import org.apache.qpid.server.message.ServerMessage;
/*
*
@@ -24,6 +24,9 @@
*/
public interface FlowCreditManager
{
+ long getMessageCredit();
+
+ long getBytesCredit();
public static interface FlowCreditManagerListener
{
@@ -34,11 +37,10 @@
boolean removeListener(FlowCreditManagerListener listener);
- public void addCredit(long messageCredit, long bytesCredit);
-
- public void removeAllCredit();
+ public void restoreCredit(long messageCredit, long bytesCredit);
public boolean hasCredit();
- public boolean useCreditForMessage(AMQMessage msg);
+ public boolean useCreditForMessage(ServerMessage msg);
+
}
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/flow/LimitlessCreditManager.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/flow/LimitlessCreditManager.java?rev=829675&r1=829674&r2=829675&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/flow/LimitlessCreditManager.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/flow/LimitlessCreditManager.java Sun Oct 25 22:58:57 2009
@@ -1,6 +1,6 @@
package org.apache.qpid.server.flow;
-import org.apache.qpid.server.queue.AMQMessage;
+import org.apache.qpid.server.message.ServerMessage;
/*
*
@@ -24,7 +24,17 @@
*/
public class LimitlessCreditManager extends AbstractFlowCreditManager implements FlowCreditManager
{
- public void addCredit(long messageCredit, long bytesCredit)
+ public long getMessageCredit()
+ {
+ return -1L;
+ }
+
+ public long getBytesCredit()
+ {
+ return -1L;
+ }
+
+ public void restoreCredit(long messageCredit, long bytesCredit)
{
}
@@ -37,7 +47,7 @@
return true;
}
- public boolean useCreditForMessage(AMQMessage msg)
+ public boolean useCreditForMessage(ServerMessage msg)
{
return true;
}
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/flow/MessageAndBytesCreditManager.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/flow/MessageAndBytesCreditManager.java?rev=829675&r1=829674&r2=829675&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/flow/MessageAndBytesCreditManager.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/flow/MessageAndBytesCreditManager.java Sun Oct 25 22:58:57 2009
@@ -1,8 +1,6 @@
package org.apache.qpid.server.flow;
-import org.apache.qpid.server.queue.AMQMessage;
-
-import java.util.concurrent.atomic.AtomicLong;
+import org.apache.qpid.server.message.ServerMessage;
/*
*
@@ -29,14 +27,24 @@
private long _messageCredit;
private long _bytesCredit;
- MessageAndBytesCreditManager(final long messageCredit, final long bytesCredit)
+ public MessageAndBytesCreditManager(final long messageCredit, final long bytesCredit)
{
_messageCredit = messageCredit;
_bytesCredit = bytesCredit;
}
- public synchronized void addCredit(long messageCredit, long bytesCredit)
+ public synchronized long getMessageCredit()
{
+ return _messageCredit;
+ }
+
+ public synchronized long getBytesCredit()
+ {
+ return _bytesCredit;
+ }
+
+ public synchronized void restoreCredit(long messageCredit, long bytesCredit)
+ {
_messageCredit += messageCredit;
_bytesCredit += bytesCredit;
setSuspended(hasCredit());
@@ -54,7 +62,7 @@
return (_messageCredit > 0L) && ( _bytesCredit > 0L );
}
- public synchronized boolean useCreditForMessage(AMQMessage msg)
+ public synchronized boolean useCreditForMessage(ServerMessage msg)
{
if(_messageCredit == 0L)
{
@@ -76,4 +84,9 @@
}
}
+
+ public synchronized void setBytesCredit(long bytesCredit)
+ {
+ _bytesCredit = bytesCredit;
+ }
}
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/flow/MessageOnlyCreditManager.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/flow/MessageOnlyCreditManager.java?rev=829675&r1=829674&r2=829675&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/flow/MessageOnlyCreditManager.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/flow/MessageOnlyCreditManager.java Sun Oct 25 22:58:57 2009
@@ -1,6 +1,6 @@
package org.apache.qpid.server.flow;
-import org.apache.qpid.server.queue.AMQMessage;
+import org.apache.qpid.server.message.ServerMessage;
import java.util.concurrent.atomic.AtomicLong;
@@ -33,10 +33,21 @@
_messageCredit = new AtomicLong(initialCredit);
}
- public void addCredit(long messageCredit, long bytesCredit)
+ public long getMessageCredit()
+ {
+ return _messageCredit.get();
+ }
+
+ public long getBytesCredit()
+ {
+ return -1L;
+ }
+
+ public void restoreCredit(long messageCredit, long bytesCredit)
{
- setSuspended(false);
_messageCredit.addAndGet(messageCredit);
+ setSuspended(false);
+
}
public void removeAllCredit()
@@ -50,7 +61,7 @@
return _messageCredit.get() > 0L;
}
- public boolean useCreditForMessage(AMQMessage msg)
+ public boolean useCreditForMessage(ServerMessage msg)
{
if(hasCredit())
{
@@ -73,4 +84,5 @@
}
}
+
}
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/flow/Pre0_10CreditManager.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/flow/Pre0_10CreditManager.java?rev=829675&r1=829674&r2=829675&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/flow/Pre0_10CreditManager.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/flow/Pre0_10CreditManager.java Sun Oct 25 22:58:57 2009
@@ -20,7 +20,7 @@
*/
package org.apache.qpid.server.flow;
-import org.apache.qpid.server.queue.AMQMessage;
+import org.apache.qpid.server.message.ServerMessage;
public class Pre0_10CreditManager extends AbstractFlowCreditManager implements FlowCreditManager
{
@@ -81,7 +81,17 @@
}
- public synchronized void addCredit(final long messageCredit, final long bytesCredit)
+ public long getMessageCredit()
+ {
+ return _messageCredit;
+ }
+
+ public long getBytesCredit()
+ {
+ return _bytesCredit;
+ }
+
+ public synchronized void restoreCredit(final long messageCredit, final long bytesCredit)
{
final long messageCreditLimit = _messageCreditLimit;
boolean notifyIncrease = true;
@@ -123,7 +133,7 @@
&& (_messageCreditLimit == 0L || _messageCredit > 0);
}
- public synchronized boolean useCreditForMessage(final AMQMessage msg)
+ public synchronized boolean useCreditForMessage(final ServerMessage msg)
{
if(_messageCreditLimit != 0L)
{
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicConsumeMethodHandler.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicConsumeMethodHandler.java?rev=829675&r1=829674&r2=829675&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicConsumeMethodHandler.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicConsumeMethodHandler.java Sun Oct 25 22:58:57 2009
@@ -25,7 +25,6 @@
import org.apache.qpid.framing.*;
import org.apache.qpid.protocol.AMQConstant;
import org.apache.qpid.server.AMQChannel;
-import org.apache.qpid.server.ConsumerTagNotUniqueException;
import org.apache.qpid.server.protocol.AMQProtocolSession;
import org.apache.qpid.server.queue.AMQQueue;
import org.apache.qpid.server.security.access.Permission;
@@ -116,17 +115,31 @@
try
{
- AMQShortString consumerTag = channel.subscribeToQueue(consumerTagName, queue, !body.getNoAck(),
- body.getArguments(), body.getNoLocal(), body.getExclusive());
- if (!body.getNowait())
+ if(consumerTagName == null || channel.getSubscription(consumerTagName) == null)
{
- MethodRegistry methodRegistry = session.getMethodRegistry();
- AMQMethodBody responseBody = methodRegistry.createBasicConsumeOkBody(consumerTag);
- session.writeFrame(responseBody.generateFrame(channelId));
+ AMQShortString consumerTag = channel.subscribeToQueue(consumerTagName, queue, !body.getNoAck(),
+ body.getArguments(), body.getNoLocal(), body.getExclusive());
+ if (!body.getNowait())
+ {
+ MethodRegistry methodRegistry = session.getMethodRegistry();
+ AMQMethodBody responseBody = methodRegistry.createBasicConsumeOkBody(consumerTag);
+ session.writeFrame(responseBody.generateFrame(channelId));
+
+ }
+ }
+ else
+ {
+ AMQShortString msg = new AMQShortString("Non-unique consumer tag, '" + body.getConsumerTag() + "'");
+
+ MethodRegistry methodRegistry = session.getMethodRegistry();
+ AMQMethodBody responseBody = methodRegistry.createConnectionCloseBody(AMQConstant.NOT_ALLOWED.getCode(), // replyCode
+ msg, // replytext
+ body.getClazz(),
+ body.getMethod());
+ session.writeFrame(responseBody.generateFrame(0));
}
-
}
catch (org.apache.qpid.AMQInvalidArgumentException ise)
{
@@ -141,17 +154,6 @@
}
- catch (ConsumerTagNotUniqueException e)
- {
- AMQShortString msg = new AMQShortString("Non-unique consumer tag, '" + body.getConsumerTag() + "'");
-
- MethodRegistry methodRegistry = session.getMethodRegistry();
- AMQMethodBody responseBody = methodRegistry.createConnectionCloseBody(AMQConstant.NOT_ALLOWED.getCode(), // replyCode
- msg, // replytext
- body.getClazz(),
- body.getMethod());
- session.writeFrame(responseBody.generateFrame(0));
- }
catch (AMQQueue.ExistingExclusiveSubscription e)
{
throw body.getChannelException(AMQConstant.ACCESS_REFUSED,
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicGetMethodHandler.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicGetMethodHandler.java?rev=829675&r1=829674&r2=829675&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicGetMethodHandler.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicGetMethodHandler.java Sun Oct 25 22:58:57 2009
@@ -30,6 +30,7 @@
import org.apache.qpid.framing.FieldTable;
import org.apache.qpid.protocol.AMQConstant;
import org.apache.qpid.server.AMQChannel;
+import org.apache.qpid.server.message.AMQMessage;
import org.apache.qpid.server.flow.FlowCreditManager;
import org.apache.qpid.server.flow.MessageOnlyCreditManager;
import org.apache.qpid.server.subscription.SubscriptionImpl;
@@ -40,8 +41,6 @@
import org.apache.qpid.server.protocol.AMQProtocolSession;
import org.apache.qpid.server.queue.AMQQueue;
import org.apache.qpid.server.queue.QueueEntry;
-import org.apache.qpid.server.queue.SimpleAMQQueue;
-import org.apache.qpid.server.security.access.Permission;
import org.apache.qpid.server.state.AMQStateManager;
import org.apache.qpid.server.state.StateAwareMethodListener;
import org.apache.qpid.server.virtualhost.VirtualHost;
@@ -130,8 +129,16 @@
throws AMQException
{
singleMessageCredit.useCreditForMessage(entry.getMessage());
- session.getProtocolOutputConverter().writeGetOk(entry.getMessage(), channel.getChannelId(),
- deliveryTag, queue.getMessageCount());
+ if(entry.getMessage() instanceof AMQMessage)
+ {
+ session.getProtocolOutputConverter().writeGetOk(entry, channel.getChannelId(),
+ deliveryTag, queue.getMessageCount());
+ }
+ else
+ {
+ //TODO Convert AMQP 0-10 message
+ throw new RuntimeException("Not implemented conversion of 0-10 message");
+ }
}
};
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org