QPID-6028 : allow binding destinations to be things other than queues
diff --git a/broker-core/src/main/java/org/apache/qpid/server/exchange/AbstractExchange.java b/broker-core/src/main/java/org/apache/qpid/server/exchange/AbstractExchange.java
index 8964bfe..5987d40 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/exchange/AbstractExchange.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/exchange/AbstractExchange.java
@@ -24,6 +24,7 @@
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
+import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
@@ -49,9 +50,8 @@
import org.apache.qpid.server.logging.subjects.ExchangeLogSubject;
import org.apache.qpid.server.message.InstanceProperties;
import org.apache.qpid.server.message.MessageDestination;
-import org.apache.qpid.server.message.MessageInstance;
-import org.apache.qpid.server.message.MessageReference;
import org.apache.qpid.server.message.MessageSender;
+import org.apache.qpid.server.message.RoutingResult;
import org.apache.qpid.server.message.ServerMessage;
import org.apache.qpid.server.model.AbstractConfiguredObject;
import org.apache.qpid.server.model.Binding;
@@ -66,13 +66,9 @@
import org.apache.qpid.server.model.Queue;
import org.apache.qpid.server.model.State;
import org.apache.qpid.server.model.StateTransition;
-import org.apache.qpid.server.queue.BaseQueue;
import org.apache.qpid.server.security.SecurityToken;
import org.apache.qpid.server.security.access.Operation;
-import org.apache.qpid.server.store.MessageEnqueueRecord;
import org.apache.qpid.server.store.StorableMessageMetaData;
-import org.apache.qpid.server.txn.ServerTransaction;
-import org.apache.qpid.server.util.Action;
import org.apache.qpid.server.util.FixedKeyMapCreator;
import org.apache.qpid.server.virtualhost.ExchangeIsAlternateException;
import org.apache.qpid.server.virtualhost.QueueManagingVirtualHost;
@@ -86,6 +82,8 @@
{
private static final Logger _logger = LoggerFactory.getLogger(AbstractExchange.class);
+ private static final ThreadLocal<Map<AbstractExchange<?>, Set<String>>> CURRENT_ROUTING = new ThreadLocal<>();
+
private static final FixedKeyMapCreator BIND_ARGUMENTS_CREATOR =
new FixedKeyMapCreator("bindingKey", "destination", "arguments");
private static final FixedKeyMapCreator UNBIND_ARGUMENTS_CREATOR =
@@ -469,131 +467,84 @@
}
- final List<? extends BaseQueue> route(final ServerMessage message,
- final String routingAddress,
- final InstanceProperties instanceProperties)
- {
- _receivedMessageCount.incrementAndGet();
- _receivedMessageSize.addAndGet(message.getSize());
- List<? extends BaseQueue> queues = doRoute(message, routingAddress, instanceProperties);
- List<? extends BaseQueue> allQueues = queues;
-
- boolean deletedQueues = false;
-
- for(BaseQueue q : allQueues)
- {
- if(q.isDeleted())
- {
- if(!deletedQueues)
- {
- deletedQueues = true;
- queues = new ArrayList<>(allQueues);
- }
- _logger.debug("Exchange: {} - attempt to enqueue message onto deleted queue {}", getName(), q.getName());
-
- queues.remove(q);
- }
- }
-
-
- if(!queues.isEmpty())
- {
- _routedMessageCount.incrementAndGet();
- _routedMessageSize.addAndGet(message.getSize());
- }
- else
- {
- _droppedMessageCount.incrementAndGet();
- _droppedMessageSize.addAndGet(message.getSize());
- }
- return queues;
- }
-
@Override
- public final <M extends ServerMessage<? extends StorableMessageMetaData>> int send(final M message,
- final String routingAddress,
- final InstanceProperties instanceProperties,
- final ServerTransaction txn,
- final Action<? super MessageInstance> postEnqueueAction)
+ public <M extends ServerMessage<? extends StorableMessageMetaData>> RoutingResult<M> route(final M message,
+ final String routingAddress,
+ final InstanceProperties instanceProperties)
{
if (_virtualHost.getState() != State.ACTIVE)
{
throw new VirtualHostUnavailableException(this._virtualHost);
}
- List<? extends BaseQueue> queues = route(message, routingAddress, instanceProperties);
+ final RoutingResult<M> routingResult = new RoutingResult<>(message);
- if(queues == null || queues.isEmpty())
+ Map<AbstractExchange<?>, Set<String>> currentThreadMap = CURRENT_ROUTING.get();
+ boolean topLevel = currentThreadMap == null;
+ try
{
- Exchange altExchange = getAlternateExchange();
- if(altExchange != null)
+ if (topLevel)
{
- return altExchange.send(message, routingAddress, instanceProperties, txn, postEnqueueAction);
+ currentThreadMap = new HashMap<>();
+ CURRENT_ROUTING.set(currentThreadMap);
+ }
+ Set<String> existingRoutes = currentThreadMap.get(this);
+ if (existingRoutes == null)
+ {
+ currentThreadMap.put(this, Collections.singleton(routingAddress));
+ }
+ else if (existingRoutes.contains(routingAddress))
+ {
+ return routingResult;
}
else
{
- return 0;
+ existingRoutes = new HashSet<>(existingRoutes);
+ existingRoutes.add(routingAddress);
+ currentThreadMap.put(this, existingRoutes);
}
+
+ _receivedMessageCount.incrementAndGet();
+ _receivedMessageSize.addAndGet(message.getSize());
+
+ doRoute(message, routingAddress, instanceProperties, routingResult);
+
+ if (!routingResult.hasRoutes())
+ {
+ Exchange altExchange = getAlternateExchange();
+ if (altExchange != null)
+ {
+ routingResult.add(altExchange.route(message, routingAddress, instanceProperties));
+ }
+ }
+
+ if (routingResult.hasRoutes())
+ {
+ _routedMessageCount.incrementAndGet();
+ _routedMessageSize.addAndGet(message.getSize());
+ }
+ else
+ {
+ _droppedMessageCount.incrementAndGet();
+ _droppedMessageSize.addAndGet(message.getSize());
+ }
+
+ return routingResult;
}
- else
+ finally
{
- for(BaseQueue q : queues)
+ if(topLevel)
{
- if(!message.isResourceAcceptable(q))
- {
- return 0;
- }
+ CURRENT_ROUTING.set(null);
}
- final BaseQueue[] baseQueues;
-
- if(message.isReferenced())
- {
- ArrayList<BaseQueue> uniqueQueues = new ArrayList<>(queues.size());
- for(BaseQueue q : queues)
- {
- if(!message.isReferenced(q))
- {
- uniqueQueues.add(q);
- }
- }
- baseQueues = uniqueQueues.toArray(new BaseQueue[uniqueQueues.size()]);
- }
- else
- {
- baseQueues = queues.toArray(new BaseQueue[queues.size()]);
- }
-
- txn.enqueue(queues,message, new ServerTransaction.EnqueueAction()
- {
- MessageReference _reference = message.newReference();
-
- public void postCommit(MessageEnqueueRecord... records)
- {
- try
- {
- for(int i = 0; i < baseQueues.length; i++)
- {
- baseQueues[i].enqueue(message, postEnqueueAction, records[i]);
- }
- }
- finally
- {
- _reference.release();
- }
- }
-
- public void onRollback()
- {
- _reference.release();
- }
- });
- return queues.size();
}
}
- protected abstract List<? extends BaseQueue> doRoute(final ServerMessage message,
- final String routingAddress,
- final InstanceProperties instanceProperties);
+
+ protected abstract <M extends ServerMessage<? extends StorableMessageMetaData>> void doRoute(final M message,
+ final String routingAddress,
+ final InstanceProperties instanceProperties,
+ final RoutingResult<M> result);
@Override
public boolean bind(final String destination,
diff --git a/broker-core/src/main/java/org/apache/qpid/server/exchange/DefaultDestination.java b/broker-core/src/main/java/org/apache/qpid/server/exchange/DefaultDestination.java
index 89afbce..2d8ac4e 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/exchange/DefaultDestination.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/exchange/DefaultDestination.java
@@ -24,8 +24,8 @@
import org.apache.qpid.exchange.ExchangeDefaults;
import org.apache.qpid.server.message.InstanceProperties;
import org.apache.qpid.server.message.MessageDestination;
-import org.apache.qpid.server.message.MessageInstance;
import org.apache.qpid.server.message.MessageSender;
+import org.apache.qpid.server.message.RoutingResult;
import org.apache.qpid.server.message.ServerMessage;
import org.apache.qpid.server.model.ConfiguredObject;
import org.apache.qpid.server.model.Exchange;
@@ -37,8 +37,6 @@
import org.apache.qpid.server.security.SecurityToken;
import org.apache.qpid.server.security.access.Operation;
import org.apache.qpid.server.store.StorableMessageMetaData;
-import org.apache.qpid.server.txn.ServerTransaction;
-import org.apache.qpid.server.util.Action;
import org.apache.qpid.server.virtualhost.QueueManagingVirtualHost;
public class DefaultDestination implements MessageDestination, PermissionedObject
@@ -94,43 +92,43 @@
}
- public final <M extends ServerMessage<? extends StorableMessageMetaData>> int send(final M message,
- String routingAddress,
- final InstanceProperties instanceProperties,
- final ServerTransaction txn,
- final Action<? super MessageInstance> postEnqueueAction)
+ @Override
+ public <M extends ServerMessage<? extends StorableMessageMetaData>> RoutingResult<M> route(M message,
+ String routingAddress,
+ InstanceProperties instanceProperties)
{
- if(routingAddress == null || routingAddress.trim().equals(""))
+ RoutingResult<M> result = new RoutingResult<>(message);
+
+ if (routingAddress != null && !routingAddress.trim().equals(""))
{
- return 0;
- }
- final MessageDestination dest = _virtualHost.getAttainedMessageDestination(routingAddress);
- if(dest == null)
- {
- routingAddress = _virtualHost.getLocalAddress(routingAddress);
- if(routingAddress.contains("/") && !routingAddress.startsWith("/"))
+ final MessageDestination dest = _virtualHost.getAttainedMessageDestination(routingAddress);
+ if (dest == null)
{
- String[] parts = routingAddress.split("/",2);
- Exchange<?> exchange = _virtualHost.getAttainedChildFromAddress(Exchange.class, parts[0]);
- if(exchange != null)
+ routingAddress = _virtualHost.getLocalAddress(routingAddress);
+ if (routingAddress.contains("/") && !routingAddress.startsWith("/"))
{
- return exchange.send(message, parts[1], instanceProperties, txn, postEnqueueAction);
+ String[] parts = routingAddress.split("/", 2);
+ Exchange<?> exchange = _virtualHost.getAttainedChildFromAddress(Exchange.class, parts[0]);
+ if (exchange != null)
+ {
+ result.add(exchange.route(message, parts[1], instanceProperties));
+ }
+ }
+ else if (!routingAddress.contains("/"))
+ {
+ Exchange<?> exchange = _virtualHost.getAttainedChildFromAddress(Exchange.class, routingAddress);
+ if (exchange != null)
+ {
+ result.add(exchange.route(message, "", instanceProperties));
+ }
}
}
- else if(!routingAddress.contains("/"))
+ else
{
- Exchange<?> exchange = _virtualHost.getAttainedChildFromAddress(Exchange.class, routingAddress);
- if(exchange != null)
- {
- return exchange.send(message, "", instanceProperties, txn, postEnqueueAction);
- }
+ result.add(dest.route(message, routingAddress, instanceProperties));
}
- return 0;
}
- else
- {
- return dest.send(message, routingAddress, instanceProperties, txn, postEnqueueAction);
- }
+ return result;
}
@Override
diff --git a/broker-core/src/main/java/org/apache/qpid/server/exchange/DirectExchangeImpl.java b/broker-core/src/main/java/org/apache/qpid/server/exchange/DirectExchangeImpl.java
index b5a24dd..6c4be90 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/exchange/DirectExchangeImpl.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/exchange/DirectExchangeImpl.java
@@ -20,11 +20,9 @@
*/
package org.apache.qpid.server.exchange;
-import java.util.ArrayList;
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.ConcurrentHashMap;
@@ -39,10 +37,11 @@
import org.apache.qpid.server.filter.Filterable;
import org.apache.qpid.server.message.InstanceProperties;
import org.apache.qpid.server.message.MessageDestination;
+import org.apache.qpid.server.message.RoutingResult;
import org.apache.qpid.server.message.ServerMessage;
import org.apache.qpid.server.model.ManagedObjectFactoryConstructor;
import org.apache.qpid.server.model.Queue;
-import org.apache.qpid.server.queue.BaseQueue;
+import org.apache.qpid.server.store.StorableMessageMetaData;
import org.apache.qpid.server.virtualhost.QueueManagingVirtualHost;
public class DirectExchangeImpl extends AbstractExchange<DirectExchangeImpl> implements DirectExchange<DirectExchangeImpl>
@@ -52,8 +51,8 @@
private final class BindingSet
{
- private final Set<BaseQueue> _unfilteredQueues;
- private final Map<BaseQueue, FilterManager> _filteredQueues;
+ private final Set<MessageDestination> _unfilteredQueues;
+ private final Map<MessageDestination, FilterManager> _filteredQueues;
public BindingSet()
{
@@ -61,14 +60,14 @@
_filteredQueues = Collections.emptyMap();
}
- private BindingSet(final Set<BaseQueue> unfilteredQueues,
- final Map<BaseQueue, FilterManager> filteredQueues)
+ private BindingSet(final Set<MessageDestination> unfilteredQueues,
+ final Map<MessageDestination, FilterManager> filteredQueues)
{
_unfilteredQueues = unfilteredQueues;
_filteredQueues = filteredQueues;
}
- public Set<BaseQueue> getUnfilteredQueues()
+ public Set<MessageDestination> getUnfilteredQueues()
{
return _unfilteredQueues;
}
@@ -83,7 +82,7 @@
return _unfilteredQueues.isEmpty() && _filteredQueues.isEmpty();
}
- public Map<BaseQueue,FilterManager> getFilteredQueues()
+ public Map<MessageDestination,FilterManager> getFilteredQueues()
{
return _filteredQueues;
}
@@ -98,8 +97,8 @@
{
try
{
- Set<BaseQueue> unfilteredQueues;
- Map<BaseQueue, FilterManager> filteredQueues;
+ Set<MessageDestination> unfilteredQueues;
+ Map<MessageDestination, FilterManager> filteredQueues;
if (_unfilteredQueues.contains(destination))
{
unfilteredQueues = new HashSet<>(_unfilteredQueues);
@@ -111,7 +110,7 @@
}
filteredQueues = new HashMap<>(_filteredQueues);
- filteredQueues.put((BaseQueue) destination,
+ filteredQueues.put(destination,
FilterSupport.createMessageFilter(arguments, (Queue<?>) destination));
return new BindingSet(Collections.unmodifiableSet(unfilteredQueues), Collections.unmodifiableMap(filteredQueues));
@@ -128,8 +127,8 @@
}
else
{
- Set<BaseQueue> unfilteredQueues;
- Map<BaseQueue, FilterManager> filteredQueues;
+ Set<MessageDestination> unfilteredQueues;
+ Map<MessageDestination, FilterManager> filteredQueues;
if (_filteredQueues.containsKey(destination))
{
filteredQueues = new HashMap<>(_filteredQueues);
@@ -141,7 +140,7 @@
}
unfilteredQueues = new HashSet<>(_unfilteredQueues);
- unfilteredQueues.add((BaseQueue)destination);
+ unfilteredQueues.add(destination);
return new BindingSet(Collections.unmodifiableSet(unfilteredQueues), Collections.unmodifiableMap(filteredQueues));
@@ -150,8 +149,8 @@
public BindingSet removeBinding(final MessageDestination destination)
{
- Set<BaseQueue> unfilteredQueues;
- Map<BaseQueue, FilterManager> filteredQueues;
+ Set<MessageDestination> unfilteredQueues;
+ Map<MessageDestination, FilterManager> filteredQueues;
if (_unfilteredQueues.contains(destination))
{
unfilteredQueues = new HashSet<>(_unfilteredQueues);
@@ -182,48 +181,42 @@
super(attributes, vhost);
}
+
@Override
- public List<? extends BaseQueue> doRoute(ServerMessage payload,
- final String routingKey,
- final InstanceProperties instanceProperties)
+ public <M extends ServerMessage<? extends StorableMessageMetaData>> void doRoute(final M payload,
+ final String routingKey,
+ final InstanceProperties instanceProperties,
+ final RoutingResult<M> result)
{
BindingSet bindings = _bindingsByKey.get(routingKey == null ? "" : routingKey);
if(bindings != null)
{
- // TODO - remove this garbage generation
- List<BaseQueue> queues = new ArrayList<>(bindings.getUnfilteredQueues());
+ final Set<MessageDestination> unfilteredQueues = bindings.getUnfilteredQueues();
+ for(MessageDestination destination : unfilteredQueues)
+ {
+ result.add(destination.route(payload, routingKey, instanceProperties));
+ }
if(bindings.hasFilteredQueues())
{
- Set<BaseQueue> queuesSet = new HashSet<BaseQueue>(queues);
Filterable filterable = Filterable.Factory.newInstance(payload, instanceProperties);
- Map<BaseQueue, FilterManager> filteredQueues = bindings.getFilteredQueues();
- for(Map.Entry<BaseQueue, FilterManager> entry : filteredQueues.entrySet())
+ Map<MessageDestination, FilterManager> filteredQueues = bindings.getFilteredQueues();
+ for(Map.Entry<MessageDestination, FilterManager> entry : filteredQueues.entrySet())
{
- if(!queuesSet.contains(entry.getKey()))
+ if(!unfilteredQueues.contains(entry.getKey()))
{
FilterManager filter = entry.getValue();
if(filter.allAllow(filterable))
{
- queuesSet.add(entry.getKey());
+ result.add(entry.getKey().route(payload, routingKey, instanceProperties));
}
}
}
- if(queues.size() != queuesSet.size())
- {
- queues = new ArrayList<>(queuesSet);
- }
}
- return queues;
}
- else
- {
- return Collections.emptyList();
- }
-
}
diff --git a/broker-core/src/main/java/org/apache/qpid/server/exchange/FanoutExchangeImpl.java b/broker-core/src/main/java/org/apache/qpid/server/exchange/FanoutExchangeImpl.java
index c4b103b..d6864d4 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/exchange/FanoutExchangeImpl.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/exchange/FanoutExchangeImpl.java
@@ -35,10 +35,11 @@
import org.apache.qpid.server.filter.Filterable;
import org.apache.qpid.server.message.InstanceProperties;
import org.apache.qpid.server.message.MessageDestination;
+import org.apache.qpid.server.message.RoutingResult;
import org.apache.qpid.server.message.ServerMessage;
import org.apache.qpid.server.model.ManagedObjectFactoryConstructor;
import org.apache.qpid.server.model.Queue;
-import org.apache.qpid.server.queue.BaseQueue;
+import org.apache.qpid.server.store.StorableMessageMetaData;
import org.apache.qpid.server.virtualhost.QueueManagingVirtualHost;
class FanoutExchangeImpl extends AbstractExchange<FanoutExchangeImpl> implements FanoutExchange<FanoutExchangeImpl>
@@ -51,15 +52,15 @@
{
private final Map<MessageDestination,Integer> _queues;
- private final List<Queue<?>> _unfilteredQueues;
- private final List<Queue<?>> _filteredQueues;
+ private final List<MessageDestination> _unfilteredQueues;
+ private final List<MessageDestination> _filteredQueues;
- private final Map<Queue<?>,Map<BindingIdentifier, FilterManager>> _filteredBindings;
+ private final Map<MessageDestination,Map<BindingIdentifier, FilterManager>> _filteredBindings;
public BindingSet(final Map<MessageDestination, Integer> queues,
- final List<Queue<?>> unfilteredQueues,
- final List<Queue<?>> filteredQueues,
- final Map<Queue<?>, Map<BindingIdentifier, FilterManager>> filteredBindings)
+ final List<MessageDestination> unfilteredQueues,
+ final List<MessageDestination> filteredQueues,
+ final Map<MessageDestination, Map<BindingIdentifier, FilterManager>> filteredBindings)
{
_queues = queues;
_unfilteredQueues = unfilteredQueues;
@@ -81,19 +82,19 @@
{
try
{
- List<Queue<?>> filteredQueues;
+ List<MessageDestination> filteredQueues;
if (!(_filteredQueues.contains(binding.getDestination())
|| _unfilteredQueues.contains(binding.getDestination())))
{
filteredQueues = new ArrayList<>(_filteredQueues);
- filteredQueues.add((Queue<?>) binding.getDestination());
+ filteredQueues.add(binding.getDestination());
filteredQueues = Collections.unmodifiableList(filteredQueues);
}
else
{
filteredQueues = _filteredQueues;
}
- Map<Queue<?>, Map<BindingIdentifier, FilterManager>> filteredBindings =
+ Map<MessageDestination, Map<BindingIdentifier, FilterManager>> filteredBindings =
new HashMap<>(_filteredBindings);
Map<BindingIdentifier, FilterManager> bindingsForQueue =
filteredBindings.get(binding.getDestination());
@@ -107,8 +108,8 @@
}
bindingsForQueue.put(binding,
FilterSupport.createMessageFilter(arguments,
- (Queue<?>) binding.getDestination()));
- filteredBindings.put((Queue<?>) binding.getDestination(), bindingsForQueue);
+ binding.getDestination()));
+ filteredBindings.put(binding.getDestination(), bindingsForQueue);
return new BindingSet(_queues, _unfilteredQueues, filteredQueues, Collections.unmodifiableMap(filteredBindings));
}
catch (AMQInvalidArgumentException e)
@@ -122,8 +123,8 @@
else
{
Map<MessageDestination, Integer> queues = new HashMap<>(_queues);
- List<Queue<?>> unfilteredQueues;
- List<Queue<?>> filteredQueues;
+ List<MessageDestination> unfilteredQueues;
+ List<MessageDestination> filteredQueues;
if (queues.containsKey(binding.getDestination()))
{
queues.put(binding.getDestination(), queues.get(binding.getDestination()) + 1);
@@ -162,10 +163,10 @@
Queue<?> queue = (Queue<?>) binding.getDestination();
if(_filteredBindings.containsKey(queue) && _filteredBindings.get(queue).containsKey(binding))
{
- final Map<Queue<?>, Map<BindingIdentifier, FilterManager>> filteredBindings = new HashMap<>(_filteredBindings);
+ final Map<MessageDestination, Map<BindingIdentifier, FilterManager>> filteredBindings = new HashMap<>(_filteredBindings);
final Map<BindingIdentifier, FilterManager> bindingsForQueue = new HashMap<>(filteredBindings.remove(queue));
bindingsForQueue.remove(binding);
- List<Queue<?>> filteredQueues;
+ List<MessageDestination> filteredQueues;
if(bindingsForQueue.isEmpty())
{
filteredQueues = new ArrayList<>(_filteredQueues);
@@ -183,8 +184,8 @@
{
Map<MessageDestination, Integer> queues = new HashMap<>(_queues);
int count = queues.remove(queue);
- List<Queue<?>> unfilteredQueues;
- List<Queue<?>> filteredQueues;
+ List<MessageDestination> unfilteredQueues;
+ List<MessageDestination> filteredQueues;
if(count > 1)
{
queues.put(queue, --count);
@@ -230,28 +231,29 @@
}
@Override
- public ArrayList<BaseQueue> doRoute(ServerMessage payload,
- final String routingKey,
- final InstanceProperties instanceProperties)
+ protected <M extends ServerMessage<? extends StorableMessageMetaData>> void doRoute(final M message,
+ final String routingAddress,
+ final InstanceProperties instanceProperties,
+ final RoutingResult<M> result)
{
-
BindingSet bindingSet = _bindingSet;
- final ArrayList<BaseQueue> result = new ArrayList<BaseQueue>(bindingSet._unfilteredQueues);
-
-
- final Map<Queue<?>, Map<BindingIdentifier, FilterManager>> filteredBindings = bindingSet._filteredBindings;
+ for(MessageDestination destination : bindingSet._unfilteredQueues)
+ {
+ result.add(destination.route(message, routingAddress, instanceProperties));
+ }
+ final Map<MessageDestination, Map<BindingIdentifier, FilterManager>> filteredBindings = bindingSet._filteredBindings;
if(!bindingSet._filteredQueues.isEmpty())
{
- for(Queue<?> q : bindingSet._filteredQueues)
+ for(MessageDestination q : bindingSet._filteredQueues)
{
final Map<BindingIdentifier, FilterManager> bindingMessageFilterMap = filteredBindings.get(q);
- if(!(bindingMessageFilterMap == null || result.contains(q)))
+ if(!(bindingMessageFilterMap == null || bindingSet._unfilteredQueues.contains(q)))
{
for(FilterManager filter : bindingMessageFilterMap.values())
{
- if(filter.allAllow(Filterable.Factory.newInstance(payload, instanceProperties)))
+ if(filter.allAllow(Filterable.Factory.newInstance(message, instanceProperties)))
{
- result.add(q);
+ result.add(q.route(message, routingAddress, instanceProperties));
break;
}
}
@@ -261,10 +263,6 @@
}
- _logger.debug("Publishing message to queue {}", result);
-
- return result;
-
}
@Override
diff --git a/broker-core/src/main/java/org/apache/qpid/server/exchange/HeadersExchangeImpl.java b/broker-core/src/main/java/org/apache/qpid/server/exchange/HeadersExchangeImpl.java
index 1d0694b..24be373 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/exchange/HeadersExchangeImpl.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/exchange/HeadersExchangeImpl.java
@@ -20,9 +20,7 @@
*/
package org.apache.qpid.server.exchange;
-import java.util.ArrayList;
import java.util.Collections;
-import java.util.LinkedHashSet;
import java.util.ListIterator;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
@@ -35,10 +33,11 @@
import org.apache.qpid.server.filter.Filterable;
import org.apache.qpid.server.message.InstanceProperties;
+import org.apache.qpid.server.message.RoutingResult;
import org.apache.qpid.server.message.ServerMessage;
import org.apache.qpid.server.model.ManagedObjectFactoryConstructor;
import org.apache.qpid.server.model.Queue;
-import org.apache.qpid.server.queue.BaseQueue;
+import org.apache.qpid.server.store.StorableMessageMetaData;
import org.apache.qpid.server.virtualhost.QueueManagingVirtualHost;
/**
@@ -86,14 +85,13 @@
}
@Override
- public ArrayList<BaseQueue> doRoute(ServerMessage payload,
- final String routingKey,
- final InstanceProperties instanceProperties)
+ public <M extends ServerMessage<? extends StorableMessageMetaData>> void doRoute(M payload,
+ final String routingKey,
+ final InstanceProperties instanceProperties,
+ RoutingResult<M> routingResult)
{
_logger.debug("Exchange {}: routing message with headers {}", getName(), payload.getMessageHeader());
- LinkedHashSet<BaseQueue> queues = new LinkedHashSet<BaseQueue>();
-
for (HeadersBinding hb : _bindingHeaderMatchers)
{
if (hb.matches(Filterable.Factory.newInstance(payload,instanceProperties)))
@@ -106,13 +104,12 @@
_logger.debug("Exchange " + getName() + ": delivering message with headers " +
payload.getMessageHeader() + " to " + b.getDestination().getName());
}
- queues.add((BaseQueue) b.getDestination());
+ routingResult.add(b.getDestination().route(payload, routingKey, instanceProperties));
}
}
-
- return new ArrayList<>(queues);
}
+
@Override
protected void onBind(final BindingIdentifier binding, Map<String,Object> arguments)
{
diff --git a/broker-core/src/main/java/org/apache/qpid/server/exchange/TopicExchangeImpl.java b/broker-core/src/main/java/org/apache/qpid/server/exchange/TopicExchangeImpl.java
index 1056085..4a30d8e 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/exchange/TopicExchangeImpl.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/exchange/TopicExchangeImpl.java
@@ -20,7 +20,6 @@
*/
package org.apache.qpid.server.exchange;
-import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
@@ -39,11 +38,12 @@
import org.apache.qpid.server.filter.FilterSupport;
import org.apache.qpid.server.filter.Filterable;
import org.apache.qpid.server.message.InstanceProperties;
+import org.apache.qpid.server.message.MessageDestination;
+import org.apache.qpid.server.message.RoutingResult;
import org.apache.qpid.server.message.ServerMessage;
-import org.apache.qpid.server.model.Binding;
import org.apache.qpid.server.model.ManagedObjectFactoryConstructor;
import org.apache.qpid.server.model.Queue;
-import org.apache.qpid.server.queue.BaseQueue;
+import org.apache.qpid.server.store.StorableMessageMetaData;
import org.apache.qpid.server.util.ConnectionScopedRuntimeException;
import org.apache.qpid.server.virtualhost.QueueManagingVirtualHost;
@@ -68,9 +68,9 @@
protected synchronized void onBindingUpdated(final BindingIdentifier binding, final Map<String, Object> newArguments)
{
final String bindingKey = binding.getBindingKey();
- Queue<?> queue = (Queue<?>) binding.getDestination();
+ final MessageDestination destination = binding.getDestination();
- _logger.debug("Updating binding of queue {} with routing key {}", queue.getName(), bindingKey);
+ _logger.debug("Updating binding of queue {} with routing key {}", destination.getName(), bindingKey);
String routingKey = TopicNormalizer.normalize(bindingKey);
@@ -87,22 +87,22 @@
{
if (FilterSupport.argumentsContainFilter(oldArgs))
{
- result.replaceQueueFilter(queue,
- FilterSupport.createMessageFilter(oldArgs, queue),
- FilterSupport.createMessageFilter(newArguments, queue));
+ result.replaceQueueFilter(destination,
+ FilterSupport.createMessageFilter(oldArgs, destination),
+ FilterSupport.createMessageFilter(newArguments, destination));
}
else
{
- result.addFilteredQueue(queue, FilterSupport.createMessageFilter(newArguments, queue));
- result.removeUnfilteredQueue(queue);
+ result.addFilteredQueue(destination, FilterSupport.createMessageFilter(newArguments, destination));
+ result.removeUnfilteredQueue(destination);
}
}
else
{
if (FilterSupport.argumentsContainFilter(oldArgs))
{
- result.addUnfilteredQueue(queue);
- result.removeFilteredQueue(queue, FilterSupport.createMessageFilter(oldArgs, queue));
+ result.addUnfilteredQueue(destination);
+ result.removeFilteredQueue(destination, FilterSupport.createMessageFilter(oldArgs, destination));
}
else
{
@@ -204,39 +204,25 @@
}
@Override
- public ArrayList<BaseQueue> doRoute(ServerMessage payload,
- final String routingAddress,
- final InstanceProperties instanceProperties)
+ public <M extends ServerMessage<? extends StorableMessageMetaData>> void doRoute(M payload,
+ String routingAddress,
+ InstanceProperties instanceProperties,
+ RoutingResult<M> result)
{
-
final String routingKey = routingAddress == null
- ? ""
- : routingAddress;
+ ? ""
+ : routingAddress;
- final Collection<Queue<?>> matchedQueues =
+ final Collection<MessageDestination> matchedQueues =
getMatchedQueues(Filterable.Factory.newInstance(payload,instanceProperties), routingKey);
- ArrayList<BaseQueue> queues;
-
- if(matchedQueues.getClass() == ArrayList.class)
+ for(MessageDestination queue : matchedQueues)
{
- queues = (ArrayList) matchedQueues;
+ result.add(queue.route(payload, routingAddress, instanceProperties));
}
- else
- {
- queues = new ArrayList<BaseQueue>();
- queues.addAll(matchedQueues);
- }
-
- if(queues == null || queues.isEmpty())
- {
- _logger.info("Message routing key: " + routingAddress + " No routes.");
- }
-
- return queues;
-
}
+
private synchronized boolean deregisterQueue(final BindingIdentifier binding)
{
if(_bindings.containsKey(binding))
@@ -274,7 +260,7 @@
}
}
- private Collection<Queue<?>> getMatchedQueues(Filterable message, String routingKey)
+ private Collection<MessageDestination> getMatchedQueues(Filterable message, String routingKey)
{
Collection<TopicMatcherResult> results = _parser.parse(routingKey);
@@ -287,7 +273,7 @@
results.toArray(resultQueues);
return ((TopicExchangeResult)resultQueues[0]).processMessage(message, null);
default:
- Collection<Queue<?>> queues = new HashSet<>();
+ Collection<MessageDestination> queues = new HashSet<>();
for(TopicMatcherResult result : results)
{
TopicExchangeResult res = (TopicExchangeResult)result;
diff --git a/broker-core/src/main/java/org/apache/qpid/server/exchange/topic/TopicExchangeResult.java b/broker-core/src/main/java/org/apache/qpid/server/exchange/topic/TopicExchangeResult.java
index 96405b9..eb9bcaf 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/exchange/topic/TopicExchangeResult.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/exchange/topic/TopicExchangeResult.java
@@ -33,22 +33,22 @@
import org.apache.qpid.server.exchange.AbstractExchange;
import org.apache.qpid.server.filter.FilterManager;
import org.apache.qpid.server.filter.Filterable;
-import org.apache.qpid.server.model.Queue;
+import org.apache.qpid.server.message.MessageDestination;
public final class TopicExchangeResult implements TopicMatcherResult
{
private final List<AbstractExchange.BindingIdentifier> _bindings = new CopyOnWriteArrayList<>();
- private final Map<Queue<?>, Integer> _unfilteredQueues = new ConcurrentHashMap<>();
- private final ConcurrentMap<Queue<?>, Map<FilterManager,Integer>> _filteredQueues = new ConcurrentHashMap<>();
- private volatile ArrayList<Queue<?>> _unfilteredQueueList = new ArrayList<>(0);
+ private final Map<MessageDestination, Integer> _unfilteredQueues = new ConcurrentHashMap<>();
+ private final ConcurrentMap<MessageDestination, Map<FilterManager,Integer>> _filteredQueues = new ConcurrentHashMap<>();
+ private volatile ArrayList<MessageDestination> _unfilteredQueueList = new ArrayList<>(0);
- public void addUnfilteredQueue(Queue<?> queue)
+ public void addUnfilteredQueue(MessageDestination queue)
{
Integer instances = _unfilteredQueues.get(queue);
if(instances == null)
{
_unfilteredQueues.put(queue, 1);
- ArrayList<Queue<?>> newList = new ArrayList<>(_unfilteredQueueList);
+ ArrayList<MessageDestination> newList = new ArrayList<>(_unfilteredQueueList);
newList.add(queue);
_unfilteredQueueList = newList;
}
@@ -58,13 +58,13 @@
}
}
- public void removeUnfilteredQueue(Queue<?> queue)
+ public void removeUnfilteredQueue(MessageDestination queue)
{
Integer instances = _unfilteredQueues.get(queue);
if(instances == 1)
{
_unfilteredQueues.remove(queue);
- ArrayList<Queue<?>> newList = new ArrayList<>(_unfilteredQueueList);
+ ArrayList<MessageDestination> newList = new ArrayList<>(_unfilteredQueueList);
newList.remove(queue);
_unfilteredQueueList = newList;
@@ -76,7 +76,7 @@
}
- public Collection<Queue<?>> getUnfilteredQueues()
+ public Collection<MessageDestination> getUnfilteredQueues()
{
return _unfilteredQueues.keySet();
}
@@ -96,7 +96,7 @@
return new ArrayList<>(_bindings);
}
- public void addFilteredQueue(Queue<?> queue, FilterManager filter)
+ public void addFilteredQueue(MessageDestination queue, FilterManager filter)
{
Map<FilterManager,Integer> filters = _filteredQueues.get(queue);
if(filters == null)
@@ -116,7 +116,7 @@
}
- public void removeFilteredQueue(Queue<?> queue, FilterManager filter)
+ public void removeFilteredQueue(MessageDestination queue, FilterManager filter)
{
Map<FilterManager,Integer> filters = _filteredQueues.get(queue);
if(filters != null)
@@ -142,7 +142,7 @@
}
- public void replaceQueueFilter(Queue<?> queue,
+ public void replaceQueueFilter(MessageDestination queue,
FilterManager oldFilter,
FilterManager newFilter)
{
@@ -169,7 +169,7 @@
_filteredQueues.put(queue,newFilters);
}
- public Collection<Queue<?>> processMessage(Filterable msg, Collection<Queue<?>> queues)
+ public Collection<MessageDestination> processMessage(Filterable msg, Collection<MessageDestination> queues)
{
if(queues == null)
{
@@ -179,18 +179,18 @@
}
else
{
- queues = new HashSet<Queue<?>>();
+ queues = new HashSet<>();
}
}
else if(!(queues instanceof Set))
{
- queues = new HashSet<Queue<?>>(queues);
+ queues = new HashSet<>(queues);
}
queues.addAll(_unfilteredQueues.keySet());
if(!_filteredQueues.isEmpty())
{
- for(Map.Entry<Queue<?>, Map<FilterManager, Integer>> entry : _filteredQueues.entrySet())
+ for(Map.Entry<MessageDestination, Map<FilterManager, Integer>> entry : _filteredQueues.entrySet())
{
if(!queues.contains(entry.getKey()))
{
diff --git a/broker-core/src/main/java/org/apache/qpid/server/filter/FilterSupport.java b/broker-core/src/main/java/org/apache/qpid/server/filter/FilterSupport.java
index ce7f4b9..45fb744 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/filter/FilterSupport.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/filter/FilterSupport.java
@@ -31,6 +31,7 @@
import org.apache.qpid.filter.SelectorParsingException;
import org.apache.qpid.filter.selector.ParseException;
import org.apache.qpid.filter.selector.TokenMgrError;
+import org.apache.qpid.server.message.MessageDestination;
import org.apache.qpid.server.model.Queue;
import org.apache.qpid.server.plugin.PluggableService;
import org.apache.qpid.server.queue.QueueConsumer;
@@ -94,13 +95,13 @@
&& ((String)args.get(AMQPFilterTypes.JMS_SELECTOR.toString())).trim().length() != 0;
}
- public static FilterManager createMessageFilter(final Map<String,Object> args, Queue<?> queue) throws AMQInvalidArgumentException
+ public static FilterManager createMessageFilter(final Map<String,Object> args, MessageDestination queue) throws AMQInvalidArgumentException
{
FilterManager filterManager = null;
- if(argumentsContainNoLocal(args))
+ if(argumentsContainNoLocal(args) && queue instanceof Queue)
{
filterManager = new FilterManager();
- filterManager.add(AMQPFilterTypes.NO_LOCAL.toString(), new NoLocalFilter(queue));
+ filterManager.add(AMQPFilterTypes.NO_LOCAL.toString(), new NoLocalFilter((Queue<?>) queue));
}
if(argumentsContainJMSSelector(args))
diff --git a/broker-core/src/main/java/org/apache/qpid/server/message/MessageDestination.java b/broker-core/src/main/java/org/apache/qpid/server/message/MessageDestination.java
index 8fc5cc0..8cad1db 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/message/MessageDestination.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/message/MessageDestination.java
@@ -27,8 +27,6 @@
import org.apache.qpid.server.model.PublishingLink;
import org.apache.qpid.server.security.SecurityToken;
import org.apache.qpid.server.store.StorableMessageMetaData;
-import org.apache.qpid.server.txn.ServerTransaction;
-import org.apache.qpid.server.util.Action;
public interface MessageDestination extends MessageNode
{
@@ -46,15 +44,10 @@
* @param message the message to be routed
* @param routingAddress
* @param instanceProperties the instance properties
- * @param txn the transaction to enqueue within
- * @param postEnqueueAction action to perform on the result of every enqueue (may be null)
- * @return the number of queues in which the message was enqueued performed
- */
- <M extends ServerMessage<? extends StorableMessageMetaData>> int send(M message,
- final String routingAddress,
- InstanceProperties instanceProperties,
- ServerTransaction txn,
- Action<? super MessageInstance> postEnqueueAction);
+ */
+ <M extends ServerMessage<? extends StorableMessageMetaData>> RoutingResult<M> route(M message,
+ String routingAddress,
+ InstanceProperties instanceProperties);
boolean isDurable();
diff --git a/broker-core/src/main/java/org/apache/qpid/server/message/RoutingResult.java b/broker-core/src/main/java/org/apache/qpid/server/message/RoutingResult.java
new file mode 100644
index 0000000..231443b
--- /dev/null
+++ b/broker-core/src/main/java/org/apache/qpid/server/message/RoutingResult.java
@@ -0,0 +1,147 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.server.message;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.Set;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.qpid.server.queue.BaseQueue;
+import org.apache.qpid.server.store.MessageEnqueueRecord;
+import org.apache.qpid.server.store.StorableMessageMetaData;
+import org.apache.qpid.server.txn.ServerTransaction;
+import org.apache.qpid.server.util.Action;
+
+public class RoutingResult<M extends ServerMessage<? extends StorableMessageMetaData>>
+{
+ private static final Logger _logger = LoggerFactory.getLogger(RoutingResult.class);
+
+ private final M _message;
+
+ private final Set<BaseQueue> _queues = new HashSet<>();
+
+ public RoutingResult(final M message)
+ {
+ _message = message;
+ }
+
+ public void addQueue(BaseQueue q)
+ {
+ if(q.isDeleted())
+ {
+ _logger.debug("Attempt to enqueue message onto deleted queue {}", q.getName());
+ }
+ else
+ {
+ _queues.add(q);
+ }
+ }
+
+ public void addQueues(Collection<? extends BaseQueue> queues)
+ {
+ boolean deletedQueues = false;
+ for(BaseQueue q : queues)
+ {
+ if(q.isDeleted())
+ {
+ if(!deletedQueues)
+ {
+ deletedQueues = true;
+ queues = new ArrayList<>(queues);
+ }
+ _logger.debug("Attempt to enqueue message onto deleted queue {}", q.getName());
+
+ queues.remove(q);
+ }
+ }
+
+ _queues.addAll(queues);
+ }
+
+ public void add(RoutingResult<M> result)
+ {
+ addQueues(result._queues);
+ }
+
+ public int send(ServerTransaction txn,
+ final Action<? super MessageInstance> postEnqueueAction)
+ {
+ for(BaseQueue q : _queues)
+ {
+ if(!_message.isResourceAcceptable(q))
+ {
+ return 0;
+ }
+ }
+ final BaseQueue[] baseQueues;
+
+ if(_message.isReferenced())
+ {
+ ArrayList<BaseQueue> uniqueQueues = new ArrayList<>(_queues.size());
+ for(BaseQueue q : _queues)
+ {
+ if(!_message.isReferenced(q))
+ {
+ uniqueQueues.add(q);
+ }
+ }
+ baseQueues = uniqueQueues.toArray(new BaseQueue[uniqueQueues.size()]);
+ }
+ else
+ {
+ baseQueues = _queues.toArray(new BaseQueue[_queues.size()]);
+ }
+ txn.enqueue(_queues, _message, new ServerTransaction.EnqueueAction()
+ {
+ MessageReference _reference = _message.newReference();
+
+ public void postCommit(MessageEnqueueRecord... records)
+ {
+ try
+ {
+ for(int i = 0; i < baseQueues.length; i++)
+ {
+ baseQueues[i].enqueue(_message, postEnqueueAction, records[i]);
+ }
+ }
+ finally
+ {
+ _reference.release();
+ }
+ }
+
+ public void onRollback()
+ {
+ _reference.release();
+ }
+ });
+ return _queues.size();
+ }
+
+ public boolean hasRoutes()
+ {
+ return !_queues.isEmpty();
+ }
+}
diff --git a/broker-core/src/main/java/org/apache/qpid/server/queue/AbstractQueue.java b/broker-core/src/main/java/org/apache/qpid/server/queue/AbstractQueue.java
index fb37b56..0b366ae 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/queue/AbstractQueue.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/queue/AbstractQueue.java
@@ -91,6 +91,7 @@
import org.apache.qpid.server.message.MessageInstance;
import org.apache.qpid.server.message.MessageReference;
import org.apache.qpid.server.message.MessageSender;
+import org.apache.qpid.server.message.RoutingResult;
import org.apache.qpid.server.message.ServerMessage;
import org.apache.qpid.server.message.internal.InternalMessage;
import org.apache.qpid.server.model.*;
@@ -2567,46 +2568,21 @@
_notificationListener = listener == null ? NULL_NOTIFICATION_LISTENER : listener;
}
- public final <M extends ServerMessage<? extends StorableMessageMetaData>> int send(final M message,
- final String routingAddress,
- final InstanceProperties instanceProperties,
- final ServerTransaction txn,
- final Action<? super MessageInstance> postEnqueueAction)
+ @Override
+ public <M extends ServerMessage<? extends StorableMessageMetaData>> RoutingResult<M> route(final M message,
+ final String routingAddress,
+ final InstanceProperties instanceProperties)
{
if (_virtualHost.getState() != State.ACTIVE)
{
throw new VirtualHostUnavailableException(this._virtualHost);
}
+ RoutingResult<M> result = new RoutingResult<>(message);
if(message.isResourceAcceptable(this) && !message.isReferenced(this))
{
- txn.enqueue(this, message, new ServerTransaction.EnqueueAction()
- {
- MessageReference _reference = message.newReference();
-
- public void postCommit(MessageEnqueueRecord... records)
- {
- try
- {
- AbstractQueue.this.enqueue(message, postEnqueueAction, records[0]);
- }
- finally
- {
- _reference.release();
- }
- }
-
- public void onRollback()
- {
- _reference.release();
- }
- });
- return 1;
+ result.addQueue(this);
}
- else
- {
- return 0;
- }
-
+ return result;
}
@Override
diff --git a/broker-core/src/main/java/org/apache/qpid/server/queue/BaseQueue.java b/broker-core/src/main/java/org/apache/qpid/server/queue/BaseQueue.java
index dc2f45a..ac74850 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/queue/BaseQueue.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/queue/BaseQueue.java
@@ -30,9 +30,5 @@
public interface BaseQueue extends TransactionLogResource
{
void enqueue(ServerMessage message, Action<? super MessageInstance> action, MessageEnqueueRecord record);
-
- boolean isDurable();
boolean isDeleted();
-
- String getName();
}
diff --git a/broker-core/src/main/java/org/apache/qpid/server/queue/QueueEntryImpl.java b/broker-core/src/main/java/org/apache/qpid/server/queue/QueueEntryImpl.java
index bfc1faa..6a6f4e0 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/queue/QueueEntryImpl.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/queue/QueueEntryImpl.java
@@ -36,6 +36,7 @@
import org.apache.qpid.server.message.MessageInstance;
import org.apache.qpid.server.message.MessageInstanceConsumer;
import org.apache.qpid.server.message.MessageReference;
+import org.apache.qpid.server.message.RoutingResult;
import org.apache.qpid.server.message.ServerMessage;
import org.apache.qpid.server.model.Exchange;
import org.apache.qpid.server.model.Queue;
@@ -569,27 +570,23 @@
final Queue<?> currentQueue = getQueue();
Exchange<?> alternateExchange = currentQueue.getAlternateExchange();
boolean autocommit = txn == null;
- int enqueues;
if(autocommit)
{
txn = new LocalTransaction(getQueue().getVirtualHost().getMessageStore());
}
+ RoutingResult result;
if (alternateExchange != null)
{
- enqueues = alternateExchange.send(getMessage(),
- getMessage().getInitialRoutingAddress(),
- getInstanceProperties(),
- txn,
- action);
+ result = alternateExchange.route(getMessage(), getMessage().getInitialRoutingAddress(),
+ getInstanceProperties());
}
else
{
- enqueues = 0;
+ result = new RoutingResult<>(getMessage());
}
-
- txn.dequeue(getEnqueueRecord(), new ServerTransaction.Action()
+ txn.addPostTransactionAction(new ServerTransaction.Action()
{
public void postCommit()
{
@@ -601,13 +598,13 @@
}
});
+ int enqueues = result.send(txn, null);
if(autocommit)
{
txn.commit();
}
-
- return enqueues;
+ return enqueues;
}
public boolean isQueueDeleted()
diff --git a/broker-core/src/main/java/org/apache/qpid/server/store/TransactionLogResource.java b/broker-core/src/main/java/org/apache/qpid/server/store/TransactionLogResource.java
index f5b1aa6..5ccf10d 100755
--- a/broker-core/src/main/java/org/apache/qpid/server/store/TransactionLogResource.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/store/TransactionLogResource.java
@@ -24,9 +24,7 @@
public interface TransactionLogResource
{
-
String getName();
- public UUID getId();
- //boolean isDurable();
+ UUID getId();
MessageDurability getMessageDurability();
}
diff --git a/broker-core/src/main/java/org/apache/qpid/server/txn/AsyncAutoCommitTransaction.java b/broker-core/src/main/java/org/apache/qpid/server/txn/AsyncAutoCommitTransaction.java
index 25de867..1ca1e21 100755
--- a/broker-core/src/main/java/org/apache/qpid/server/txn/AsyncAutoCommitTransaction.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/txn/AsyncAutoCommitTransaction.java
@@ -259,7 +259,7 @@
}
- public void enqueue(List<? extends BaseQueue> queues, EnqueueableMessage message, EnqueueAction postTransactionAction)
+ public void enqueue(Collection<? extends BaseQueue> queues, EnqueueableMessage message, EnqueueAction postTransactionAction)
{
Transaction txn = null;
try
diff --git a/broker-core/src/main/java/org/apache/qpid/server/txn/AutoCommitTransaction.java b/broker-core/src/main/java/org/apache/qpid/server/txn/AutoCommitTransaction.java
index c789199..a59cb7e 100755
--- a/broker-core/src/main/java/org/apache/qpid/server/txn/AutoCommitTransaction.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/txn/AutoCommitTransaction.java
@@ -185,7 +185,7 @@
}
- public void enqueue(List<? extends BaseQueue> queues, EnqueueableMessage message, EnqueueAction postTransactionAction)
+ public void enqueue(Collection<? extends BaseQueue> queues, EnqueueableMessage message, EnqueueAction postTransactionAction)
{
Transaction txn = null;
try
diff --git a/broker-core/src/main/java/org/apache/qpid/server/txn/DistributedTransaction.java b/broker-core/src/main/java/org/apache/qpid/server/txn/DistributedTransaction.java
index 6f2e9b9..4a220e8 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/txn/DistributedTransaction.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/txn/DistributedTransaction.java
@@ -21,6 +21,8 @@
package org.apache.qpid.server.txn;
+import java.util.Collection;
+
import org.apache.qpid.server.message.EnqueueableMessage;
import org.apache.qpid.server.message.MessageInstance;
import org.apache.qpid.server.protocol.AMQSessionModel;
@@ -29,9 +31,6 @@
import org.apache.qpid.server.store.TransactionLogResource;
import org.apache.qpid.transport.Xid;
-import java.util.Collection;
-import java.util.List;
-
public class DistributedTransaction implements ServerTransaction
{
@@ -137,7 +136,7 @@
}
}
- public void enqueue(List<? extends BaseQueue> queues, EnqueueableMessage message,
+ public void enqueue(Collection<? extends BaseQueue> queues, EnqueueableMessage message,
final EnqueueAction postTransactionAction)
{
if(_branch != null)
diff --git a/broker-core/src/main/java/org/apache/qpid/server/txn/LocalTransaction.java b/broker-core/src/main/java/org/apache/qpid/server/txn/LocalTransaction.java
index 291ec6b..9c96083 100755
--- a/broker-core/src/main/java/org/apache/qpid/server/txn/LocalTransaction.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/txn/LocalTransaction.java
@@ -264,7 +264,7 @@
}
}
- public void enqueue(List<? extends BaseQueue> queues, EnqueueableMessage message, EnqueueAction postTransactionAction)
+ public void enqueue(Collection<? extends BaseQueue> queues, EnqueueableMessage message, EnqueueAction postTransactionAction)
{
sync();
initTransactionStartTimeIfNecessaryAndAdvanceUpdateTime();
diff --git a/broker-core/src/main/java/org/apache/qpid/server/txn/ServerTransaction.java b/broker-core/src/main/java/org/apache/qpid/server/txn/ServerTransaction.java
index 5251eea..53eab50 100755
--- a/broker-core/src/main/java/org/apache/qpid/server/txn/ServerTransaction.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/txn/ServerTransaction.java
@@ -21,7 +21,6 @@
package org.apache.qpid.server.txn;
import java.util.Collection;
-import java.util.List;
import org.apache.qpid.server.message.EnqueueableMessage;
import org.apache.qpid.server.message.MessageInstance;
@@ -111,7 +110,7 @@
*
* Store operations will result only for a persistent messages on durable queues.
*/
- void enqueue(List<? extends BaseQueue> queues, EnqueueableMessage message, EnqueueAction postTransactionAction);
+ void enqueue(Collection<? extends BaseQueue> queues, EnqueueableMessage message, EnqueueAction postTransactionAction);
/**
* Commit the transaction represented by this object.
diff --git a/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractVirtualHost.java b/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractVirtualHost.java
index 1a1f101..111b634 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractVirtualHost.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractVirtualHost.java
@@ -89,6 +89,7 @@
import org.apache.qpid.server.message.MessageDestination;
import org.apache.qpid.server.message.MessageNode;
import org.apache.qpid.server.message.MessageSource;
+import org.apache.qpid.server.message.RoutingResult;
import org.apache.qpid.server.message.ServerMessage;
import org.apache.qpid.server.message.internal.InternalMessage;
import org.apache.qpid.server.model.*;
@@ -801,8 +802,9 @@
}
}
};
-
- return destination.send(internalMessage, address, instanceProperties, txn, null);
+ final RoutingResult<InternalMessage> result =
+ destination.route(internalMessage, address, instanceProperties);
+ return result.send(txn, null);
}
diff --git a/broker-core/src/test/java/org/apache/qpid/server/exchange/FanoutExchangeTest.java b/broker-core/src/test/java/org/apache/qpid/server/exchange/FanoutExchangeTest.java
index 98ea2b1..a9e1d80 100644
--- a/broker-core/src/test/java/org/apache/qpid/server/exchange/FanoutExchangeTest.java
+++ b/broker-core/src/test/java/org/apache/qpid/server/exchange/FanoutExchangeTest.java
@@ -20,11 +20,15 @@
*/
package org.apache.qpid.server.exchange;
+import static org.mockito.Matchers.any;
import static org.mockito.Matchers.anySet;
+import static org.mockito.Matchers.anyString;
import static org.mockito.Matchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
+import java.util.ArrayList;
+import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
@@ -40,15 +44,22 @@
import org.apache.qpid.server.configuration.updater.TaskExecutor;
import org.apache.qpid.server.logging.EventLogger;
import org.apache.qpid.server.message.AMQMessageHeader;
+import org.apache.qpid.server.message.EnqueueableMessage;
import org.apache.qpid.server.message.InstanceProperties;
+import org.apache.qpid.server.message.MessageInstance;
+import org.apache.qpid.server.message.RoutingResult;
import org.apache.qpid.server.message.ServerMessage;
import org.apache.qpid.server.model.Broker;
import org.apache.qpid.server.model.BrokerModel;
import org.apache.qpid.server.model.Exchange;
import org.apache.qpid.server.model.Queue;
+import org.apache.qpid.server.model.State;
import org.apache.qpid.server.model.VirtualHost;
import org.apache.qpid.server.model.VirtualHostNode;
import org.apache.qpid.server.queue.BaseQueue;
+import org.apache.qpid.server.store.MessageEnqueueRecord;
+import org.apache.qpid.server.store.TransactionLogResource;
+import org.apache.qpid.server.txn.ServerTransaction;
import org.apache.qpid.server.virtualhost.QueueManagingVirtualHost;
import org.apache.qpid.test.utils.QpidTestCase;
@@ -79,6 +90,7 @@
_virtualHost = mock(QueueManagingVirtualHost.class);
when(_virtualHost.getEventLogger()).thenReturn(new EventLogger());
+ when(_virtualHost.getState()).thenReturn(State.ACTIVE);
when(_virtualHost.getTaskExecutor()).thenReturn(_taskExecutor);
when(_virtualHost.getChildExecutor()).thenReturn(_taskExecutor);
when(_virtualHost.getModel()).thenReturn(BrokerModel.getInstance());
@@ -153,6 +165,9 @@
when(queue.getChildExecutor()).thenReturn(taskExecutor);
when(queue.getParent()).thenReturn(_virtualHost);
when(_virtualHost.getAttainedQueue(eq(name))).thenReturn(queue);
+ RoutingResult result = new RoutingResult(null);
+ result.addQueue(queue);
+ when(queue.route(any(ServerMessage.class),anyString(),any(InstanceProperties.class))).thenReturn(result);
return queue;
}
@@ -165,8 +180,8 @@
_exchange.addBinding("key",queue1, null);
_exchange.addBinding("key",queue2, null);
-
- List<? extends BaseQueue> result = _exchange.route(mockMessage(true), "", InstanceProperties.EMPTY);
+ List<? extends BaseQueue> result;
+ result = routeToQueues(mockMessage(true), "", InstanceProperties.EMPTY);
assertEquals("Expected message to be routed to both queues", 2, result.size());
assertTrue("Expected queue1 to be routed to", result.contains(queue1));
@@ -174,8 +189,7 @@
_exchange.addBinding("key2",queue2, Collections.singletonMap(AMQPFilterTypes.JMS_SELECTOR.toString(),(Object)"select = True"));
-
- result = _exchange.route(mockMessage(true), "", InstanceProperties.EMPTY);
+ result = routeToQueues(mockMessage(true), "", InstanceProperties.EMPTY);
assertEquals("Expected message to be routed to both queues", 2, result.size());
assertTrue("Expected queue1 to be routed to", result.contains(queue1));
@@ -183,14 +197,13 @@
_exchange.deleteBinding("key",queue2);
- result = _exchange.route(mockMessage(true), "", InstanceProperties.EMPTY);
+ result = routeToQueues(mockMessage(true), "", InstanceProperties.EMPTY);
assertEquals("Expected message to be routed to both queues", 2, result.size());
assertTrue("Expected queue1 to be routed to", result.contains(queue1));
assertTrue("Expected queue2 to be routed to", result.contains(queue2));
-
- result = _exchange.route(mockMessage(false), "", InstanceProperties.EMPTY);
+ result = routeToQueues(mockMessage(false), "", InstanceProperties.EMPTY);
assertEquals("Expected message to be routed to queue1 only", 1, result.size());
assertTrue("Expected queue1 to be routed to", result.contains(queue1));
@@ -198,8 +211,7 @@
_exchange.addBinding("key",queue2, Collections.singletonMap(AMQPFilterTypes.JMS_SELECTOR.toString(),(Object)"select = False"));
-
- result = _exchange.route(mockMessage(false), "", InstanceProperties.EMPTY);
+ result = routeToQueues(mockMessage(false), "", InstanceProperties.EMPTY);
assertEquals("Expected message to be routed to both queues", 2, result.size());
assertTrue("Expected queue1 to be routed to", result.contains(queue1));
assertTrue("Expected queue2 to be routed to", result.contains(queue2));
@@ -207,6 +219,88 @@
}
+ private List<? extends BaseQueue> routeToQueues(final ServerMessage message,
+ final String routingAddress,
+ final InstanceProperties instanceProperties)
+ {
+ RoutingResult result = _exchange.route(message, routingAddress, instanceProperties);
+ final List<BaseQueue> resultQueues = new ArrayList<>();
+ result.send(new ServerTransaction()
+ {
+ @Override
+ public long getTransactionStartTime()
+ {
+ return 0;
+ }
+
+ @Override
+ public long getTransactionUpdateTime()
+ {
+ return 0;
+ }
+
+ @Override
+ public void addPostTransactionAction(final Action postTransactionAction)
+ {
+
+ }
+
+ @Override
+ public void dequeue(final MessageEnqueueRecord record, final Action postTransactionAction)
+ {
+
+ }
+
+ @Override
+ public void dequeue(final Collection<MessageInstance> messages, final Action postTransactionAction)
+ {
+
+ }
+
+ @Override
+ public void enqueue(final TransactionLogResource queue,
+ final EnqueueableMessage message,
+ final EnqueueAction postTransactionAction)
+ {
+ resultQueues.add((BaseQueue) queue);
+ }
+
+ @Override
+ public void enqueue(final Collection<? extends BaseQueue> queues,
+ final EnqueueableMessage message,
+ final EnqueueAction postTransactionAction)
+ {
+ resultQueues.addAll(queues);
+ }
+
+ @Override
+ public void commit()
+ {
+
+ }
+
+ @Override
+ public void commit(final Runnable immediatePostTransactionAction)
+ {
+
+ }
+
+ @Override
+ public void rollback()
+ {
+
+ }
+
+ @Override
+ public boolean isTransactional()
+ {
+ return false;
+ }
+ }, null);
+
+ return resultQueues;
+ }
+
private ServerMessage mockMessage(boolean val)
{
final AMQMessageHeader header = mock(AMQMessageHeader.class);
@@ -225,6 +319,7 @@
});
final ServerMessage serverMessage = mock(ServerMessage.class);
when(serverMessage.getMessageHeader()).thenReturn(header);
+ when(serverMessage.isResourceAcceptable(any(TransactionLogResource.class))).thenReturn(true);
return serverMessage;
}
}
diff --git a/broker-core/src/test/java/org/apache/qpid/server/exchange/HeadersExchangeTest.java b/broker-core/src/test/java/org/apache/qpid/server/exchange/HeadersExchangeTest.java
index a0b0cee..13b38ab 100644
--- a/broker-core/src/test/java/org/apache/qpid/server/exchange/HeadersExchangeTest.java
+++ b/broker-core/src/test/java/org/apache/qpid/server/exchange/HeadersExchangeTest.java
@@ -20,6 +20,7 @@
*/
package org.apache.qpid.server.exchange;
+import static org.mockito.Matchers.any;
import static org.mockito.Matchers.anySet;
import static org.mockito.Matchers.anyString;
import static org.mockito.Mockito.mock;
@@ -27,6 +28,7 @@
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -42,16 +44,23 @@
import org.apache.qpid.server.configuration.updater.TaskExecutor;
import org.apache.qpid.server.logging.EventLogger;
import org.apache.qpid.server.message.AMQMessageHeader;
+import org.apache.qpid.server.message.EnqueueableMessage;
import org.apache.qpid.server.message.InstanceProperties;
+import org.apache.qpid.server.message.MessageInstance;
+import org.apache.qpid.server.message.RoutingResult;
import org.apache.qpid.server.message.ServerMessage;
import org.apache.qpid.server.model.Broker;
import org.apache.qpid.server.model.BrokerModel;
import org.apache.qpid.server.model.ConfiguredObjectFactoryImpl;
import org.apache.qpid.server.model.Exchange;
import org.apache.qpid.server.model.Queue;
+import org.apache.qpid.server.model.State;
import org.apache.qpid.server.model.VirtualHost;
import org.apache.qpid.server.model.VirtualHostNode;
import org.apache.qpid.server.queue.BaseQueue;
+import org.apache.qpid.server.store.MessageEnqueueRecord;
+import org.apache.qpid.server.store.TransactionLogResource;
+import org.apache.qpid.server.txn.ServerTransaction;
import org.apache.qpid.server.virtualhost.QueueManagingVirtualHost;
import org.apache.qpid.test.utils.QpidTestCase;
@@ -84,6 +93,7 @@
when(_virtualHost.getCategoryClass()).thenReturn(VirtualHost.class);
when(_virtualHost.getTaskExecutor()).thenReturn(_taskExecutor);
when(_virtualHost.getChildExecutor()).thenReturn(_taskExecutor);
+ when(_virtualHost.getState()).thenReturn(State.ACTIVE);
_factory = new ConfiguredObjectFactoryImpl(BrokerModel.getInstance());
when(_virtualHost.getObjectFactory()).thenReturn(_factory);
@@ -106,7 +116,7 @@
protected void routeAndTest(ServerMessage msg, Queue<?>... expected) throws Exception
{
- List<? extends BaseQueue> results = _exchange.route(msg, "", InstanceProperties.EMPTY);
+ List<? extends BaseQueue> results = routeToQueues(msg, "", InstanceProperties.EMPTY);
List<? extends BaseQueue> unexpected = new ArrayList<BaseQueue>(results);
unexpected.removeAll(Arrays.asList(expected));
assertTrue("Message delivered to unexpected queues: " + unexpected, unexpected.isEmpty());
@@ -116,6 +126,88 @@
assertTrue("Duplicates " + results, results.size()==(new HashSet<BaseQueue>(results)).size());
}
+ private List<? extends BaseQueue> routeToQueues(final ServerMessage message,
+ final String routingAddress,
+ final InstanceProperties instanceProperties)
+ {
+ RoutingResult result = _exchange.route(message, routingAddress, instanceProperties);
+ final List<BaseQueue> resultQueues = new ArrayList<>();
+ result.send(new ServerTransaction()
+ {
+ @Override
+ public long getTransactionStartTime()
+ {
+ return 0;
+ }
+
+ @Override
+ public long getTransactionUpdateTime()
+ {
+ return 0;
+ }
+
+ @Override
+ public void addPostTransactionAction(final Action postTransactionAction)
+ {
+
+ }
+
+ @Override
+ public void dequeue(final MessageEnqueueRecord record, final Action postTransactionAction)
+ {
+
+ }
+
+ @Override
+ public void dequeue(final Collection<MessageInstance> messages, final Action postTransactionAction)
+ {
+
+ }
+
+ @Override
+ public void enqueue(final TransactionLogResource queue,
+ final EnqueueableMessage message,
+ final EnqueueAction postTransactionAction)
+ {
+ resultQueues.add((BaseQueue) queue);
+ }
+
+ @Override
+ public void enqueue(final Collection<? extends BaseQueue> queues,
+ final EnqueueableMessage message,
+ final EnqueueAction postTransactionAction)
+ {
+ resultQueues.addAll(queues);
+ }
+
+ @Override
+ public void commit()
+ {
+
+ }
+
+ @Override
+ public void commit(final Runnable immediatePostTransactionAction)
+ {
+
+ }
+
+ @Override
+ public void rollback()
+ {
+
+ }
+
+ @Override
+ public boolean isTransactional()
+ {
+ return false;
+ }
+ }, null);
+
+ return resultQueues;
+ }
+
private Queue<?> createAndBind(final String name, String... arguments)
throws Exception
@@ -169,6 +261,9 @@
when(q.getTaskExecutor()).thenReturn(taskExecutor);
when(q.getChildExecutor()).thenReturn(taskExecutor);
when(_virtualHost.getAttainedQueue(name)).thenReturn(q);
+ final RoutingResult routingResult = new RoutingResult(null);
+ routingResult.addQueue(q);
+ when(q.route(any(ServerMessage.class), anyString(), any(InstanceProperties.class))).thenReturn(routingResult);
return q;
}
@@ -282,6 +377,7 @@
});
final ServerMessage serverMessage = mock(ServerMessage.class);
when(serverMessage.getMessageHeader()).thenReturn(header);
+ when(serverMessage.isResourceAcceptable(any(TransactionLogResource.class))).thenReturn(true);
return serverMessage;
}
diff --git a/broker-core/src/test/java/org/apache/qpid/server/exchange/TopicExchangeTest.java b/broker-core/src/test/java/org/apache/qpid/server/exchange/TopicExchangeTest.java
index 1ba6762..e445f9c 100644
--- a/broker-core/src/test/java/org/apache/qpid/server/exchange/TopicExchangeTest.java
+++ b/broker-core/src/test/java/org/apache/qpid/server/exchange/TopicExchangeTest.java
@@ -25,6 +25,8 @@
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
+import java.util.ArrayList;
+import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
@@ -34,15 +36,20 @@
import org.apache.qpid.exchange.ExchangeDefaults;
import org.apache.qpid.server.message.AMQMessageHeader;
+import org.apache.qpid.server.message.EnqueueableMessage;
import org.apache.qpid.server.message.InstanceProperties;
+import org.apache.qpid.server.message.MessageInstance;
import org.apache.qpid.server.message.MessageReference;
+import org.apache.qpid.server.message.RoutingResult;
import org.apache.qpid.server.message.ServerMessage;
import org.apache.qpid.server.model.BrokerTestHelper;
import org.apache.qpid.server.model.Exchange;
import org.apache.qpid.server.model.Queue;
import org.apache.qpid.server.model.VirtualHost;
import org.apache.qpid.server.queue.BaseQueue;
+import org.apache.qpid.server.store.MessageEnqueueRecord;
import org.apache.qpid.server.store.TransactionLogResource;
+import org.apache.qpid.server.txn.ServerTransaction;
import org.apache.qpid.test.utils.QpidTestCase;
public class TopicExchangeTest extends QpidTestCase
@@ -353,24 +360,31 @@
_exchange.bind(queue.getName(), bindingKey, bindArgs, false);
ServerMessage matchMsg1 = mock(ServerMessage.class);
+ when(matchMsg1.isResourceAcceptable(any(TransactionLogResource.class))).thenReturn(true);
AMQMessageHeader msgHeader1 = createMessageHeader(Collections.<String, Object>singletonMap("arg", 6));
when(matchMsg1.getMessageHeader()).thenReturn(msgHeader1);
routeMessage(matchMsg1, bindingKey, 1);
Assert.assertEquals("First message should be routed to queue", 1, queue.getQueueDepthMessages());
ServerMessage nonmatchMsg2 = mock(ServerMessage.class);
+ when(nonmatchMsg2.isResourceAcceptable(any(TransactionLogResource.class))).thenReturn(true);
+
AMQMessageHeader msgHeader2 = createMessageHeader(Collections.<String, Object>singletonMap("arg", 5));
when(nonmatchMsg2.getMessageHeader()).thenReturn(msgHeader2);
routeMessage(nonmatchMsg2, bindingKey, 2);
Assert.assertEquals("Second message should not be routed to queue", 1, queue.getQueueDepthMessages());
ServerMessage nonmatchMsg3 = mock(ServerMessage.class);
+ when(nonmatchMsg3.isResourceAcceptable(any(TransactionLogResource.class))).thenReturn(true);
+
AMQMessageHeader msgHeader3 = createMessageHeader(Collections.<String, Object>emptyMap());
when(nonmatchMsg3.getMessageHeader()).thenReturn(msgHeader3);
routeMessage(nonmatchMsg3, bindingKey, 3);
Assert.assertEquals("Third message should not be routed to queue", 1, queue.getQueueDepthMessages());
ServerMessage matchMsg4 = mock(ServerMessage.class);
+ when(matchMsg4.isResourceAcceptable(any(TransactionLogResource.class))).thenReturn(true);
+
AMQMessageHeader msgHeader4 = createMessageHeader(Collections.<String, Object>singletonMap("arg", 7));
when(matchMsg4.getMessageHeader()).thenReturn(msgHeader4);
routeMessage(matchMsg4, bindingKey, 4);
@@ -388,6 +402,7 @@
AMQMessageHeader mgsHeader1 = createMessageHeader(Collections.<String, Object>singletonMap("arg", 6));
ServerMessage msg1 = mock(ServerMessage.class);
+ when(msg1.isResourceAcceptable(any(TransactionLogResource.class))).thenReturn(true);
when(msg1.getMessageHeader()).thenReturn(mgsHeader1);
routeMessage(msg1, bindingKey, 1);
@@ -400,6 +415,7 @@
// Message that would have matched the original selector but not the new
AMQMessageHeader mgsHeader2 = createMessageHeader(Collections.<String, Object>singletonMap("arg", 6));
ServerMessage msg2 = mock(ServerMessage.class);
+ when(msg2.isResourceAcceptable(any(TransactionLogResource.class))).thenReturn(true);
when(msg2.getMessageHeader()).thenReturn(mgsHeader2);
routeMessage(msg2, bindingKey, 2);
@@ -408,6 +424,7 @@
// Message that matches only the second
AMQMessageHeader mgsHeader3 = createMessageHeader(Collections.<String, Object>singletonMap("arg", 7));
ServerMessage msg3 = mock(ServerMessage.class);
+ when(msg3.isResourceAcceptable(any(TransactionLogResource.class))).thenReturn(true);
when(msg3.getMessageHeader()).thenReturn(mgsHeader3);
routeMessage(msg3, bindingKey, 2);
@@ -425,7 +442,7 @@
_exchange.bind(queue.getName(), bindingKey, null, false);
ServerMessage msg1 = mock(ServerMessage.class);
-
+ when(msg1.isResourceAcceptable(any(TransactionLogResource.class))).thenReturn(true);
routeMessage(msg1, bindingKey, 1);
Assert.assertEquals(1, queue.getQueueDepthMessages());
@@ -436,6 +453,7 @@
// Message that does not match the new selector
AMQMessageHeader mgsHeader2 = createMessageHeader(Collections.<String, Object>singletonMap("arg", 6));
ServerMessage msg2 = mock(ServerMessage.class);
+ when(msg2.isResourceAcceptable(any(TransactionLogResource.class))).thenReturn(true);
when(msg2.getMessageHeader()).thenReturn(mgsHeader2);
routeMessage(msg2, bindingKey, 2);
@@ -444,6 +462,8 @@
// Message that matches the selector
AMQMessageHeader mgsHeader3 = createMessageHeader(Collections.<String, Object>singletonMap("arg", 7));
ServerMessage msg3 = mock(ServerMessage.class);
+ when(msg3.isResourceAcceptable(any(TransactionLogResource.class))).thenReturn(true);
+
when(msg3.getMessageHeader()).thenReturn(mgsHeader3);
routeMessage(msg3, bindingKey, 2);
@@ -455,13 +475,14 @@
private int routeMessage(String routingKey, long messageNumber)
{
ServerMessage message = mock(ServerMessage.class);
+ when(message.isResourceAcceptable(any(TransactionLogResource.class))).thenReturn(true);
return routeMessage(message, routingKey, messageNumber);
}
private int routeMessage(ServerMessage message, String routingKey, long messageNumber)
{
when(message.getInitialRoutingAddress()).thenReturn(routingKey);
- List<? extends BaseQueue> queues = _exchange.route(message, routingKey, InstanceProperties.EMPTY);
+ List<? extends BaseQueue> queues = routeToQueues(message, routingKey, InstanceProperties.EMPTY);
MessageReference ref = mock(MessageReference.class);
when(ref.getMessage()).thenReturn(message);
when(message.newReference()).thenReturn(ref);
@@ -475,6 +496,88 @@
return queues.size();
}
+ private List<? extends BaseQueue> routeToQueues(final ServerMessage message,
+ final String routingAddress,
+ final InstanceProperties instanceProperties)
+ {
+ RoutingResult result = _exchange.route(message, routingAddress, instanceProperties);
+ final List<BaseQueue> resultQueues = new ArrayList<>();
+ result.send(new ServerTransaction()
+ {
+ @Override
+ public long getTransactionStartTime()
+ {
+ return 0;
+ }
+
+ @Override
+ public long getTransactionUpdateTime()
+ {
+ return 0;
+ }
+
+ @Override
+ public void addPostTransactionAction(final Action postTransactionAction)
+ {
+
+ }
+
+ @Override
+ public void dequeue(final MessageEnqueueRecord record, final Action postTransactionAction)
+ {
+
+ }
+
+ @Override
+ public void dequeue(final Collection<MessageInstance> messages, final Action postTransactionAction)
+ {
+
+ }
+
+ @Override
+ public void enqueue(final TransactionLogResource queue,
+ final EnqueueableMessage message,
+ final EnqueueAction postTransactionAction)
+ {
+ resultQueues.add((BaseQueue) queue);
+ }
+
+ @Override
+ public void enqueue(final Collection<? extends BaseQueue> queues,
+ final EnqueueableMessage message,
+ final EnqueueAction postTransactionAction)
+ {
+ resultQueues.addAll(queues);
+ }
+
+ @Override
+ public void commit()
+ {
+
+ }
+
+ @Override
+ public void commit(final Runnable immediatePostTransactionAction)
+ {
+
+ }
+
+ @Override
+ public void rollback()
+ {
+
+ }
+
+ @Override
+ public boolean isTransactional()
+ {
+ return false;
+ }
+ }, null);
+
+ return resultQueues;
+ }
+
private AMQMessageHeader createMessageHeader(Map<String, Object> headers)
{
AMQMessageHeader messageHeader = mock(AMQMessageHeader.class);
diff --git a/broker-core/src/test/java/org/apache/qpid/server/txn/AsyncAutoCommitTransactionTest.java b/broker-core/src/test/java/org/apache/qpid/server/txn/AsyncAutoCommitTransactionTest.java
index 066fe61..ce226d2 100644
--- a/broker-core/src/test/java/org/apache/qpid/server/txn/AsyncAutoCommitTransactionTest.java
+++ b/broker-core/src/test/java/org/apache/qpid/server/txn/AsyncAutoCommitTransactionTest.java
@@ -54,7 +54,6 @@
when(_messageStore.newTransaction()).thenReturn(_storeTransaction);
when(_storeTransaction.commitTranAsync((Void) null)).thenReturn(_future);
- when(_queue.isDurable()).thenReturn(true);
when(_queue.getMessageDurability()).thenReturn(MessageDurability.DEFAULT);
}
diff --git a/broker-core/src/test/java/org/apache/qpid/server/txn/AutoCommitTransactionTest.java b/broker-core/src/test/java/org/apache/qpid/server/txn/AutoCommitTransactionTest.java
index 49ececb..f5deace 100644
--- a/broker-core/src/test/java/org/apache/qpid/server/txn/AutoCommitTransactionTest.java
+++ b/broker-core/src/test/java/org/apache/qpid/server/txn/AutoCommitTransactionTest.java
@@ -424,7 +424,6 @@
private BaseQueue createTestAMQQueue(final boolean durable)
{
BaseQueue queue = mock(BaseQueue.class);
- when(queue.isDurable()).thenReturn(durable);
when(queue.getMessageDurability()).thenReturn(durable ? MessageDurability.DEFAULT : MessageDurability.NEVER);
return queue;
}
diff --git a/broker-core/src/test/java/org/apache/qpid/server/txn/LocalTransactionTest.java b/broker-core/src/test/java/org/apache/qpid/server/txn/LocalTransactionTest.java
index 00ea9d6..4cc20a6 100644
--- a/broker-core/src/test/java/org/apache/qpid/server/txn/LocalTransactionTest.java
+++ b/broker-core/src/test/java/org/apache/qpid/server/txn/LocalTransactionTest.java
@@ -658,7 +658,6 @@
private BaseQueue createQueue(final boolean durable)
{
BaseQueue queue = mock(BaseQueue.class);
- when(queue.isDurable()).thenReturn(durable);
when(queue.getMessageDurability()).thenReturn(durable ? MessageDurability.DEFAULT : MessageDurability.NEVER);
return queue;
}
diff --git a/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSession.java b/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSession.java
index af2cda7..75bf70d 100644
--- a/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSession.java
+++ b/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSession.java
@@ -55,7 +55,6 @@
import org.slf4j.LoggerFactory;
import org.apache.qpid.server.connection.SessionPrincipal;
-import org.apache.qpid.server.consumer.ConsumerTarget;
import org.apache.qpid.server.consumer.ScheduledConsumerTargetSet;
import org.apache.qpid.server.logging.LogMessage;
import org.apache.qpid.server.logging.LogSubject;
@@ -65,6 +64,7 @@
import org.apache.qpid.server.message.MessageDestination;
import org.apache.qpid.server.message.MessageInstance;
import org.apache.qpid.server.message.MessageInstanceConsumer;
+import org.apache.qpid.server.message.RoutingResult;
import org.apache.qpid.server.model.AbstractConfigurationChangeListener;
import org.apache.qpid.server.model.Broker;
import org.apache.qpid.server.model.ConfigurationChangeListener;
@@ -98,22 +98,7 @@
import org.apache.qpid.server.util.Action;
import org.apache.qpid.server.util.Deletable;
import org.apache.qpid.server.util.ServerScopedRuntimeException;
-import org.apache.qpid.transport.Binary;
-import org.apache.qpid.transport.Connection;
-import org.apache.qpid.transport.MessageCreditUnit;
-import org.apache.qpid.transport.MessageFlow;
-import org.apache.qpid.transport.MessageFlowMode;
-import org.apache.qpid.transport.MessageSetFlowMode;
-import org.apache.qpid.transport.MessageStop;
-import org.apache.qpid.transport.MessageTransfer;
-import org.apache.qpid.transport.Method;
-import org.apache.qpid.transport.Option;
-import org.apache.qpid.transport.Range;
-import org.apache.qpid.transport.RangeSet;
-import org.apache.qpid.transport.RangeSetFactory;
-import org.apache.qpid.transport.Session;
-import org.apache.qpid.transport.SessionDelegate;
-import org.apache.qpid.transport.Xid;
+import org.apache.qpid.transport.*;
import org.apache.qpid.transport.network.Ticker;
public class ServerSession extends Session
@@ -298,10 +283,9 @@
_outstandingCredit.addAndGet(PRODUCER_CREDIT_TOPUP_THRESHOLD);
invoke(new MessageFlow("",MessageCreditUnit.MESSAGE, PRODUCER_CREDIT_TOPUP_THRESHOLD));
}
- int enqueues = exchange.send(message,
- message.getInitialRoutingAddress(),
- instanceProperties, _transaction, _checkCapacityAction
- );
+ final RoutingResult<MessageTransferMessage> result =
+ exchange.route(message, message.getInitialRoutingAddress(), instanceProperties);
+ int enqueues = result.send(_transaction, _checkCapacityAction);
getAMQPConnection().registerMessageReceived(message.getSize(), message.getArrivalTime());
incrementOutstandingTxnsIfNecessary();
incrementUncommittedMessageSize(message.getStoredMessage());
diff --git a/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java b/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java
index 8b3f219..26a0249 100644
--- a/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java
+++ b/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java
@@ -81,6 +81,7 @@
import org.apache.qpid.server.message.MessageInstanceConsumer;
import org.apache.qpid.server.message.MessageReference;
import org.apache.qpid.server.message.MessageSource;
+import org.apache.qpid.server.message.RoutingResult;
import org.apache.qpid.server.message.ServerMessage;
import org.apache.qpid.server.model.*;
import org.apache.qpid.server.protocol.AMQSessionModel;
@@ -494,11 +495,12 @@
}
};
- int enqueues = destination.send(amqMessage,
- amqMessage.getInitialRoutingAddress(),
- instanceProperties, _transaction,
- immediate ? _immediateAction : _capacityCheckAction
- );
+ final RoutingResult<AMQMessage> result =
+ destination.route(amqMessage,
+ amqMessage.getInitialRoutingAddress(),
+ instanceProperties);
+
+ int enqueues = result.send(_transaction, immediate ? _immediateAction : _capacityCheckAction);
if (enqueues == 0)
{
finallyAction = handleUnroutableMessage(amqMessage);
diff --git a/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/AMQChannelTest.java b/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/AMQChannelTest.java
index 32b3a00..dcb58ec 100644
--- a/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/AMQChannelTest.java
+++ b/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/AMQChannelTest.java
@@ -23,6 +23,7 @@
import static org.mockito.Matchers.any;
import static org.mockito.Matchers.anyString;
import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
@@ -37,6 +38,10 @@
import javax.security.auth.Subject;
+import org.mockito.ArgumentCaptor;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
+
import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.framing.BasicContentHeaderProperties;
import org.apache.qpid.framing.MethodRegistry;
@@ -46,6 +51,7 @@
import org.apache.qpid.server.logging.EventLogger;
import org.apache.qpid.server.message.InstanceProperties;
import org.apache.qpid.server.message.MessageDestination;
+import org.apache.qpid.server.message.RoutingResult;
import org.apache.qpid.server.message.ServerMessage;
import org.apache.qpid.server.model.Broker;
import org.apache.qpid.server.model.BrokerModel;
@@ -61,8 +67,6 @@
import org.apache.qpid.server.store.NullMessageStore;
import org.apache.qpid.server.store.StorableMessageMetaData;
import org.apache.qpid.server.store.StoredMemoryMessage;
-import org.apache.qpid.server.txn.ServerTransaction;
-import org.apache.qpid.server.util.Action;
import org.apache.qpid.server.virtualhost.QueueManagingVirtualHost;
import org.apache.qpid.test.utils.QpidTestCase;
@@ -213,7 +217,16 @@
return messageHandle;
}
});
-
+ final ArgumentCaptor<ServerMessage> messageCaptor = ArgumentCaptor.forClass(ServerMessage.class);
+ doAnswer(new Answer()
+ {
+ @Override
+ public Object answer(final InvocationOnMock invocation) throws Throwable
+ {
+ ServerMessage message = messageCaptor.getValue();
+ return new RoutingResult(message);
+ }
+ }).when(_messageDestination).route(messageCaptor.capture(), eq(ROUTING_KEY.toString()), any(InstanceProperties.class));
AMQChannel channel = new AMQChannel(_amqConnection, 1, _virtualHost.getMessageStore());
BasicContentHeaderProperties properties = new BasicContentHeaderProperties();
@@ -221,10 +234,8 @@
channel.receiveBasicPublish(AMQShortString.EMPTY_STRING, ROUTING_KEY, false, false);
channel.receiveMessageHeader(properties, 0);
- verify(_messageDestination).send((ServerMessage) any(),
+ verify(_messageDestination).route((ServerMessage) any(),
eq(ROUTING_KEY.toString()),
- any(InstanceProperties.class),
- any(ServerTransaction.class),
- any(Action.class) );
+ any(InstanceProperties.class));
}
}
diff --git a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ExchangeDestination.java b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ExchangeDestination.java
index 590f227..8cc2bab 100644
--- a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ExchangeDestination.java
+++ b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ExchangeDestination.java
@@ -29,6 +29,7 @@
import org.apache.qpid.server.logging.messages.ExchangeMessages;
import org.apache.qpid.server.message.InstanceProperties;
import org.apache.qpid.server.message.MessageInstance;
+import org.apache.qpid.server.message.RoutingResult;
import org.apache.qpid.server.message.ServerMessage;
import org.apache.qpid.server.model.Exchange;
import org.apache.qpid.server.protocol.v1_0.type.Outcome;
@@ -105,11 +106,10 @@
return null;
}};
- int enqueues = _exchange.send(message,
- routingAddress,
- instanceProperties,
- txn,
- action);
+ final RoutingResult result = _exchange.route(message,
+ routingAddress,
+ instanceProperties);
+ int enqueues = result.send(txn, action);
if(enqueues == 0)
{
diff --git a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/NodeReceivingDestination.java b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/NodeReceivingDestination.java
index c2c4cf6..8bfd2a6 100644
--- a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/NodeReceivingDestination.java
+++ b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/NodeReceivingDestination.java
@@ -28,6 +28,7 @@
import org.apache.qpid.server.message.InstanceProperties;
import org.apache.qpid.server.message.MessageDestination;
import org.apache.qpid.server.message.MessageInstance;
+import org.apache.qpid.server.message.RoutingResult;
import org.apache.qpid.server.message.ServerMessage;
import org.apache.qpid.server.model.Exchange;
import org.apache.qpid.server.protocol.v1_0.type.Outcome;
@@ -104,7 +105,8 @@
return null;
}};
- int enqueues = _destination.send(message, routingAddress, instanceProperties, txn, action);
+ RoutingResult result = _destination.route(message, routingAddress, instanceProperties);
+ int enqueues = result.send(txn, action);
if(enqueues == 0)
{
diff --git a/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementAddressSpace.java b/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementAddressSpace.java
index 8e4e609..d6c3b37 100644
--- a/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementAddressSpace.java
+++ b/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementAddressSpace.java
@@ -43,9 +43,9 @@
import org.apache.qpid.server.connection.SessionPrincipal;
import org.apache.qpid.server.message.InstanceProperties;
import org.apache.qpid.server.message.MessageDestination;
-import org.apache.qpid.server.message.MessageInstance;
import org.apache.qpid.server.message.MessageSender;
import org.apache.qpid.server.message.MessageSource;
+import org.apache.qpid.server.message.RoutingResult;
import org.apache.qpid.server.message.ServerMessage;
import org.apache.qpid.server.model.Broker;
import org.apache.qpid.server.model.Connection;
@@ -64,7 +64,6 @@
import org.apache.qpid.server.transport.AMQPConnection;
import org.apache.qpid.server.txn.DtxNotSupportedException;
import org.apache.qpid.server.txn.DtxRegistry;
-import org.apache.qpid.server.txn.ServerTransaction;
import org.apache.qpid.server.util.Action;
import org.apache.qpid.server.util.ConnectionScopedRuntimeException;
import org.apache.qpid.server.virtualhost.VirtualHostPropertiesNode;
@@ -354,20 +353,18 @@
}
@Override
- public <M extends ServerMessage<? extends StorableMessageMetaData>> int send(final M message,
- final String routingAddress,
- final InstanceProperties instanceProperties,
- final ServerTransaction txn,
- final Action<? super MessageInstance> postEnqueueAction)
+ public <M extends ServerMessage<? extends StorableMessageMetaData>> RoutingResult<M> route(final M message,
+ final String routingAddress,
+ final InstanceProperties instanceProperties)
{
MessageDestination destination = getAttainedMessageDestination(routingAddress);
if(destination == null || destination == this)
{
- return 0;
+ return new RoutingResult<>(message);
}
else
{
- return destination.send(message, routingAddress, instanceProperties, txn, postEnqueueAction);
+ return destination.route(message, routingAddress, instanceProperties);
}
}
diff --git a/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementNode.java b/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementNode.java
index 311dff5..4b9e013 100644
--- a/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementNode.java
+++ b/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementNode.java
@@ -64,6 +64,7 @@
import org.apache.qpid.server.message.MessageInstanceConsumer;
import org.apache.qpid.server.message.MessageSender;
import org.apache.qpid.server.message.MessageSource;
+import org.apache.qpid.server.message.RoutingResult;
import org.apache.qpid.server.message.ServerMessage;
import org.apache.qpid.server.message.internal.InternalMessage;
import org.apache.qpid.server.message.internal.InternalMessageHeader;
@@ -81,6 +82,7 @@
import org.apache.qpid.server.plugin.MessageConverter;
import org.apache.qpid.server.protocol.AMQSessionModel;
import org.apache.qpid.server.protocol.MessageConverterRegistry;
+import org.apache.qpid.server.queue.BaseQueue;
import org.apache.qpid.server.security.SecurityToken;
import org.apache.qpid.server.store.MessageDurability;
import org.apache.qpid.server.store.MessageEnqueueRecord;
@@ -92,7 +94,7 @@
import org.apache.qpid.server.util.ServerScopedRuntimeException;
import org.apache.qpid.server.util.StateChangeListener;
-class ManagementNode implements MessageSource, MessageDestination
+class ManagementNode implements MessageSource, MessageDestination, BaseQueue
{
private static final Logger LOGGER = LoggerFactory.getLogger(ManagementNode.class);
@@ -173,6 +175,15 @@
private final ManagementInputConverter _managementInputConverter;
+ private static final InstanceProperties CONSUMED_INSTANCE_PROPERTIES = new InstanceProperties()
+ {
+ @Override
+ public Object getProperty(final Property prop)
+ {
+ return null;
+ }
+ };
+
ManagementNode(final NamedAddressSpace addressSpace,
final ConfiguredObject<?> configuredObject)
{
@@ -332,12 +343,11 @@
}
@Override
- public <M extends ServerMessage<? extends StorableMessageMetaData>> int send(final M message,
- final String routingAddress,
- final InstanceProperties instanceProperties,
- final ServerTransaction txn,
- final Action<? super MessageInstance> postEnqueueAction)
+ public <M extends ServerMessage<? extends StorableMessageMetaData>> RoutingResult<M> route(final M message,
+ final String routingAddress,
+ final InstanceProperties instanceProperties)
{
+ final RoutingResult<M> result = new RoutingResult<>(message);
if(message.isResourceAcceptable(this))
{
@SuppressWarnings("unchecked")
@@ -347,33 +357,12 @@
if (converter != null)
{
- final InternalMessage msg = converter.convert(message, _addressSpace);
- txn.addPostTransactionAction(new ServerTransaction.Action()
- {
- @Override
- public void postCommit()
- {
- enqueue(msg, instanceProperties, postEnqueueAction);
- }
-
- @Override
- public void onRollback()
- {
-
- }
- });
-
- return 1;
+ result.addQueue(this);
}
- else
- {
- return 0;
- }
+
}
- else
- {
- return 0;
- }
+ return result;
+
}
@Override
@@ -408,7 +397,7 @@
String id = (String) message.getMessageHeader().getHeader(IDENTITY_ATTRIBUTE);
String type = (String) message.getMessageHeader().getHeader(TYPE_ATTRIBUTE);
String operation = (String) message.getMessageHeader().getHeader(OPERATION_HEADER);
-
+ LOGGER.debug("Management Node identity: {}, type: {}, operation {}", id, type, operation);
InternalMessage response;
// TODO - handle runtime exceptions
@@ -433,6 +422,26 @@
}
+ @Override
+ public void enqueue(final ServerMessage message,
+ final Action<? super MessageInstance> action,
+ final MessageEnqueueRecord record)
+ {
+ MessageConverter<ServerMessage, InternalMessage> converter =
+ (MessageConverter<ServerMessage, InternalMessage>) MessageConverterRegistry.getConverter((message.getClass()), InternalMessage.class);
+
+ final InternalMessage msg = converter.convert(message, _addressSpace);
+
+ enqueue(msg, CONSUMED_INSTANCE_PROPERTIES, action);
+
+ }
+
+ @Override
+ public boolean isDeleted()
+ {
+ return false;
+ }
+
private interface StandardOperation
{
String getName();
@@ -982,11 +991,9 @@
String replyTo = message.getMessageHeader().getReplyTo();
response.setInitialRoutingAddress(replyTo);
-
- getResponseDestination(replyTo).send(response,
- replyTo, InstanceProperties.EMPTY,
- new AutoCommitTransaction(_addressSpace.getMessageStore()),
- null);
+ final MessageDestination responseDestination = getResponseDestination(replyTo);
+ RoutingResult<InternalMessage> result = responseDestination.route(response, replyTo, InstanceProperties.EMPTY);
+ result.send(new AutoCommitTransaction(_addressSpace.getMessageStore()), null);
}
@@ -1533,14 +1540,13 @@
private class ConsumedMessageInstance implements MessageInstance
{
+
private final ServerMessage _message;
- private final InstanceProperties _properties;
ConsumedMessageInstance(final ServerMessage message,
final InstanceProperties properties)
{
_message = message;
- _properties = properties;
}
@Override
@@ -1741,7 +1747,7 @@
@Override
public InstanceProperties getInstanceProperties()
{
- return _properties;
+ return CONSUMED_INSTANCE_PROPERTIES;
}
@Override
diff --git a/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementNodeConsumer.java b/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementNodeConsumer.java
index 174f1c6..9e787b2 100644
--- a/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementNodeConsumer.java
+++ b/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementNodeConsumer.java
@@ -25,6 +25,7 @@
import java.util.Collections;
import java.util.List;
import java.util.Map;
+import java.util.UUID;
import org.apache.qpid.server.consumer.ConsumerTarget;
import org.apache.qpid.server.message.InstanceProperties;
@@ -32,24 +33,28 @@
import org.apache.qpid.server.message.MessageInstance;
import org.apache.qpid.server.message.MessageInstanceConsumer;
import org.apache.qpid.server.message.MessageSender;
+import org.apache.qpid.server.message.RoutingResult;
import org.apache.qpid.server.message.ServerMessage;
import org.apache.qpid.server.message.internal.InternalMessage;
import org.apache.qpid.server.model.NamedAddressSpace;
import org.apache.qpid.server.model.PublishingLink;
import org.apache.qpid.server.protocol.AMQSessionModel;
import org.apache.qpid.server.message.MessageContainer;
+import org.apache.qpid.server.queue.BaseQueue;
import org.apache.qpid.server.security.SecurityToken;
+import org.apache.qpid.server.store.MessageDurability;
+import org.apache.qpid.server.store.MessageEnqueueRecord;
import org.apache.qpid.server.store.StorableMessageMetaData;
-import org.apache.qpid.server.txn.ServerTransaction;
import org.apache.qpid.server.util.Action;
-class ManagementNodeConsumer<T extends ConsumerTarget> implements MessageInstanceConsumer<T>, MessageDestination
+class ManagementNodeConsumer<T extends ConsumerTarget> implements MessageInstanceConsumer<T>, MessageDestination,
+ BaseQueue
{
private final ManagementNode _managementNode;
private final List<ManagementResponse> _queue = Collections.synchronizedList(new ArrayList<ManagementResponse>());
private final T _target;
private final String _name;
- private final Object _identifier = new Object();
+ private final UUID _identifier = UUID.randomUUID();
public ManagementNodeConsumer(final String consumerName, final ManagementNode managementNode, T target)
@@ -142,14 +147,25 @@
}
@Override
- public <M extends ServerMessage<? extends StorableMessageMetaData>> int send(final M message,
- final String routingAddress,
- final InstanceProperties instanceProperties,
- final ServerTransaction txn,
- final Action<? super MessageInstance> postEnqueueAction)
+ public UUID getId()
{
- send((InternalMessage)message);
- return 1;
+ return _identifier;
+ }
+
+ @Override
+ public MessageDurability getMessageDurability()
+ {
+ return MessageDurability.NEVER;
+ }
+
+ @Override
+ public <M extends ServerMessage<? extends StorableMessageMetaData>> RoutingResult<M> route(final M message,
+ final String routingAddress,
+ final InstanceProperties instanceProperties)
+ {
+ RoutingResult<M> result = new RoutingResult<>(message);
+ result.addQueue(this);
+ return result;
}
@Override
@@ -181,10 +197,30 @@
return _managementNode;
}
- void send(final InternalMessage response)
+ void send(ManagementResponse responseEntry)
{
- final ManagementResponse responseEntry = new ManagementResponse(this, response);
_queue.add(responseEntry);
_target.notifyWork();
}
+
+ @Override
+ public void enqueue(final ServerMessage message,
+ final Action<? super MessageInstance> action,
+ final MessageEnqueueRecord record)
+ {
+ final InternalMessage internalMessage = (InternalMessage) message;
+ final ManagementResponse responseEntry = new ManagementResponse(this, internalMessage);
+
+ send(responseEntry);
+ if(action != null)
+ {
+ action.performAction(responseEntry);
+ }
+ }
+
+ @Override
+ public boolean isDeleted()
+ {
+ return isClosed();
+ }
}
diff --git a/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ProxyMessageSource.java b/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ProxyMessageSource.java
index 2f43170..17ab37e 100644
--- a/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ProxyMessageSource.java
+++ b/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ProxyMessageSource.java
@@ -47,6 +47,7 @@
import org.apache.qpid.server.message.MessageInstanceConsumer;
import org.apache.qpid.server.message.MessageSender;
import org.apache.qpid.server.message.MessageSource;
+import org.apache.qpid.server.message.RoutingResult;
import org.apache.qpid.server.message.ServerMessage;
import org.apache.qpid.server.model.ConfiguredObject;
import org.apache.qpid.server.model.NamedAddressSpace;
@@ -55,8 +56,6 @@
import org.apache.qpid.server.security.SecurityToken;
import org.apache.qpid.server.store.MessageDurability;
import org.apache.qpid.server.store.StorableMessageMetaData;
-import org.apache.qpid.server.txn.ServerTransaction;
-import org.apache.qpid.server.util.Action;
public class ProxyMessageSource implements MessageSource, MessageDestination
{
@@ -94,13 +93,11 @@
}
@Override
- public <M extends ServerMessage<? extends StorableMessageMetaData>> int send(final M message,
- final String routingAddress,
- final InstanceProperties instanceProperties,
- final ServerTransaction txn,
- final Action<? super MessageInstance> postEnqueueAction)
+ public <M extends ServerMessage<? extends StorableMessageMetaData>> RoutingResult<M> route(final M message,
+ final String routingAddress,
+ final InstanceProperties instanceProperties)
{
- return 0;
+ return new RoutingResult<>(message);
}
@Override
diff --git a/systests/src/test/java/org/apache/qpid/server/store/VirtualHostMessageStoreTest.java b/systests/src/test/java/org/apache/qpid/server/store/VirtualHostMessageStoreTest.java
index c883a11..7680071 100644
--- a/systests/src/test/java/org/apache/qpid/server/store/VirtualHostMessageStoreTest.java
+++ b/systests/src/test/java/org/apache/qpid/server/store/VirtualHostMessageStoreTest.java
@@ -48,6 +48,7 @@
import org.apache.qpid.server.connection.SessionPrincipal;
import org.apache.qpid.server.message.InstanceProperties;
import org.apache.qpid.server.message.MessageSource;
+import org.apache.qpid.server.message.RoutingResult;
import org.apache.qpid.server.model.Binding;
import org.apache.qpid.server.model.Broker;
import org.apache.qpid.server.model.ConfiguredObject;
@@ -615,7 +616,8 @@
ServerTransaction trans = new AutoCommitTransaction(_virtualHost.getMessageStore());
- exchange.send(currentMessage, routingKey, InstanceProperties.EMPTY, trans, null);
+ RoutingResult result = exchange.route(currentMessage, routingKey, InstanceProperties.EMPTY);
+ result.send(trans, null);
}
