diff options
Diffstat (limited to 'java/broker/src/main/java/org/apache/qpid/server/exchange')
22 files changed, 2646 insertions, 670 deletions
diff --git a/java/broker/src/main/java/org/apache/qpid/server/exchange/AbstractExchange.java b/java/broker/src/main/java/org/apache/qpid/server/exchange/AbstractExchange.java index 9ebb893362..8d24626b73 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/exchange/AbstractExchange.java +++ b/java/broker/src/main/java/org/apache/qpid/server/exchange/AbstractExchange.java @@ -191,9 +191,7 @@ public abstract class AbstractExchange implements Exchange, Managable { _exchangeMbean.unregister(); } - } - - abstract public Map<AMQShortString, List<AMQQueue>> getBindings(); + } public String toString() { diff --git a/java/broker/src/main/java/org/apache/qpid/server/exchange/DefaultExchangeFactory.java b/java/broker/src/main/java/org/apache/qpid/server/exchange/DefaultExchangeFactory.java index 636aa7eb03..9d4c090971 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/exchange/DefaultExchangeFactory.java +++ b/java/broker/src/main/java/org/apache/qpid/server/exchange/DefaultExchangeFactory.java @@ -43,8 +43,8 @@ public class DefaultExchangeFactory implements ExchangeFactory public DefaultExchangeFactory(VirtualHost host) { _host = host; - registerExchangeType(DestNameExchange.TYPE); - registerExchangeType(DestWildExchange.TYPE); + registerExchangeType(DirectExchange.TYPE); + registerExchangeType(TopicExchange.TYPE); registerExchangeType(HeadersExchange.TYPE); registerExchangeType(FanoutExchange.TYPE); } @@ -67,7 +67,7 @@ public class DefaultExchangeFactory implements ExchangeFactory if (exchType == null) { - throw new AMQUnknownExchangeType("Unknown exchange type: " + type, null); + throw new AMQUnknownExchangeType("Unknown exchange type: " + type,null); } Exchange e = exchType.newInstance(_host, exchange, durable, ticket, autoDelete); return e; diff --git a/java/broker/src/main/java/org/apache/qpid/server/exchange/DefaultExchangeRegistry.java b/java/broker/src/main/java/org/apache/qpid/server/exchange/DefaultExchangeRegistry.java index 98abf7977a..0ab8208d88 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/exchange/DefaultExchangeRegistry.java +++ b/java/broker/src/main/java/org/apache/qpid/server/exchange/DefaultExchangeRegistry.java @@ -25,6 +25,7 @@ import org.apache.qpid.AMQException; import org.apache.qpid.framing.AMQShortString; import org.apache.qpid.server.protocol.ExchangeInitialiser; import org.apache.qpid.server.queue.AMQMessage; +import org.apache.qpid.server.queue.IncomingMessage; import org.apache.qpid.server.store.MessageStore; import org.apache.qpid.server.virtualhost.VirtualHost; @@ -121,9 +122,9 @@ public class DefaultExchangeRegistry implements ExchangeRegistry * @param payload * @throws AMQException if something goes wrong delivering data */ - public void routeContent(AMQMessage payload) throws AMQException + public void routeContent(IncomingMessage payload) throws AMQException { - final AMQShortString exchange = payload.getMessagePublishInfo().getExchange(); + final AMQShortString exchange = payload.getExchange(); final Exchange exch = getExchange(exchange); // there is a small window of opportunity for the exchange to be deleted in between // the BasicPublish being received (where the exchange is validated) and the final diff --git a/java/broker/src/main/java/org/apache/qpid/server/exchange/DestWildExchange.java b/java/broker/src/main/java/org/apache/qpid/server/exchange/DestWildExchange.java deleted file mode 100644 index 6fa3686152..0000000000 --- a/java/broker/src/main/java/org/apache/qpid/server/exchange/DestWildExchange.java +++ /dev/null @@ -1,579 +0,0 @@ -/* - * - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - * - */ -package org.apache.qpid.server.exchange; - -import org.apache.log4j.Logger; -import org.apache.qpid.AMQException; -import org.apache.qpid.protocol.AMQConstant; -import org.apache.qpid.exchange.ExchangeDefaults; -import org.apache.qpid.framing.AMQShortString; -import org.apache.qpid.framing.FieldTable; -import org.apache.qpid.framing.AMQShortStringTokenizer; -import org.apache.qpid.framing.abstraction.MessagePublishInfo; -import org.apache.qpid.server.management.MBeanConstructor; -import org.apache.qpid.server.management.MBeanDescription; -import org.apache.qpid.server.queue.AMQMessage; -import org.apache.qpid.server.queue.AMQQueue; -import org.apache.qpid.server.virtualhost.VirtualHost; - -import javax.management.JMException; -import javax.management.MBeanException; -import javax.management.openmbean.CompositeData; -import javax.management.openmbean.CompositeDataSupport; -import javax.management.openmbean.OpenDataException; -import javax.management.openmbean.TabularData; -import javax.management.openmbean.TabularDataSupport; -import java.util.*; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.CopyOnWriteArrayList; - -public class DestWildExchange extends AbstractExchange -{ - - public static final ExchangeType<DestWildExchange> TYPE = new ExchangeType<DestWildExchange>() - { - - public AMQShortString getName() - { - return ExchangeDefaults.TOPIC_EXCHANGE_CLASS; - } - - public Class<DestWildExchange> getExchangeClass() - { - return DestWildExchange.class; - } - - public DestWildExchange newInstance(VirtualHost host, - AMQShortString name, - boolean durable, - int ticket, - boolean autoDelete) throws AMQException - { - DestWildExchange exch = new DestWildExchange(); - exch.initialise(host, name, durable, ticket, autoDelete); - return exch; - } - - public AMQShortString getDefaultExchangeName() - { - return ExchangeDefaults.TOPIC_EXCHANGE_NAME; - } - }; - - - private static final Logger _logger = Logger.getLogger(DestWildExchange.class); - - private final ConcurrentHashMap<AMQShortString, List<AMQQueue>> _bindingKey2queues = - new ConcurrentHashMap<AMQShortString, List<AMQQueue>>(); - private final ConcurrentHashMap<AMQShortString, List<AMQQueue>> _simpleBindingKey2queues = - new ConcurrentHashMap<AMQShortString, List<AMQQueue>>(); - private final ConcurrentHashMap<AMQShortString, List<AMQQueue>> _wildCardBindingKey2queues = - new ConcurrentHashMap<AMQShortString, List<AMQQueue>>(); - - private static final byte TOPIC_SEPARATOR = (byte)'.'; - private static final AMQShortString TOPIC_SEPARATOR_AS_SHORTSTRING = new AMQShortString("."); - private static final AMQShortString AMQP_STAR_TOKEN = new AMQShortString("*"); - private static final AMQShortString AMQP_HASH_TOKEN = new AMQShortString("#"); - private ConcurrentHashMap<AMQShortString, AMQShortString[]> _bindingKey2Tokenized = - new ConcurrentHashMap<AMQShortString, AMQShortString[]>(); - private static final byte HASH_BYTE = (byte)'#'; - private static final byte STAR_BYTE = (byte)'*'; - - /** DestWildExchangeMBean class implements the management interface for the Topic exchanges. */ - @MBeanDescription("Management Bean for Topic Exchange") - private final class DestWildExchangeMBean extends ExchangeMBean - { - @MBeanConstructor("Creates an MBean for AMQ topic exchange") - public DestWildExchangeMBean() throws JMException - { - super(); - _exchangeType = "topic"; - init(); - } - - /** returns exchange bindings in tabular form */ - public TabularData bindings() throws OpenDataException - { - _bindingList = new TabularDataSupport(_bindinglistDataType); - for (Map.Entry<AMQShortString, List<AMQQueue>> entry : _bindingKey2queues.entrySet()) - { - AMQShortString key = entry.getKey(); - List<String> queueList = new ArrayList<String>(); - - List<AMQQueue> queues = getMatchedQueues(key); - for (AMQQueue q : queues) - { - queueList.add(q.getName().toString()); - } - - Object[] bindingItemValues = {key.toString(), queueList.toArray(new String[0])}; - CompositeData bindingData = new CompositeDataSupport(_bindingDataType, _bindingItemNames, bindingItemValues); - _bindingList.put(bindingData); - } - - return _bindingList; - } - - public void createNewBinding(String queueName, String binding) throws JMException - { - AMQQueue queue = getQueueRegistry().getQueue(new AMQShortString(queueName)); - if (queue == null) - { - throw new JMException("Queue \"" + queueName + "\" is not registered with the exchange."); - } - - try - { - queue.bind(new AMQShortString(binding), null, DestWildExchange.this); - } - catch (AMQException ex) - { - throw new MBeanException(ex); - } - } - - } // End of MBean class - - public AMQShortString getType() - { - return ExchangeDefaults.TOPIC_EXCHANGE_CLASS; - } - - public synchronized void registerQueue(AMQShortString rKey, AMQQueue queue, FieldTable args) throws AMQException - { - assert queue != null; - assert rKey != null; - - _logger.debug("Registering queue " + queue.getName() + " with routing key " + rKey); - - // we need to use putIfAbsent, which is an atomic operation, to avoid a race condition - List<AMQQueue> queueList = _bindingKey2queues.putIfAbsent(rKey, new CopyOnWriteArrayList<AMQQueue>()); - - - - - - - - // if we got null back, no previous value was associated with the specified routing key hence - // we need to read back the new value just put into the map - if (queueList == null) - { - queueList = _bindingKey2queues.get(rKey); - } - - - - if (!queueList.contains(queue)) - { - queueList.add(queue); - - - if(rKey.contains(HASH_BYTE) || rKey.contains(STAR_BYTE)) - { - AMQShortString routingKey = normalize(rKey); - List<AMQQueue> queueList2 = _wildCardBindingKey2queues.putIfAbsent(routingKey, new CopyOnWriteArrayList<AMQQueue>()); - - if(queueList2 == null) - { - queueList2 = _wildCardBindingKey2queues.get(routingKey); - AMQShortStringTokenizer keyTok = routingKey.tokenize(TOPIC_SEPARATOR); - - ArrayList<AMQShortString> keyTokList = new ArrayList<AMQShortString>(keyTok.countTokens()); - - while (keyTok.hasMoreTokens()) - { - keyTokList.add(keyTok.nextToken()); - } - - _bindingKey2Tokenized.put(routingKey, keyTokList.toArray(new AMQShortString[keyTokList.size()])); - } - queueList2.add(queue); - - } - else - { - List<AMQQueue> queueList2 = _simpleBindingKey2queues.putIfAbsent(rKey, new CopyOnWriteArrayList<AMQQueue>()); - if(queueList2 == null) - { - queueList2 = _simpleBindingKey2queues.get(rKey); - } - queueList2.add(queue); - - } - - - - - } - else if (_logger.isDebugEnabled()) - { - _logger.debug("Queue " + queue + " is already registered with routing key " + rKey); - } - - - - } - - private AMQShortString normalize(AMQShortString routingKey) - { - if(routingKey == null) - { - routingKey = AMQShortString.EMPTY_STRING; - } - - AMQShortStringTokenizer routingTokens = routingKey.tokenize(TOPIC_SEPARATOR); - - List<AMQShortString> subscriptionList = new ArrayList<AMQShortString>(); - - while (routingTokens.hasMoreTokens()) - { - subscriptionList.add(routingTokens.nextToken()); - } - - int size = subscriptionList.size(); - - for (int index = 0; index < size; index++) - { - // if there are more levels - if ((index + 1) < size) - { - if (subscriptionList.get(index).equals(AMQP_HASH_TOKEN)) - { - if (subscriptionList.get(index + 1).equals(AMQP_HASH_TOKEN)) - { - // we don't need #.# delete this one - subscriptionList.remove(index); - size--; - // redo this normalisation - index--; - } - - if (subscriptionList.get(index + 1).equals(AMQP_STAR_TOKEN)) - { - // we don't want #.* swap to *.# - // remove it and put it in at index + 1 - subscriptionList.add(index + 1, subscriptionList.remove(index)); - } - } - } // if we have more levels - } - - - - AMQShortString normalizedString = AMQShortString.join(subscriptionList, TOPIC_SEPARATOR_AS_SHORTSTRING); - - return normalizedString; - } - - public void route(AMQMessage payload) throws AMQException - { - MessagePublishInfo info = payload.getMessagePublishInfo(); - - final AMQShortString routingKey = info.getRoutingKey(); - - List<AMQQueue> queues = getMatchedQueues(routingKey); - // if we have no registered queues we have nothing to do - // TODO: add support for the immediate flag - if ((queues == null) || queues.isEmpty()) - { - if (info.isMandatory() || info.isImmediate()) - { - String msg = "Topic " + routingKey + " is not known to " + this; - throw new NoRouteException(msg, payload); - } - else - { - _logger.warn("No queues found for routing key " + routingKey); - _logger.warn("Routing map contains: " + _bindingKey2queues); - - return; - } - } - - payload.enqueue(queues); - - } - - public boolean isBound(AMQShortString routingKey, FieldTable arguments, AMQQueue queue) - { - return isBound(routingKey, queue); - } - - public boolean isBound(AMQShortString routingKey, AMQQueue queue) - { - List<AMQQueue> queues = _bindingKey2queues.get(normalize(routingKey)); - - return (queues != null) && queues.contains(queue); - } - - public boolean isBound(AMQShortString routingKey) - { - List<AMQQueue> queues = _bindingKey2queues.get(normalize(routingKey)); - - return (queues != null) && !queues.isEmpty(); - } - - public boolean isBound(AMQQueue queue) - { - for (List<AMQQueue> queues : _bindingKey2queues.values()) - { - if (queues.contains(queue)) - { - return true; - } - } - - return false; - } - - public boolean hasBindings() - { - return !_bindingKey2queues.isEmpty(); - } - - public synchronized void deregisterQueue(AMQShortString rKey, AMQQueue queue, FieldTable args) throws AMQException - { - assert queue != null; - assert rKey != null; - - List<AMQQueue> queues = _bindingKey2queues.get(rKey); - if (queues == null) - { - throw new AMQException(AMQConstant.NOT_FOUND, "Queue " + queue + " was not registered with exchange " + this.getName() - + " with routing key " + rKey + ". No queue was registered with that _routing key"); - - } - - boolean removedQ = queues.remove(queue); - if (!removedQ) - { - throw new AMQException(AMQConstant.NOT_FOUND, "Queue " + queue + " was not registered with exchange " + this.getName() - + " with routing key " + rKey); - } - - - if(rKey.contains(HASH_BYTE) || rKey.contains(STAR_BYTE)) - { - AMQShortString bindingKey = normalize(rKey); - List<AMQQueue> queues2 = _wildCardBindingKey2queues.get(bindingKey); - queues2.remove(queue); - if(queues2.isEmpty()) - { - _wildCardBindingKey2queues.remove(bindingKey); - _bindingKey2Tokenized.remove(bindingKey); - } - - } - else - { - List<AMQQueue> queues2 = _simpleBindingKey2queues.get(rKey); - queues2.remove(queue); - if(queues2.isEmpty()) - { - _simpleBindingKey2queues.remove(rKey); - } - - } - - - - - if (queues.isEmpty()) - { - _bindingKey2queues.remove(rKey); - } - } - - protected ExchangeMBean createMBean() throws AMQException - { - try - { - return new DestWildExchangeMBean(); - } - catch (JMException ex) - { - _logger.error("Exception occured in creating the topic exchenge mbean", ex); - throw new AMQException("Exception occured in creating the topic exchenge mbean", ex); - } - } - - public Map<AMQShortString, List<AMQQueue>> getBindings() - { - return _bindingKey2queues; - } - - private List<AMQQueue> getMatchedQueues(AMQShortString routingKey) - { - - List<AMQQueue> list = null; - - if(!_wildCardBindingKey2queues.isEmpty()) - { - - - AMQShortStringTokenizer routingTokens = routingKey.tokenize(TOPIC_SEPARATOR); - - final int routingTokensCount = routingTokens.countTokens(); - - - AMQShortString[] routingkeyTokens = new AMQShortString[routingTokensCount]; - - if(routingTokensCount == 1) - { - routingkeyTokens[0] =routingKey; - } - else - { - - - int token = 0; - while (routingTokens.hasMoreTokens()) - { - - AMQShortString next = routingTokens.nextToken(); - - routingkeyTokens[token++] = next; - } - } - for (AMQShortString bindingKey : _wildCardBindingKey2queues.keySet()) - { - - AMQShortString[] bindingKeyTokens = _bindingKey2Tokenized.get(bindingKey); - - - boolean matching = true; - boolean done = false; - - int depthPlusRoutingSkip = 0; - int depthPlusQueueSkip = 0; - - final int bindingKeyTokensCount = bindingKeyTokens.length; - - while (matching && !done) - { - - if ((bindingKeyTokensCount == depthPlusQueueSkip) || (routingTokensCount == depthPlusRoutingSkip)) - { - done = true; - - // if it was the routing key that ran out of digits - if (routingTokensCount == depthPlusRoutingSkip) - { - if (bindingKeyTokensCount > depthPlusQueueSkip) - { // a hash and it is the last entry - matching = - bindingKeyTokens[depthPlusQueueSkip].equals(AMQP_HASH_TOKEN) - && (bindingKeyTokensCount == (depthPlusQueueSkip + 1)); - } - } - else if (routingTokensCount > depthPlusRoutingSkip) - { - // There is still more routing key to check - matching = false; - } - - continue; - } - - // if the values on the two topics don't match - if (!bindingKeyTokens[depthPlusQueueSkip].equals(routingkeyTokens[depthPlusRoutingSkip])) - { - if (bindingKeyTokens[depthPlusQueueSkip].equals(AMQP_STAR_TOKEN)) - { - depthPlusQueueSkip++; - depthPlusRoutingSkip++; - - continue; - } - else if (bindingKeyTokens[depthPlusQueueSkip].equals(AMQP_HASH_TOKEN)) - { - // Is this a # at the end - if (bindingKeyTokensCount == (depthPlusQueueSkip + 1)) - { - done = true; - - continue; - } - - // otherwise # in the middle - while (routingTokensCount > depthPlusRoutingSkip) - { - if (routingkeyTokens[depthPlusRoutingSkip].equals(bindingKeyTokens[depthPlusQueueSkip + 1])) - { - depthPlusQueueSkip += 2; - depthPlusRoutingSkip++; - - break; - } - - depthPlusRoutingSkip++; - } - - continue; - } - - matching = false; - } - - depthPlusQueueSkip++; - depthPlusRoutingSkip++; - } - - if (matching) - { - if(list == null) - { - list = new ArrayList<AMQQueue>(_wildCardBindingKey2queues.get(bindingKey)); - } - else - { - list.addAll(_wildCardBindingKey2queues.get(bindingKey)); - } - } - } - - } - if(!_simpleBindingKey2queues.isEmpty()) - { - List<AMQQueue> queues = _simpleBindingKey2queues.get(routingKey); - if(list == null) - { - if(queues == null) - { - list = Collections.EMPTY_LIST; - } - else - { - list = new ArrayList<AMQQueue>(queues); - } - } - else if(queues != null) - { - list.addAll(queues); - } - - } - - return list; - - } -} diff --git a/java/broker/src/main/java/org/apache/qpid/server/exchange/DestNameExchange.java b/java/broker/src/main/java/org/apache/qpid/server/exchange/DirectExchange.java index 12347c0278..5dcc2cf143 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/exchange/DestNameExchange.java +++ b/java/broker/src/main/java/org/apache/qpid/server/exchange/DirectExchange.java @@ -38,23 +38,22 @@ import org.apache.qpid.protocol.AMQConstant; import org.apache.qpid.exchange.ExchangeDefaults; import org.apache.qpid.framing.AMQShortString; import org.apache.qpid.framing.FieldTable; -import org.apache.qpid.framing.abstraction.MessagePublishInfo; import org.apache.qpid.server.management.MBeanConstructor; import org.apache.qpid.server.management.MBeanDescription; -import org.apache.qpid.server.queue.AMQMessage; +import org.apache.qpid.server.queue.IncomingMessage; import org.apache.qpid.server.queue.AMQQueue; import org.apache.qpid.server.virtualhost.VirtualHost; -public class DestNameExchange extends AbstractExchange +public class DirectExchange extends AbstractExchange { - private static final Logger _logger = Logger.getLogger(DestNameExchange.class); + private static final Logger _logger = Logger.getLogger(DirectExchange.class); /** * Maps from queue name to queue instances */ private final Index _index = new Index(); - public static final ExchangeType<DestNameExchange> TYPE = new ExchangeType<DestNameExchange>() + public static final ExchangeType<DirectExchange> TYPE = new ExchangeType<DirectExchange>() { public AMQShortString getName() @@ -62,18 +61,18 @@ public class DestNameExchange extends AbstractExchange return ExchangeDefaults.DIRECT_EXCHANGE_CLASS; } - public Class<DestNameExchange> getExchangeClass() + public Class<DirectExchange> getExchangeClass() { - return DestNameExchange.class; + return DirectExchange.class; } - public DestNameExchange newInstance(VirtualHost host, + public DirectExchange newInstance(VirtualHost host, AMQShortString name, boolean durable, int ticket, boolean autoDelete) throws AMQException { - DestNameExchange exch = new DestNameExchange(); + DirectExchange exch = new DirectExchange(); exch.initialise(host,name,durable,ticket,autoDelete); return exch; } @@ -88,10 +87,10 @@ public class DestNameExchange extends AbstractExchange * MBean class implementing the management interfaces. */ @MBeanDescription("Management Bean for Direct Exchange") - private final class DestNameExchangeMBean extends ExchangeMBean + private final class DirectExchangeMBean extends ExchangeMBean { @MBeanConstructor("Creates an MBean for AMQ direct exchange") - public DestNameExchangeMBean() throws JMException + public DirectExchangeMBean() throws JMException { super(); _exchangeType = "direct"; @@ -132,7 +131,7 @@ public class DestNameExchange extends AbstractExchange try { - queue.bind(new AMQShortString(binding), null, DestNameExchange.this); + queue.bind(DirectExchange.this, new AMQShortString(binding), null); } catch (AMQException ex) { @@ -147,7 +146,7 @@ public class DestNameExchange extends AbstractExchange { try { - return new DestNameExchangeMBean(); + return new DirectExchangeMBean(); } catch (JMException ex) { @@ -187,35 +186,21 @@ public class DestNameExchange extends AbstractExchange } } - public void route(AMQMessage payload) throws AMQException + public void route(IncomingMessage payload) throws AMQException { - final MessagePublishInfo info = payload.getMessagePublishInfo(); - final AMQShortString routingKey = info.getRoutingKey() == null ? AMQShortString.EMPTY_STRING : info.getRoutingKey(); + + final AMQShortString routingKey = payload.getRoutingKey() == null ? AMQShortString.EMPTY_STRING : payload.getRoutingKey(); + final List<AMQQueue> queues = (routingKey == null) ? null : _index.get(routingKey); - if (queues == null || queues.isEmpty()) + + if (_logger.isDebugEnabled()) { - String msg = "Routing key " + routingKey + " is not known to " + this; - if (info.isMandatory() || info.isImmediate()) - { - throw new NoRouteException(msg, payload); - } - else - { - _logger.error("MESSAGE LOSS: Message should be sent on a Dead Letter Queue"); - _logger.warn(msg); - } + _logger.debug("Publishing message to queue " + queues); } - else - { - if (_logger.isDebugEnabled()) - { - _logger.debug("Publishing message to queue " + queues); - } payload.enqueue(queues); - } } public boolean isBound(AMQShortString routingKey, FieldTable arguments, AMQQueue queue) diff --git a/java/broker/src/main/java/org/apache/qpid/server/exchange/Exchange.java b/java/broker/src/main/java/org/apache/qpid/server/exchange/Exchange.java index 37cd85a8f8..06209c5458 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/exchange/Exchange.java +++ b/java/broker/src/main/java/org/apache/qpid/server/exchange/Exchange.java @@ -23,7 +23,8 @@ package org.apache.qpid.server.exchange; import org.apache.qpid.AMQException; import org.apache.qpid.framing.AMQShortString; import org.apache.qpid.framing.FieldTable; -import org.apache.qpid.server.queue.AMQMessage; + +import org.apache.qpid.server.queue.IncomingMessage; import org.apache.qpid.server.queue.AMQQueue; import org.apache.qpid.server.virtualhost.VirtualHost; @@ -53,7 +54,7 @@ public interface Exchange void deregisterQueue(AMQShortString routingKey, AMQQueue queue, FieldTable args) throws AMQException; - void route(AMQMessage message) throws AMQException; + void route(IncomingMessage message) throws AMQException; /** @@ -92,6 +93,6 @@ public interface Exchange */ boolean hasBindings(); - Map<AMQShortString, List<AMQQueue>> getBindings(); + } diff --git a/java/broker/src/main/java/org/apache/qpid/server/exchange/FanoutExchange.java b/java/broker/src/main/java/org/apache/qpid/server/exchange/FanoutExchange.java index f1b383eac9..e9fd4d548b 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/exchange/FanoutExchange.java +++ b/java/broker/src/main/java/org/apache/qpid/server/exchange/FanoutExchange.java @@ -26,10 +26,9 @@ import org.apache.qpid.protocol.AMQConstant; import org.apache.qpid.exchange.ExchangeDefaults; import org.apache.qpid.framing.AMQShortString; import org.apache.qpid.framing.FieldTable; -import org.apache.qpid.framing.abstraction.MessagePublishInfo; import org.apache.qpid.server.management.MBeanConstructor; import org.apache.qpid.server.management.MBeanDescription; -import org.apache.qpid.server.queue.AMQMessage; +import org.apache.qpid.server.queue.IncomingMessage; import org.apache.qpid.server.queue.AMQQueue; import org.apache.qpid.server.virtualhost.VirtualHost; @@ -95,7 +94,7 @@ public class FanoutExchange extends AbstractExchange try { - queue.bind(new AMQShortString(binding), null, FanoutExchange.this); + queue.bind(FanoutExchange.this, new AMQShortString(binding), null); } catch (AMQException ex) { @@ -183,32 +182,17 @@ public class FanoutExchange extends AbstractExchange } } - public void route(AMQMessage payload) throws AMQException + public void route(IncomingMessage payload) throws AMQException { - final MessagePublishInfo publishInfo = payload.getMessagePublishInfo(); - final AMQShortString routingKey = publishInfo.getRoutingKey(); - if ((_queues == null) || _queues.isEmpty()) + + + if (_logger.isDebugEnabled()) { - String msg = "No queues bound to " + this; - if (publishInfo.isMandatory() || publishInfo.isImmediate()) - { - throw new NoRouteException(msg, payload); - } - else - { - _logger.warn(msg); - } + _logger.debug("Publishing message to queue " + _queues); } - else - { - if (_logger.isDebugEnabled()) - { - _logger.debug("Publishing message to queue " + _queues); - } - payload.enqueue(new ArrayList(_queues)); + payload.enqueue(new ArrayList(_queues)); - } } public boolean isBound(AMQShortString routingKey, FieldTable arguments, AMQQueue queue) diff --git a/java/broker/src/main/java/org/apache/qpid/server/exchange/HeadersExchange.java b/java/broker/src/main/java/org/apache/qpid/server/exchange/HeadersExchange.java index 68ad88c4cb..d1bea3410b 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/exchange/HeadersExchange.java +++ b/java/broker/src/main/java/org/apache/qpid/server/exchange/HeadersExchange.java @@ -31,7 +31,7 @@ import org.apache.qpid.framing.ContentHeaderBody; import org.apache.qpid.framing.FieldTable; import org.apache.qpid.server.management.MBeanConstructor; import org.apache.qpid.server.management.MBeanDescription; -import org.apache.qpid.server.queue.AMQMessage; +import org.apache.qpid.server.queue.IncomingMessage; import org.apache.qpid.server.queue.AMQQueue; import org.apache.qpid.server.virtualhost.VirtualHost; @@ -50,6 +50,7 @@ import java.util.ArrayList; import java.util.Iterator; import java.util.List; import java.util.Map; +import java.util.Collection; import java.util.concurrent.CopyOnWriteArrayList; /** @@ -240,7 +241,7 @@ public class HeadersExchange extends AbstractExchange } } - public void route(AMQMessage payload) throws AMQException + public void route(IncomingMessage payload) throws AMQException { FieldTable headers = getHeaders(payload.getContentHeaderBody()); if (_logger.isDebugEnabled()) @@ -248,8 +249,10 @@ public class HeadersExchange extends AbstractExchange _logger.debug("Exchange " + getName() + ": routing message with headers " + headers); } boolean routed = false; + Collection<AMQQueue> queues = new ArrayList<AMQQueue>(); for (Registration e : _bindings) { + if (e.binding.matches(headers)) { if (_logger.isDebugEnabled()) @@ -257,25 +260,12 @@ public class HeadersExchange extends AbstractExchange _logger.debug("Exchange " + getName() + ": delivering message with headers " + headers + " to " + e.queue.getName()); } - payload.enqueue(e.queue); - routed = true; - } - } - if (!routed) - { - - String msg = "Exchange " + getName() + ": message not routable."; + queues.add(e.queue); - if (payload.getMessagePublishInfo().isMandatory() || payload.getMessagePublishInfo().isImmediate()) - { - throw new NoRouteException(msg, payload); - } - else - { - _logger.warn(msg); + routed = true; } - } + payload.enqueue(queues); } public boolean isBound(AMQShortString routingKey, FieldTable arguments, AMQQueue queue) diff --git a/java/broker/src/main/java/org/apache/qpid/server/exchange/Index.java b/java/broker/src/main/java/org/apache/qpid/server/exchange/Index.java index eacdad8a8e..4f1f550e94 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/exchange/Index.java +++ b/java/broker/src/main/java/org/apache/qpid/server/exchange/Index.java @@ -32,7 +32,7 @@ import org.apache.qpid.server.queue.AMQQueue; /** * An index of queues against routing key. Allows multiple queues to be stored - * against the same key. Used in the DestNameExchange. + * against the same key. Used in the DirectExchange. */ class Index { diff --git a/java/broker/src/main/java/org/apache/qpid/server/exchange/MessageRouter.java b/java/broker/src/main/java/org/apache/qpid/server/exchange/MessageRouter.java index 7508e80f7f..db9beb6da7 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/exchange/MessageRouter.java +++ b/java/broker/src/main/java/org/apache/qpid/server/exchange/MessageRouter.java @@ -22,6 +22,7 @@ package org.apache.qpid.server.exchange; import org.apache.qpid.AMQException; import org.apache.qpid.server.queue.AMQMessage; +import org.apache.qpid.server.queue.IncomingMessage; /** * Separated out from the ExchangeRegistry interface to allow components @@ -36,5 +37,5 @@ public interface MessageRouter * * @throws org.apache.qpid.AMQException if something goes wrong delivering data */ - void routeContent(AMQMessage message) throws AMQException; + void routeContent(IncomingMessage message) throws AMQException; } diff --git a/java/broker/src/main/java/org/apache/qpid/server/exchange/NoRouteException.java b/java/broker/src/main/java/org/apache/qpid/server/exchange/NoRouteException.java index 1d6ab3842d..d18ad7ab14 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/exchange/NoRouteException.java +++ b/java/broker/src/main/java/org/apache/qpid/server/exchange/NoRouteException.java @@ -23,6 +23,7 @@ package org.apache.qpid.server.exchange; import org.apache.qpid.protocol.AMQConstant; import org.apache.qpid.server.RequiredDeliveryException; import org.apache.qpid.server.queue.AMQMessage; +import org.apache.qpid.server.queue.IncomingMessage; /** * NoRouteException is a {@link RequiredDeliveryException} that represents the failure case where a manadatory message @@ -36,9 +37,9 @@ import org.apache.qpid.server.queue.AMQMessage; */ public class NoRouteException extends RequiredDeliveryException { - public NoRouteException(String msg, AMQMessage message) + public NoRouteException(String msg, AMQMessage amqMessage) { - super(msg, message); + super(msg, amqMessage); } public AMQConstant getReplyCode() diff --git a/java/broker/src/main/java/org/apache/qpid/server/exchange/TopicExchange.java b/java/broker/src/main/java/org/apache/qpid/server/exchange/TopicExchange.java new file mode 100644 index 0000000000..d07501a188 --- /dev/null +++ b/java/broker/src/main/java/org/apache/qpid/server/exchange/TopicExchange.java @@ -0,0 +1,651 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + * + */ +package org.apache.qpid.server.exchange; + +import org.apache.log4j.Logger; +import org.apache.qpid.AMQException; +import org.apache.qpid.common.AMQPFilterTypes; +import org.apache.qpid.protocol.AMQConstant; +import org.apache.qpid.exchange.ExchangeDefaults; +import org.apache.qpid.framing.AMQShortString; +import org.apache.qpid.framing.FieldTable; +import org.apache.qpid.framing.AMQShortStringTokenizer; +import org.apache.qpid.server.management.MBeanConstructor; +import org.apache.qpid.server.management.MBeanDescription; +import org.apache.qpid.server.queue.IncomingMessage; +import org.apache.qpid.server.queue.AMQQueue; +import org.apache.qpid.server.queue.AMQMessage; +import org.apache.qpid.server.virtualhost.VirtualHost; +import org.apache.qpid.server.exchange.topic.TopicParser; +import org.apache.qpid.server.exchange.topic.TopicMatcherResult; +import org.apache.qpid.server.filter.MessageFilter; +import org.apache.qpid.server.filter.JMSSelectorFilter; + +import javax.management.JMException; +import javax.management.MBeanException; +import javax.management.openmbean.CompositeData; +import javax.management.openmbean.CompositeDataSupport; +import javax.management.openmbean.OpenDataException; +import javax.management.openmbean.TabularData; +import javax.management.openmbean.TabularDataSupport; +import java.util.*; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.CopyOnWriteArraySet; +import java.util.concurrent.atomic.AtomicInteger; +import java.lang.ref.WeakReference; + +public class TopicExchange extends AbstractExchange +{ + + public static final ExchangeType<TopicExchange> TYPE = new ExchangeType<TopicExchange>() + { + + public AMQShortString getName() + { + return ExchangeDefaults.TOPIC_EXCHANGE_CLASS; + } + + public Class<TopicExchange> getExchangeClass() + { + return TopicExchange.class; + } + + public TopicExchange newInstance(VirtualHost host, + AMQShortString name, + boolean durable, + int ticket, + boolean autoDelete) throws AMQException + { + TopicExchange exch = new TopicExchange(); + exch.initialise(host, name, durable, ticket, autoDelete); + return exch; + } + + public AMQShortString getDefaultExchangeName() + { + return ExchangeDefaults.TOPIC_EXCHANGE_NAME; + } + }; + + + private static final Logger _logger = Logger.getLogger(TopicExchange.class); + +/* + private final ConcurrentHashMap<AMQShortString, List<AMQQueue>> _bindingKey2queues = + new ConcurrentHashMap<AMQShortString, List<AMQQueue>>(); + private final ConcurrentHashMap<AMQShortString, List<AMQQueue>> _simpleBindingKey2queues = + new ConcurrentHashMap<AMQShortString, List<AMQQueue>>(); + private final ConcurrentHashMap<AMQShortString, List<AMQQueue>> _wildCardBindingKey2queues = + new ConcurrentHashMap<AMQShortString, List<AMQQueue>>(); +*/ + // private ConcurrentHashMap<AMQShortString, AMQQueue> _routingKey2queue = new ConcurrentHashMap<AMQShortString, AMQQueue>(); + private static final byte TOPIC_SEPARATOR = (byte)'.'; + private static final AMQShortString TOPIC_SEPARATOR_AS_SHORTSTRING = new AMQShortString("."); + private static final AMQShortString AMQP_STAR_TOKEN = new AMQShortString("*"); + private static final AMQShortString AMQP_HASH_TOKEN = new AMQShortString("#"); + + private static final byte HASH_BYTE = (byte)'#'; + private static final byte STAR_BYTE = (byte)'*'; + + private final TopicParser _parser = new TopicParser(); + + private final Map<AMQShortString, TopicExchangeResult> _topicExchangeResults = + new ConcurrentHashMap<AMQShortString, TopicExchangeResult>(); + + private final Map<Binding, FieldTable> _bindings = new HashMap<Binding, FieldTable>(); + + private final Map<String, WeakReference<JMSSelectorFilter<RuntimeException>>> _selectorCache = new WeakHashMap<String, WeakReference<JMSSelectorFilter<RuntimeException>>>(); + + public static class Binding + { + private final AMQShortString _bindingKey; + private final AMQQueue _queue; + + public Binding(AMQShortString bindingKey, AMQQueue queue) + { + _bindingKey = bindingKey; + _queue = queue; + } + + public AMQShortString getBindingKey() + { + return _bindingKey; + } + + public AMQQueue getQueue() + { + return _queue; + } + + public int hashCode() + { + return (_bindingKey == null ? 1 : _bindingKey.hashCode())*31 + _queue.hashCode(); + } + + public boolean equals(Object o) + { + if(this == o) + { + return true; + } + if(o instanceof Binding) + { + Binding other = (Binding) o; + return (_queue == other._queue) + && ((_bindingKey == null) ? other._bindingKey == null : _bindingKey.equals(other._bindingKey)); + } + return false; + } + } + + + + private final class TopicExchangeResult implements TopicMatcherResult + { + private final Map<AMQQueue, Integer> _unfilteredQueues = new ConcurrentHashMap<AMQQueue, Integer>(); + private final ConcurrentHashMap<AMQQueue, Map<MessageFilter<RuntimeException>,Integer>> _filteredQueues = new ConcurrentHashMap<AMQQueue, Map<MessageFilter<RuntimeException>, Integer>>(); + + public void addUnfilteredQueue(AMQQueue queue) + { + Integer instances = _unfilteredQueues.get(queue); + if(instances == null) + { + _unfilteredQueues.put(queue, 1); + } + else + { + _unfilteredQueues.put(queue, instances + 1); + } + } + + public void removeUnfilteredQueue(AMQQueue queue) + { + Integer instances = _unfilteredQueues.get(queue); + if(instances == 1) + { + _unfilteredQueues.remove(queue); + } + else + { + _unfilteredQueues.put(queue,instances - 1); + } + + } + + + public void addFilteredQueue(AMQQueue queue, MessageFilter<RuntimeException> filter) + { + Map<MessageFilter<RuntimeException>,Integer> filters = _filteredQueues.get(queue); + if(filters == null) + { + filters = new ConcurrentHashMap<MessageFilter<RuntimeException>,Integer>(); + _filteredQueues.put(queue, filters); + } + Integer instances = filters.get(filter); + if(instances == null) + { + filters.put(filter,1); + } + else + { + filters.put(filter, instances + 1); + } + + } + + public void removeFilteredQueue(AMQQueue queue, MessageFilter<RuntimeException> filter) + { + Map<MessageFilter<RuntimeException>,Integer> filters = _filteredQueues.get(queue); + if(filters != null) + { + Integer instances = filters.get(filter); + if(instances == 1) + { + filters.remove(filter); + if(filters.isEmpty()) + { + _filteredQueues.remove(queue); + } + } + else if(instances != null) + { + filters.put(filter, instances - 1); + } + + } + + } + + public void replaceQueueFilter(AMQQueue queue, + MessageFilter<RuntimeException> oldFilter, + MessageFilter<RuntimeException> newFilter) + { + Map<MessageFilter<RuntimeException>,Integer> filters = _filteredQueues.get(queue); + Map<MessageFilter<RuntimeException>,Integer> newFilters = new ConcurrentHashMap<MessageFilter<RuntimeException>,Integer>(filters); + Integer oldFilterInstances = filters.get(oldFilter); + if(oldFilterInstances == 1) + { + newFilters.remove(oldFilter); + } + else + { + newFilters.put(oldFilter, oldFilterInstances-1); + } + Integer newFilterInstances = filters.get(newFilter); + if(newFilterInstances == null) + { + newFilters.put(newFilter, 1); + } + else + { + newFilters.put(newFilter, newFilterInstances+1); + } + _filteredQueues.put(queue,newFilters); + } + + public Collection<AMQQueue> processMessage(IncomingMessage msg, Collection<AMQQueue> queues) + { + if(queues == null) + { + if(_filteredQueues.isEmpty()) + { + return new ArrayList<AMQQueue>(_unfilteredQueues.keySet()); + } + else + { + queues = new HashSet<AMQQueue>(); + } + } + else if(!(queues instanceof Set)) + { + queues = new HashSet<AMQQueue>(queues); + } + + queues.addAll(_unfilteredQueues.keySet()); + if(!_filteredQueues.isEmpty()) + { + for(Map.Entry<AMQQueue, Map<MessageFilter<RuntimeException>, Integer>> entry : _filteredQueues.entrySet()) + { + if(!queues.contains(entry.getKey())) + { + for(MessageFilter<RuntimeException> filter : entry.getValue().keySet()) + { + if(filter.matches(msg)) + { + queues.add(entry.getKey()); + } + } + } + } + } + return queues; + } + + } + + + /** TopicExchangeMBean class implements the management interface for the Topic exchanges. */ + @MBeanDescription("Management Bean for Topic Exchange") + private final class TopicExchangeMBean extends ExchangeMBean + { + @MBeanConstructor("Creates an MBean for AMQ topic exchange") + public TopicExchangeMBean() throws JMException + { + super(); + _exchangeType = "topic"; + init(); + } + + /** returns exchange bindings in tabular form */ + public TabularData bindings() throws OpenDataException + { + _bindingList = new TabularDataSupport(_bindinglistDataType); + Map<String, List<String>> bindingData = new HashMap<String, List<String>>(); + for (Binding binding : _bindings.keySet()) + { + String key = binding.getBindingKey().toString(); + List<String> queueNames = bindingData.get(key); + if(queueNames == null) + { + queueNames = new ArrayList<String>(); + bindingData.put(key, queueNames); + } + queueNames.add(binding.getQueue().getName().toString()); + + } + for(Map.Entry<String, List<String>> entry : bindingData.entrySet()) + { + Object[] bindingItemValues = {entry.getKey(), entry.getValue().toArray(new String[entry.getValue().size()]) }; + CompositeData bindingCompositeData = new CompositeDataSupport(_bindingDataType, _bindingItemNames, bindingItemValues); + _bindingList.put(bindingCompositeData); + } + + return _bindingList; + } + + public void createNewBinding(String queueName, String binding) throws JMException + { + AMQQueue queue = getQueueRegistry().getQueue(new AMQShortString(queueName)); + if (queue == null) + { + throw new JMException("Queue \"" + queueName + "\" is not registered with the exchange."); + } + + try + { + queue.bind(TopicExchange.this, new AMQShortString(binding), null); + } + catch (AMQException ex) + { + throw new MBeanException(ex); + } + } + + } // End of MBean class + + public AMQShortString getType() + { + return ExchangeDefaults.TOPIC_EXCHANGE_CLASS; + } + + public synchronized void registerQueue(AMQShortString rKey, AMQQueue queue, FieldTable args) throws AMQException + { + assert queue != null; + assert rKey != null; + + _logger.debug("Registering queue " + queue.getName() + " with routing key " + rKey); + + + AMQShortString routingKey; + + if(rKey.contains(HASH_BYTE) || rKey.contains(STAR_BYTE)) + { + routingKey = normalize(rKey); + } + else + { + routingKey = rKey; + } + + Binding binding = new Binding(rKey, queue); + + if(_bindings.containsKey(binding)) + { + FieldTable oldArgs = _bindings.get(binding); + TopicExchangeResult result = _topicExchangeResults.get(routingKey); + + if(argumentsContainSelector(args)) + { + if(argumentsContainSelector(oldArgs)) + { + result.replaceQueueFilter(queue,createSelectorFilter(oldArgs), createSelectorFilter(args)); + } + else + { + result.addFilteredQueue(queue,createSelectorFilter(args)); + result.removeUnfilteredQueue(queue); + } + } + else + { + if(argumentsContainSelector(oldArgs)) + { + result.addUnfilteredQueue(queue); + result.removeFilteredQueue(queue, createSelectorFilter(oldArgs)); + } + else + { + // TODO - fix control flow + return; + } + } + + } + else + { + + TopicExchangeResult result = _topicExchangeResults.get(routingKey); + if(result == null) + { + result = new TopicExchangeResult(); + if(argumentsContainSelector(args)) + { + result.addFilteredQueue(queue, createSelectorFilter(args)); + } + else + { + result.addUnfilteredQueue(queue); + } + _parser.addBinding(routingKey, result); + _topicExchangeResults.put(routingKey,result); + } + else + { + if(argumentsContainSelector(args)) + { + result.addFilteredQueue(queue, createSelectorFilter(args)); + } + else + { + result.addUnfilteredQueue(queue); + } + } + _bindings.put(binding, args); + } + + + } + + private JMSSelectorFilter<RuntimeException> createSelectorFilter(final FieldTable args) + throws AMQException + { + + final String selectorString = args.getString(AMQPFilterTypes.JMS_SELECTOR.getValue()); + WeakReference<JMSSelectorFilter<RuntimeException>> selectorRef = _selectorCache.get(selectorString); + JMSSelectorFilter selector = null; + + if(selectorRef == null || (selector = selectorRef.get())==null) + { + selector = new JMSSelectorFilter<RuntimeException>(selectorString); + _selectorCache.put(selectorString, new WeakReference<JMSSelectorFilter<RuntimeException>>(selector)); + } + return selector; + } + + private static boolean argumentsContainSelector(final FieldTable args) + { + return args != null && args.containsKey(AMQPFilterTypes.JMS_SELECTOR.getValue()) && args.getString(AMQPFilterTypes.JMS_SELECTOR.getValue()).trim().length() != 0; + } + + private AMQShortString normalize(AMQShortString routingKey) + { + if(routingKey == null) + { + routingKey = AMQShortString.EMPTY_STRING; + } + + AMQShortStringTokenizer routingTokens = routingKey.tokenize(TOPIC_SEPARATOR); + + List<AMQShortString> subscriptionList = new ArrayList<AMQShortString>(); + + while (routingTokens.hasMoreTokens()) + { + subscriptionList.add(routingTokens.nextToken()); + } + + int size = subscriptionList.size(); + + for (int index = 0; index < size; index++) + { + // if there are more levels + if ((index + 1) < size) + { + if (subscriptionList.get(index).equals(AMQP_HASH_TOKEN)) + { + if (subscriptionList.get(index + 1).equals(AMQP_HASH_TOKEN)) + { + // we don't need #.# delete this one + subscriptionList.remove(index); + size--; + // redo this normalisation + index--; + } + + if (subscriptionList.get(index + 1).equals(AMQP_STAR_TOKEN)) + { + // we don't want #.* swap to *.# + // remove it and put it in at index + 1 + subscriptionList.add(index + 1, subscriptionList.remove(index)); + } + } + } // if we have more levels + } + + + + AMQShortString normalizedString = AMQShortString.join(subscriptionList, TOPIC_SEPARATOR_AS_SHORTSTRING); + + return normalizedString; + } + + public void route(IncomingMessage payload) throws AMQException + { + + final AMQShortString routingKey = payload.getRoutingKey(); + + Collection<AMQQueue> queues = getMatchedQueues(payload, routingKey); + + if(queues == null || queues.isEmpty()) + { + _logger.info("Message routing key: " + payload.getRoutingKey() + " No routes."); + } + + payload.enqueue(queues); + + } + + public boolean isBound(AMQShortString routingKey, FieldTable arguments, AMQQueue queue) + { + return isBound(routingKey, queue); + } + + public boolean isBound(AMQShortString routingKey, AMQQueue queue) + { + Binding binding = new Binding(routingKey, queue); + + return _bindings.containsKey(binding); + } + + public boolean isBound(AMQShortString routingKey) + { + for(Binding b : _bindings.keySet()) + { + if(b.getBindingKey().equals(routingKey)) + { + return true; + } + } + + return false; + } + + public boolean isBound(AMQQueue queue) + { + for(Binding b : _bindings.keySet()) + { + if(b.getQueue().equals(queue)) + { + return true; + } + } + + return false; + } + + public boolean hasBindings() + { + return !_bindings.isEmpty(); + } + + public synchronized void deregisterQueue(AMQShortString rKey, AMQQueue queue, FieldTable args) throws AMQException + { + assert queue != null; + assert rKey != null; + + Binding binding = new Binding(rKey, queue); + + + if (!_bindings.containsKey(binding)) + { + throw new AMQException(AMQConstant.NOT_FOUND, "Queue " + queue.getName() + " was not registered with exchange " + this.getName() + + " with routing key " + rKey + "."); + } + + FieldTable bindingArgs = _bindings.remove(binding); + AMQShortString bindingKey = normalize(rKey); + TopicExchangeResult result = _topicExchangeResults.get(bindingKey); + if(argumentsContainSelector(bindingArgs)) + { + result.removeFilteredQueue(queue, createSelectorFilter(bindingArgs)); + } + else + { + result.removeUnfilteredQueue(queue); + } + + } + + protected ExchangeMBean createMBean() throws AMQException + { + try + { + return new TopicExchangeMBean(); + } + catch (JMException ex) + { + _logger.error("Exception occured in creating the topic exchenge mbean", ex); + throw new AMQException("Exception occured in creating the topic exchenge mbean", ex); + } + } + + private Collection<AMQQueue> getMatchedQueues(IncomingMessage message, AMQShortString routingKey) + { + + Collection<TopicMatcherResult> results = _parser.parse(routingKey); + if(results.isEmpty()) + { + return Collections.EMPTY_SET; + } + else + { + Collection<AMQQueue> queues = results.size() == 1 ? null : new HashSet<AMQQueue>(); + for(TopicMatcherResult result : results) + { + + queues = ((TopicExchangeResult)result).processMessage(message, queues); + } + return queues; + } + + + } +} diff --git a/java/broker/src/main/java/org/apache/qpid/server/exchange/headers/HeaderKey.java b/java/broker/src/main/java/org/apache/qpid/server/exchange/headers/HeaderKey.java new file mode 100644 index 0000000000..8fdb91cbef --- /dev/null +++ b/java/broker/src/main/java/org/apache/qpid/server/exchange/headers/HeaderKey.java @@ -0,0 +1,40 @@ +package org.apache.qpid.server.exchange.headers; + +import org.apache.qpid.framing.AMQShortString; + +/* +* +* 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. +* +*/ +public class HeaderKey +{ + public static final HeaderKey UNKNOWN = new HeaderKey(new AMQShortString("<< UNKNOWN >>")); + private AMQShortString _key; + + public HeaderKey(final AMQShortString key) + { + _key = key; + } + + public String toString() + { + return _key.toString(); + } + +} diff --git a/java/broker/src/main/java/org/apache/qpid/server/exchange/headers/HeaderKeyDictionary.java b/java/broker/src/main/java/org/apache/qpid/server/exchange/headers/HeaderKeyDictionary.java new file mode 100644 index 0000000000..7be99a88c9 --- /dev/null +++ b/java/broker/src/main/java/org/apache/qpid/server/exchange/headers/HeaderKeyDictionary.java @@ -0,0 +1,50 @@ +package org.apache.qpid.server.exchange.headers; + +import org.apache.qpid.framing.AMQShortString; + +import java.util.Map; +import java.util.HashMap; + +/* +* +* 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. +* +*/ +public class HeaderKeyDictionary +{ + + private final Map<AMQShortString, HeaderKey> _dictionary = new HashMap<AMQShortString, HeaderKey>(); + + + public HeaderKey get(final AMQShortString key) + { + HeaderKey headerKey = _dictionary.get(key); + return headerKey == null ? HeaderKey.UNKNOWN : headerKey; + } + + public HeaderKey getOrCreate(final AMQShortString key) + { + HeaderKey headerKey = _dictionary.get(key); + if(headerKey == null) + { + headerKey = new HeaderKey(key); + _dictionary.put(key, headerKey); + } + return headerKey; + } +} diff --git a/java/broker/src/main/java/org/apache/qpid/server/exchange/headers/HeaderMatcherResult.java b/java/broker/src/main/java/org/apache/qpid/server/exchange/headers/HeaderMatcherResult.java new file mode 100644 index 0000000000..518064bb29 --- /dev/null +++ b/java/broker/src/main/java/org/apache/qpid/server/exchange/headers/HeaderMatcherResult.java @@ -0,0 +1,25 @@ +package org.apache.qpid.server.exchange.headers; + +/* +* +* 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. +* +*/ +public class HeaderMatcherResult +{ +} diff --git a/java/broker/src/main/java/org/apache/qpid/server/exchange/headers/HeadersMatcherDFAState.java b/java/broker/src/main/java/org/apache/qpid/server/exchange/headers/HeadersMatcherDFAState.java new file mode 100644 index 0000000000..9da93d483a --- /dev/null +++ b/java/broker/src/main/java/org/apache/qpid/server/exchange/headers/HeadersMatcherDFAState.java @@ -0,0 +1,339 @@ +package org.apache.qpid.server.exchange.headers; + +import org.apache.qpid.framing.AMQTypedValue; +import org.apache.qpid.framing.AMQShortString; +import org.apache.qpid.framing.FieldTable; +import org.apache.qpid.server.exchange.topic.TopicMatcherDFAState; +import org.apache.qpid.server.exchange.topic.TopicWord; +import org.apache.qpid.server.exchange.topic.TopicMatcherResult; + +import java.util.*; + +/* +* +* 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. +* +*/ +public class HeadersMatcherDFAState +{ + + + private final Collection<HeaderMatcherResult> _results; + private final Map<HeaderKey, Map<AMQTypedValue,HeadersMatcherDFAState>> _nextStateMap; + private final HeaderKeyDictionary _dictionary; + + public HeadersMatcherDFAState(Map<HeaderKey, Map<AMQTypedValue,HeadersMatcherDFAState>> nextStateMap, + Collection<HeaderMatcherResult> results, + HeaderKeyDictionary dictionary) + { + _nextStateMap = nextStateMap; + _results = results; + _dictionary = dictionary; + } + + + public Collection<HeaderMatcherResult> match(final FieldTable table) + { + return match(table.iterator()); + } + + + + public Collection<HeaderMatcherResult> match(Iterator<Map.Entry<AMQShortString,AMQTypedValue>> fieldTableIterator) + { + + if(_nextStateMap.isEmpty()) + { + return _results; + } + + while(fieldTableIterator.hasNext()) + { + + Map.Entry<AMQShortString, AMQTypedValue> fieldTableEntry = fieldTableIterator.next(); + HeaderKey key = _dictionary.get(fieldTableEntry.getKey()); + if(key != HeaderKey.UNKNOWN) + { + Map<AMQTypedValue, HeadersMatcherDFAState> valueToStateMap = _nextStateMap.get(key); + + if(valueToStateMap != null) + { + HeadersMatcherDFAState nextState = valueToStateMap.get(fieldTableEntry.getValue()); + + if(nextState == null) + { + nextState = valueToStateMap.get(null); + } + if(nextState != null && nextState != this) + { + return nextState.match(fieldTableIterator); + } + } + + } + } + + return _results; + + } + + + HeadersMatcherDFAState mergeStateMachines(HeadersMatcherDFAState otherStateMachine) + { + + assert(otherStateMachine._dictionary == _dictionary); + + Map<Set<HeadersMatcherDFAState>, HeadersMatcherDFAState> newStateMap= new HashMap<Set<HeadersMatcherDFAState>, HeadersMatcherDFAState>(); + + Collection<HeaderMatcherResult> results; + + if(_results.isEmpty()) + { + results = otherStateMachine._results; + } + else if(otherStateMachine._results.isEmpty()) + { + results = _results; + } + else + { + results = new HashSet<HeaderMatcherResult>(_results); + results.addAll(otherStateMachine._results); + } + + + final Map<HeaderKey, Map<AMQTypedValue, HeadersMatcherDFAState>> newNextStateMap = new HashMap<HeaderKey, Map<AMQTypedValue, HeadersMatcherDFAState>>(); + + HeadersMatcherDFAState newState = new HeadersMatcherDFAState(newNextStateMap, results, _dictionary); + + + Set<HeadersMatcherDFAState> oldStates = new HashSet<HeadersMatcherDFAState>(); + oldStates.add(this); + oldStates.add(otherStateMachine); + + newStateMap.put(oldStates, newState); + + mergeStateMachines(oldStates, newNextStateMap, newStateMap); + + return newState; + + + } + + private void mergeStateMachines(final Set<HeadersMatcherDFAState> oldStates, + final Map<HeaderKey, Map<AMQTypedValue, HeadersMatcherDFAState>> newNextStateMap, + final Map<Set<HeadersMatcherDFAState>, HeadersMatcherDFAState> newStateMap) + { + Map<HeaderKey, Map<AMQTypedValue, Set<HeadersMatcherDFAState>>> nfaMap = new HashMap<HeaderKey, Map<AMQTypedValue, Set<HeadersMatcherDFAState>>>(); + + Set<HeaderKey> distinctKeys = new HashSet<HeaderKey>(); + + for(HeadersMatcherDFAState state : oldStates) + { + Map<HeaderKey, Map<AMQTypedValue, HeadersMatcherDFAState>> map = state._nextStateMap; + + for(Map.Entry<HeaderKey, Map<AMQTypedValue, HeadersMatcherDFAState>> entry : map.entrySet()) + { + Map<AMQTypedValue, Set<HeadersMatcherDFAState>> valueToStatesMap = nfaMap.get(entry.getKey()); + + if(valueToStatesMap == null) + { + valueToStatesMap = new HashMap<AMQTypedValue, Set<HeadersMatcherDFAState>>(); + nfaMap.put(entry.getKey(), valueToStatesMap); + } + + for(Map.Entry<AMQTypedValue, HeadersMatcherDFAState> valueToStateEntry : entry.getValue().entrySet()) + { + Set<HeadersMatcherDFAState> states = valueToStatesMap.get(valueToStateEntry.getKey()); + if(states == null) + { + states = new HashSet<HeadersMatcherDFAState>(); + valueToStatesMap.put(valueToStateEntry.getKey(),states); + } + states.add(valueToStateEntry.getValue()); + } + + distinctKeys.add(entry.getKey()); + } + } + + Map<HeaderKey, Set<HeadersMatcherDFAState>> anyValueStates = new HashMap<HeaderKey, Set<HeadersMatcherDFAState>>(); + + for(HeaderKey distinctKey : distinctKeys) + { + Map<AMQTypedValue, Set<HeadersMatcherDFAState>> valueToStateMap = nfaMap.get(distinctKey); + if(valueToStateMap != null) + { + Set<HeadersMatcherDFAState> statesForKeyDefault = valueToStateMap.get(null); + if(statesForKeyDefault != null) + { + anyValueStates.put(distinctKey, statesForKeyDefault); + } + } + } + + // add the defaults for "null" to all other specified values of a given header key + + for( Map.Entry<HeaderKey,Map<AMQTypedValue,Set<HeadersMatcherDFAState>>> entry : nfaMap.entrySet()) + { + Map<AMQTypedValue, Set<HeadersMatcherDFAState>> valueToStatesMap = entry.getValue(); + for(Map.Entry<AMQTypedValue, Set<HeadersMatcherDFAState>> valueToStates : valueToStatesMap.entrySet()) + { + if(valueToStates.getKey() != null) + { + + + Set<HeadersMatcherDFAState> defaults = anyValueStates.get(entry.getKey()); + if(defaults != null) + { + valueToStates.getValue().addAll(defaults); + } + } + } + } + + // if a given header key is not mentioned in the map of a machine; then that machine would stay at the same state + // for that key. + for(HeaderKey distinctKey : distinctKeys) + { + Map<AMQTypedValue, Set<HeadersMatcherDFAState>> valueToStatesMap = nfaMap.get(distinctKey); + for(HeadersMatcherDFAState oldState : oldStates) + { + if(!oldState._nextStateMap.containsKey(distinctKey)) + { + for(Set<HeadersMatcherDFAState> endStates : valueToStatesMap.values()) + { + endStates.add(oldState); + } + } + } + } + + + + + for(Map.Entry<HeaderKey,Map<AMQTypedValue,Set<HeadersMatcherDFAState>>> transitionClass : nfaMap.entrySet()) + { + Map<AMQTypedValue, HeadersMatcherDFAState> valueToDFAState = newNextStateMap.get(transitionClass.getKey()); + if(valueToDFAState == null) + { + valueToDFAState = new HashMap<AMQTypedValue, HeadersMatcherDFAState>(); + newNextStateMap.put(transitionClass.getKey(), valueToDFAState); + } + + for(Map.Entry<AMQTypedValue,Set<HeadersMatcherDFAState>> transition : transitionClass.getValue().entrySet()) + { + Set<HeadersMatcherDFAState> destinations = transition.getValue(); + + + HeadersMatcherDFAState nextState = newStateMap.get(destinations); + + if(nextState == null) + { + + if(destinations.size() == 1) + { + nextState = destinations.iterator().next(); + newStateMap.put(destinations, nextState); + } + else + { + Collection<HeaderMatcherResult> results; + + Set<Collection<HeaderMatcherResult>> resultSets = new HashSet<Collection<HeaderMatcherResult>>(); + for(HeadersMatcherDFAState destination : destinations) + { + resultSets.add(destination._results); + } + resultSets.remove(Collections.EMPTY_SET); + if(resultSets.size() == 0) + { + results = Collections.EMPTY_SET; + } + else if(resultSets.size() == 1) + { + results = resultSets.iterator().next(); + } + else + { + results = new HashSet<HeaderMatcherResult>(); + for(Collection<HeaderMatcherResult> oldResult : resultSets) + { + results.addAll(oldResult); + } + } + + final Map<HeaderKey, Map<AMQTypedValue, HeadersMatcherDFAState>> nextStateMap = new HashMap<HeaderKey, Map<AMQTypedValue, HeadersMatcherDFAState>>(); + + nextState = new HeadersMatcherDFAState(nextStateMap, results, _dictionary); + newStateMap.put(destinations, nextState); + + mergeStateMachines( + destinations, + nextStateMap, + newStateMap); + + + } + + + } + valueToDFAState.put(transition.getKey(),nextState); + } + } + + + + final ArrayList<HeaderKey> removeKeyList = new ArrayList<HeaderKey>(); + + for(Map.Entry<HeaderKey,Map<AMQTypedValue,HeadersMatcherDFAState>> entry : _nextStateMap.entrySet()) + { + final ArrayList<AMQTypedValue> removeValueList = new ArrayList<AMQTypedValue>(); + + for(Map.Entry<AMQTypedValue,HeadersMatcherDFAState> valueToDFAState : entry.getValue().entrySet()) + { + if(valueToDFAState.getValue() == this) + { + HeadersMatcherDFAState defaultState = entry.getValue().get(null); + if(defaultState == null || defaultState == this) + { + removeValueList.add(valueToDFAState.getKey()); + } + } + } + + for(AMQTypedValue removeValue : removeValueList) + { + entry.getValue().remove(removeValue); + } + + if(entry.getValue().isEmpty()) + { + removeKeyList.add(entry.getKey()); + } + + } + + for(HeaderKey removeKey : removeKeyList) + { + _nextStateMap.remove(removeKey); + } + + } + +} diff --git a/java/broker/src/main/java/org/apache/qpid/server/exchange/headers/HeadersParser.java b/java/broker/src/main/java/org/apache/qpid/server/exchange/headers/HeadersParser.java new file mode 100644 index 0000000000..85e74122c3 --- /dev/null +++ b/java/broker/src/main/java/org/apache/qpid/server/exchange/headers/HeadersParser.java @@ -0,0 +1,439 @@ +package org.apache.qpid.server.exchange.headers; + +import org.apache.qpid.framing.*; + +import java.util.*; + +/* +* +* 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. +* +*/ +public class HeadersParser +{ + + private final HeaderKeyDictionary _dictionary = new HeaderKeyDictionary(); + private static final AMQShortString MATCHING_TYPE_KEY = new AMQShortString("x-match"); + private static final String ANY_MATCHING = "any"; + private static final AMQShortString RESERVED_KEY_PREFIX = new AMQShortString("x-"); + + + HeadersMatcherDFAState createStateMachine(FieldTable bindingArguments, HeaderMatcherResult result) + { + String matchingType = bindingArguments.getString(MATCHING_TYPE_KEY); + boolean matchAny = matchingType.equalsIgnoreCase(ANY_MATCHING); + if(matchAny) + { + return createStateMachineForAnyMatch(bindingArguments, result); + } + else + { + return createStateMachineForAllMatch(bindingArguments, result); + } + + + } + + + private HeadersMatcherDFAState createStateMachineForAnyMatch(final FieldTable bindingArguments, + final HeaderMatcherResult result) + { + + // DFAs for "any" matches have only two states, "not-matched" and "matched"... they start in the former + // and upon meeting any of the criteria they move to the latter + + //noinspection unchecked + final HeadersMatcherDFAState successState = + new HeadersMatcherDFAState(Collections.EMPTY_MAP,Collections.singleton(result),_dictionary); + + Map<HeaderKey, Map<AMQTypedValue, HeadersMatcherDFAState>> nextStateMap = + new HashMap<HeaderKey, Map<AMQTypedValue, HeadersMatcherDFAState>>(); + + Set<AMQShortString> seenKeys = new HashSet<AMQShortString>(); + + Iterator<Map.Entry<AMQShortString, AMQTypedValue>> tableIterator = bindingArguments.iterator(); + + while(tableIterator.hasNext()) + { + final Map.Entry<AMQShortString, AMQTypedValue> entry = tableIterator.next(); + final AMQShortString key = entry.getKey(); + final AMQTypedValue value = entry.getValue(); + + + if(seenKeys.add(key) && !key.startsWith(RESERVED_KEY_PREFIX)) + { + final AMQType type = value.getType(); + + final HeaderKey headerKey = _dictionary.getOrCreate(key); + final Map<AMQTypedValue, HeadersMatcherDFAState> valueMap; + + if(type == AMQType.VOID || + ((type == AMQType.ASCII_STRING || type == AMQType.WIDE_STRING) && ((CharSequence)value.getValue()).length() == 0)) + { + valueMap = Collections.singletonMap(null,successState); + + } + else + { + valueMap = Collections.singletonMap(value,successState); + } + nextStateMap.put(headerKey,valueMap); + + } + + } + + if(seenKeys.size() == 0) + { + return successState; + } + else + { + return new HeadersMatcherDFAState(nextStateMap,Collections.EMPTY_SET,_dictionary); + } + + + } + + + private HeadersMatcherDFAState createStateMachineForAllMatch(final FieldTable bindingArguments, + final HeaderMatcherResult result) + { + // DFAs for "all" matches have a "success" state, a "fail" state, and states for every subset of + // matches which are possible, starting with the empty subset. For example if we have a binding + // { x-match="all" + // a=1 + // b=1 + // c=1 + // d=1 } + // Then we would have the following states + // (1) Seen none of a, b, c, or d + // (2) Seen a=1 ; none of b,c, or d + // (3) Seen b=1 ; none of a,c, or d + // (4) Seen c=1 ; none of a,b, or d + // (5) Seen d=1 ; none of a,b, or c + // (6) Seen a=1,b=1 ; none of c,d + // (7) Seen a=1,c=1 ; none of b,d + // (8) Seen a=1,d=1 ; none of b,c + // (9) Seen b=1,c=1 ; none of a,d + //(10) Seen b=1,d=1 ; none of c,d + //(11) Seen c=1,d=1 ; none of a,b + //(12) Seen a=1,b=1,c=1 ; not d + //(13) Seen a=1,b=1,d=1 ; not c + //(14) Seen a=1,c=1,d=1 ; not b + //(15) Seen b=1,c=1,d=1 ; not a + //(16) success + //(17) fail + // + // All states but (16) can transition to (17); additionally: + // (1) can transition to (2),(3),(4),(5) + // (2) can transition to (6),(7),(8) + // (3) can transition to (6),(9),(10) + // (4) can transition to (7),(9),(11) + // (5) can transition to (8),(10),(11) + // (6) can transition to (12),(13) + // (7) can transition to (12),(14) + // (8) can transition to (13),(14) + // (9) can transition to (12),(15) + //(10) can transition to (13),(15) + //(11) can transition to (14),(15) + //(12)-(15) can transition to (16) + + Set<AMQShortString> seenKeys = new HashSet<AMQShortString>(); + List<KeyValuePair> requiredTerms = new ArrayList<KeyValuePair>(bindingArguments.size()); + + Iterator<Map.Entry<AMQShortString, AMQTypedValue>> tableIterator = bindingArguments.iterator(); + + + + while(tableIterator.hasNext()) + { + final Map.Entry<AMQShortString, AMQTypedValue> entry = tableIterator.next(); + final AMQShortString key = entry.getKey(); + final AMQTypedValue value = entry.getValue(); + + + if(seenKeys.add(key) && !key.startsWith(RESERVED_KEY_PREFIX)) + { + final AMQType type = value.getType(); + + if(type == AMQType.VOID || + ((type == AMQType.ASCII_STRING || type == AMQType.WIDE_STRING) && ((CharSequence)value.getValue()).length() == 0)) + { + requiredTerms.add(new KeyValuePair(_dictionary.getOrCreate(key),null)); + } + else + { + requiredTerms.add(new KeyValuePair(_dictionary.getOrCreate(key),value)); + } + } + + } + + final HeadersMatcherDFAState successState = + new HeadersMatcherDFAState(Collections.EMPTY_MAP,Collections.singleton(result),_dictionary); + + final HeadersMatcherDFAState failState = + new HeadersMatcherDFAState(Collections.EMPTY_MAP,Collections.EMPTY_SET,_dictionary); + + Map<Set<KeyValuePair>, HeadersMatcherDFAState> notSeenTermsToStateMap = + new HashMap<Set<KeyValuePair>, HeadersMatcherDFAState>(); + + notSeenTermsToStateMap.put(Collections.EMPTY_SET, successState); + + + final int numberOfTerms = requiredTerms.size(); + + for(int numMissingTerms = 1; numMissingTerms <= numberOfTerms; numMissingTerms++) + { + int[] pos = new int[numMissingTerms]; + for(int i = 0; i < numMissingTerms; i++) + { + pos[i] = i; + } + + final int maxTermValue = (numberOfTerms - (numMissingTerms - 1)); + + while(pos[0] < maxTermValue) + { + + Set<KeyValuePair> stateSet = new HashSet<KeyValuePair>(); + for(int posIndex = 0; posIndex < pos.length; posIndex++) + { + stateSet.add(requiredTerms.get(pos[posIndex])); + } + + final Map<HeaderKey, Map<AMQTypedValue,HeadersMatcherDFAState>> nextStateMap = + new HashMap<HeaderKey, Map<AMQTypedValue,HeadersMatcherDFAState>>(); + + + for(int posIndex = 0; posIndex < pos.length; posIndex++) + { + KeyValuePair nextTerm = requiredTerms.get(pos[posIndex]); + HashSet<KeyValuePair> nextStateSet = + new HashSet<KeyValuePair>(stateSet); + nextStateSet.remove(nextTerm); + + Map<AMQTypedValue, HeadersMatcherDFAState> valueToStateMap = + new HashMap<AMQTypedValue, HeadersMatcherDFAState>(); + nextStateMap.put(nextTerm._key, valueToStateMap); + + valueToStateMap.put( nextTerm._value,notSeenTermsToStateMap.get(nextStateSet)); + if(nextTerm._value != null) + { + valueToStateMap.put(null, failState); + } + + + } + + + HeadersMatcherDFAState newState = new HeadersMatcherDFAState(nextStateMap, Collections.EMPTY_SET, _dictionary); + + notSeenTermsToStateMap.put(stateSet, newState); + + int i = numMissingTerms; + while(i-- != 0) + { + if(++pos[i] <= numberOfTerms -(numMissingTerms-i)) + { + int k = pos[i]; + for(int j = i+1; j < numMissingTerms; j++) + { + pos[j] = ++k; + } + break; + } + } + } + + + + + } + + + return notSeenTermsToStateMap.get(new HashSet<KeyValuePair>(requiredTerms)); + + + + } + + public static void main(String[] args) throws AMQFrameDecodingException + { + + FieldTable bindingTable = new FieldTable(); + + bindingTable.setString(new AMQShortString("x-match"),"all"); + bindingTable.setInteger("a",1); + bindingTable.setVoid(new AMQShortString("b")); + bindingTable.setString("c",""); + bindingTable.setInteger("d",4); + bindingTable.setInteger("e",1); + + + + FieldTable bindingTable2 = new FieldTable(); + bindingTable2.setString(new AMQShortString("x-match"),"all"); + bindingTable2.setInteger("a",1); + bindingTable2.setVoid(new AMQShortString("b")); + bindingTable2.setString("c",""); + bindingTable2.setInteger("d",4); + bindingTable2.setInteger("e",1); + bindingTable2.setInteger("f",1); + + + FieldTable table = new FieldTable(); + table.setInteger("a",1); + table.setInteger("b",2); + table.setString("c",""); + table.setInteger("d",4); + table.setInteger("e",1); + table.setInteger("f",1); + table.setInteger("h",1); + table.setInteger("i",1); + table.setInteger("j",1); + table.setInteger("k",1); + table.setInteger("l",1); + + org.apache.mina.common.ByteBuffer buffer = org.apache.mina.common.ByteBuffer.allocate( (int) table.getEncodedSize()); + EncodingUtils.writeFieldTableBytes(buffer, table); + buffer.flip(); + + FieldTable table2 = EncodingUtils.readFieldTable(buffer); + + + + FieldTable bindingTable3 = new FieldTable(); + bindingTable3.setString(new AMQShortString("x-match"),"any"); + bindingTable3.setInteger("a",1); + bindingTable3.setInteger("b",3); + + + FieldTable bindingTable4 = new FieldTable(); + bindingTable4.setString(new AMQShortString("x-match"),"any"); + bindingTable4.setVoid(new AMQShortString("a")); + + + FieldTable bindingTable5 = new FieldTable(); + bindingTable5.setString(new AMQShortString("x-match"),"all"); + bindingTable5.setString(new AMQShortString("h"),"hello"); + + for(int i = 0; i < 100; i++) + { + printMatches(new FieldTable[] {bindingTable5} , table2); + } + + + + } + + + + private static void printMatches(final FieldTable[] bindingKeys, final FieldTable routingKey) + { + HeadersMatcherDFAState sm = null; + Map<HeaderMatcherResult, String> resultMap = new HashMap<HeaderMatcherResult, String>(); + + HeadersParser parser = new HeadersParser(); + + for(int i = 0; i < bindingKeys.length; i++) + { + HeaderMatcherResult r = new HeaderMatcherResult(); + resultMap.put(r, bindingKeys[i].toString()); + + + if(i==0) + { + sm = parser.createStateMachine(bindingKeys[i], r); + } + else + { + sm = sm.mergeStateMachines(parser.createStateMachine(bindingKeys[i], r)); + } + } + + Collection<HeaderMatcherResult> results = null; + long beforeTime = System.currentTimeMillis(); + for(int i = 0; i < 1000000; i++) + { + routingKey.size(); + + assert sm != null; + results = sm.match(routingKey); + + } + long elapsed = System.currentTimeMillis() - beforeTime; + System.out.println("1000000 Iterations took: " + elapsed); + Collection<String> resultStrings = new ArrayList<String>(); + + assert results != null; + for(HeaderMatcherResult result : results) + { + resultStrings.add(resultMap.get(result)); + } + + final ArrayList<String> nonMatches = new ArrayList<String>(); + for(FieldTable key : bindingKeys) + { + nonMatches.add(key.toString()); + } + nonMatches.removeAll(resultStrings); + System.out.println("\""+routingKey+"\" matched with " + resultStrings + " DID NOT MATCH with " + nonMatches); + + + } + + + public final static class KeyValuePair + { + public final HeaderKey _key; + public final AMQTypedValue _value; + private final int _hashCode; + + public KeyValuePair(final HeaderKey key, final AMQTypedValue value) + { + _key = key; + _value = value; + int hash = (1 + 31 * _key.hashCode()); + if(_value != null) + { + hash+=_value.hashCode(); + } + _hashCode = hash; + } + + public int hashCode() + { + return _hashCode; + } + + public boolean equals(Object o) + { + KeyValuePair other = (KeyValuePair)o; + return (_key == other._key) && (_value == null ? other._value == null : _value.equals(other._value)); + } + + + public String toString() + { + return "{" + _key + " -> " + _value + "}"; + } + + } +} diff --git a/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicMatcherDFAState.java b/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicMatcherDFAState.java new file mode 100644 index 0000000000..36076cf75b --- /dev/null +++ b/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicMatcherDFAState.java @@ -0,0 +1,295 @@ +package org.apache.qpid.server.exchange.topic; + +import org.apache.qpid.framing.AMQShortString; +import org.apache.qpid.framing.AMQShortStringTokenizer; + +import java.util.*; +import java.util.concurrent.atomic.AtomicInteger; + +/* +* +* 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. +* +*/ +public class TopicMatcherDFAState +{ + private static final AtomicInteger stateId = new AtomicInteger(); + + private final int _id = stateId.incrementAndGet(); + + private final Collection<TopicMatcherResult> _results; + private final Map<TopicWord, TopicMatcherDFAState> _nextStateMap; + private static final byte TOPIC_DELIMITTER = (byte)'.'; + + + public TopicMatcherDFAState(Map<TopicWord, TopicMatcherDFAState> nextStateMap, + Collection<TopicMatcherResult> results ) + { + _nextStateMap = nextStateMap; + _results = results; + } + + + public TopicMatcherDFAState nextState(TopicWord word) + { + final TopicMatcherDFAState nextState = _nextStateMap.get(word); + return nextState == null ? _nextStateMap.get(TopicWord.ANY_WORD) : nextState; + } + + public Collection<TopicMatcherResult> terminate() + { + return _results; + } + + + public Collection<TopicMatcherResult> parse(TopicWordDictionary dictionary, AMQShortString routingKey) + { + return parse(dictionary, routingKey.tokenize(TOPIC_DELIMITTER)); + } + + private Collection<TopicMatcherResult> parse(final TopicWordDictionary dictionary, + final AMQShortStringTokenizer tokens) + { + if(!tokens.hasMoreTokens()) + { + return _results; + } + TopicWord word = dictionary.getWord(tokens.nextToken()); + TopicMatcherDFAState nextState = _nextStateMap.get(word); + if(nextState == null && word != TopicWord.ANY_WORD) + { + nextState = _nextStateMap.get(TopicWord.ANY_WORD); + } + if(nextState == null) + { + return Collections.EMPTY_SET; + } + // Shortcut if we are at a looping terminal state + if((nextState == this) && (_nextStateMap.size() == 1) && _nextStateMap.containsKey(TopicWord.ANY_WORD)) + { + return _results; + } + + return nextState.parse(dictionary, tokens); + + } + + + public TopicMatcherDFAState mergeStateMachines(TopicMatcherDFAState otherStateMachine) + { + Map<Set<TopicMatcherDFAState>, TopicMatcherDFAState> newStateMap= new HashMap<Set<TopicMatcherDFAState>, TopicMatcherDFAState>(); + + Collection<TopicMatcherResult> results; + + if(_results.isEmpty()) + { + results = otherStateMachine._results; + } + else if(otherStateMachine._results.isEmpty()) + { + results = _results; + } + else + { + results = new HashSet<TopicMatcherResult>(_results); + results.addAll(otherStateMachine._results); + } + + + final Map<TopicWord, TopicMatcherDFAState> newNextStateMap = new HashMap<TopicWord, TopicMatcherDFAState>(); + + TopicMatcherDFAState newState = new TopicMatcherDFAState(newNextStateMap, results); + + + Set<TopicMatcherDFAState> oldStates = new HashSet<TopicMatcherDFAState>(); + oldStates.add(this); + oldStates.add(otherStateMachine); + + newStateMap.put(oldStates, newState); + + mergeStateMachines(oldStates, newNextStateMap, newStateMap); + + return newState; + + } + + private static void mergeStateMachines( + final Set<TopicMatcherDFAState> oldStates, + final Map<TopicWord, TopicMatcherDFAState> newNextStateMap, + final Map<Set<TopicMatcherDFAState>, TopicMatcherDFAState> newStateMap) + { + Map<TopicWord, Set<TopicMatcherDFAState>> nfaMap = new HashMap<TopicWord, Set<TopicMatcherDFAState>>(); + + for(TopicMatcherDFAState state : oldStates) + { + Map<TopicWord, TopicMatcherDFAState> map = state._nextStateMap; + for(Map.Entry<TopicWord, TopicMatcherDFAState> entry : map.entrySet()) + { + Set<TopicMatcherDFAState> states = nfaMap.get(entry.getKey()); + if(states == null) + { + states = new HashSet<TopicMatcherDFAState>(); + nfaMap.put(entry.getKey(), states); + } + states.add(entry.getValue()); + } + } + + Set<TopicMatcherDFAState> anyWordStates = nfaMap.get(TopicWord.ANY_WORD); + + for(Map.Entry<TopicWord, Set<TopicMatcherDFAState>> transition : nfaMap.entrySet()) + { + Set<TopicMatcherDFAState> destinations = transition.getValue(); + + if(anyWordStates != null) + { + destinations.addAll(anyWordStates); + } + + TopicMatcherDFAState nextState = newStateMap.get(destinations); + if(nextState == null) + { + + if(destinations.size() == 1) + { + nextState = destinations.iterator().next(); + newStateMap.put(destinations, nextState); + } + else + { + Collection<TopicMatcherResult> results; + + Set<Collection<TopicMatcherResult>> resultSets = new HashSet<Collection<TopicMatcherResult>>(); + for(TopicMatcherDFAState destination : destinations) + { + resultSets.add(destination._results); + } + resultSets.remove(Collections.EMPTY_SET); + if(resultSets.size() == 0) + { + results = Collections.EMPTY_SET; + } + else if(resultSets.size() == 1) + { + results = resultSets.iterator().next(); + } + else + { + results = new HashSet<TopicMatcherResult>(); + for(Collection<TopicMatcherResult> oldResult : resultSets) + { + results.addAll(oldResult); + } + } + + final Map<TopicWord, TopicMatcherDFAState> nextStateMap = new HashMap<TopicWord, TopicMatcherDFAState>(); + + nextState = new TopicMatcherDFAState(nextStateMap, results); + newStateMap.put(destinations, nextState); + + mergeStateMachines( + destinations, + nextStateMap, + newStateMap); + + + } + + + } + newNextStateMap.put(transition.getKey(),nextState); + } + + // Remove redundant transitions where defined tokenWord has same action as ANY_WORD + TopicMatcherDFAState anyWordState = newNextStateMap.get(TopicWord.ANY_WORD); + if(anyWordState != null) + { + List<TopicWord> removeList = new ArrayList<TopicWord>(); + for(Map.Entry<TopicWord,TopicMatcherDFAState> entry : newNextStateMap.entrySet()) + { + if(entry.getValue() == anyWordState && entry.getKey() != TopicWord.ANY_WORD) + { + removeList.add(entry.getKey()); + } + } + for(TopicWord removeKey : removeList) + { + newNextStateMap.remove(removeKey); + } + } + + + + } + + + public String toString() + { + StringBuilder transitions = new StringBuilder(); + for(Map.Entry<TopicWord, TopicMatcherDFAState> entry : _nextStateMap.entrySet()) + { + transitions.append("[ "); + transitions.append(entry.getKey()); + transitions.append("\t ->\t "); + transitions.append(entry.getValue()._id); + transitions.append(" ]\n"); + } + + + return "[ State " + _id + " ]\n" + transitions + "\n"; + + } + + public String reachableStates() + { + StringBuilder result = new StringBuilder("Start state: " + _id + "\n"); + + SortedSet<TopicMatcherDFAState> reachableStates = + new TreeSet<TopicMatcherDFAState>(new Comparator<TopicMatcherDFAState>() + { + public int compare(final TopicMatcherDFAState o1, final TopicMatcherDFAState o2) + { + return o1._id - o2._id; + } + }); + reachableStates.add(this); + + int count; + + do + { + count = reachableStates.size(); + Collection<TopicMatcherDFAState> originalStates = new ArrayList<TopicMatcherDFAState>(reachableStates); + for(TopicMatcherDFAState state : originalStates) + { + reachableStates.addAll(state._nextStateMap.values()); + } + } + while(reachableStates.size() != count); + + + + for(TopicMatcherDFAState state : reachableStates) + { + result.append(state.toString()); + } + + return result.toString(); + } + +} diff --git a/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicMatcherResult.java b/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicMatcherResult.java new file mode 100644 index 0000000000..71d30adfac --- /dev/null +++ b/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicMatcherResult.java @@ -0,0 +1,25 @@ +package org.apache.qpid.server.exchange.topic; + +/* +* +* 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. +* +*/ +public interface TopicMatcherResult +{ +} diff --git a/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicParser.java b/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicParser.java new file mode 100644 index 0000000000..3e9facf412 --- /dev/null +++ b/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicParser.java @@ -0,0 +1,613 @@ +package org.apache.qpid.server.exchange.topic; + +import org.apache.qpid.framing.AMQShortString; +import org.apache.qpid.framing.AMQShortStringTokenizer; + +import java.util.*; +import java.util.concurrent.atomic.AtomicReference; +import java.io.IOException; + +/* +* +* 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. +* +*/ +public class TopicParser +{ + private static final byte TOPIC_DELIMITER = (byte)'.'; + + private final TopicWordDictionary _dictionary = new TopicWordDictionary(); + private final AtomicReference<TopicMatcherDFAState> _stateMachine = new AtomicReference<TopicMatcherDFAState>(); + + private static class Position + { + private final TopicWord _word; + private final boolean _selfTransition; + private final int _position; + private final boolean _endState; + private boolean _followedByAnyLoop; + + + public Position(final int position, final TopicWord word, final boolean selfTransition, final boolean endState) + { + _position = position; + _word = word; + _selfTransition = selfTransition; + _endState = endState; + } + + + } + + private static final Position ERROR_POSITION = new Position(Integer.MAX_VALUE,null, true, false); + + private static class SimpleState + { + Set<Position> _positions; + Map<TopicWord, SimpleState> _nextState; + } + + + public void addBinding(AMQShortString bindingKey, TopicMatcherResult result) + { + + TopicMatcherDFAState startingStateMachine; + TopicMatcherDFAState newStateMachine; + + do + { + startingStateMachine = _stateMachine.get(); + if(startingStateMachine == null) + { + newStateMachine = createStateMachine(bindingKey, result); + } + else + { + newStateMachine = startingStateMachine.mergeStateMachines(createStateMachine(bindingKey, result)); + } + + } + while(!_stateMachine.compareAndSet(startingStateMachine,newStateMachine)); + + } + + public Collection<TopicMatcherResult> parse(AMQShortString routingKey) + { + TopicMatcherDFAState stateMachine = _stateMachine.get(); + if(stateMachine == null) + { + return Collections.EMPTY_SET; + } + else + { + return stateMachine.parse(_dictionary,routingKey); + } + } + + + TopicMatcherDFAState createStateMachine(AMQShortString bindingKey, TopicMatcherResult result) + { + List<TopicWord> wordList = createTopicWordList(bindingKey); + int wildCards = 0; + for(TopicWord word : wordList) + { + if(word == TopicWord.WILDCARD_WORD) + { + wildCards++; + } + } + if(wildCards == 0) + { + TopicMatcherDFAState[] states = new TopicMatcherDFAState[wordList.size()+1]; + states[states.length-1] = new TopicMatcherDFAState(Collections.EMPTY_MAP, Collections.singleton(result)); + for(int i = states.length-2; i >= 0; i--) + { + states[i] = new TopicMatcherDFAState(Collections.singletonMap(wordList.get(i),states[i+1]),Collections.EMPTY_SET); + + } + return states[0]; + } + else if(wildCards == wordList.size()) + { + Map<TopicWord,TopicMatcherDFAState> stateMap = new HashMap<TopicWord,TopicMatcherDFAState>(); + TopicMatcherDFAState state = new TopicMatcherDFAState(stateMap, Collections.singleton(result)); + stateMap.put(TopicWord.ANY_WORD, state); + return state; + } + + + int positionCount = wordList.size() - wildCards; + + Position[] positions = new Position[positionCount+1]; + + int lastWord; + + if(wordList.get(wordList.size()-1)== TopicWord.WILDCARD_WORD) + { + lastWord = wordList.size()-1; + positions[positionCount] = new Position(positionCount, TopicWord.ANY_WORD, true, true); + } + else + { + lastWord = wordList.size(); + positions[positionCount] = new Position(positionCount, TopicWord.ANY_WORD, false, true); + } + + + int pos = 0; + int wordPos = 0; + + + while(wordPos < lastWord) + { + TopicWord word = wordList.get(wordPos++); + + if(word == TopicWord.WILDCARD_WORD) + { + int nextWordPos = wordPos++; + word = wordList.get(nextWordPos); + + positions[pos] = new Position(pos++,word,true,false); + } + else + { + positions[pos] = new Position(pos++,word,false,false); + } + + } + + + for(int p = 0; p<positionCount; p++) + { + boolean followedByWildcards = true; + + int n = p; + while(followedByWildcards && n<(positionCount+1)) + { + + if(positions[n]._selfTransition) + { + break; + } + else if(positions[n]._word!=TopicWord.ANY_WORD) + { + followedByWildcards = false; + } + n++; + } + + + positions[p]._followedByAnyLoop = followedByWildcards && (n!= positionCount+1); + } + + + // from each position you transition to a set of other positions. + // we approach this by examining steps of increasing length - so we + // look how far we can go from the start position in 1 word, 2 words, etc... + + Map<Set<Position>,SimpleState> stateMap = new HashMap<Set<Position>,SimpleState>(); + + + SimpleState state = new SimpleState(); + state._positions = Collections.singleton( positions[0] ); + stateMap.put(state._positions, state); + + calculateNextStates(state, stateMap, positions); + + SimpleState[] simpleStates = stateMap.values().toArray(new SimpleState[stateMap.size()]); + HashMap<TopicWord, TopicMatcherDFAState>[] dfaStateMaps = new HashMap[simpleStates.length]; + Map<SimpleState, TopicMatcherDFAState> simple2DFAMap = new HashMap<SimpleState, TopicMatcherDFAState>(); + + for(int i = 0; i < simpleStates.length; i++) + { + + Collection<TopicMatcherResult> results; + boolean endState = false; + + for(Position p : simpleStates[i]._positions) + { + if(p._endState) + { + endState = true; + break; + } + } + + if(endState) + { + results = Collections.singleton(result); + } + else + { + results = Collections.EMPTY_SET; + } + + dfaStateMaps[i] = new HashMap<TopicWord, TopicMatcherDFAState>(); + simple2DFAMap.put(simpleStates[i], new TopicMatcherDFAState(dfaStateMaps[i],results)); + + } + for(int i = 0; i < simpleStates.length; i++) + { + SimpleState simpleState = simpleStates[i]; + + Map<TopicWord, SimpleState> nextSimpleStateMap = simpleState._nextState; + for(Map.Entry<TopicWord, SimpleState> stateMapEntry : nextSimpleStateMap.entrySet()) + { + dfaStateMaps[i].put(stateMapEntry.getKey(), simple2DFAMap.get(stateMapEntry.getValue())); + } + + } + + return simple2DFAMap.get(state); + + } + + + + private void calculateNextStates(final SimpleState state, + final Map<Set<Position>, SimpleState> stateMap, + final Position[] positions) + { + Map<TopicWord, Set<Position>> transitions = new HashMap<TopicWord,Set<Position>>(); + + for(Position pos : state._positions) + { + if(pos._selfTransition) + { + Set<Position> dest = transitions.get(TopicWord.ANY_WORD); + if(dest == null) + { + dest = new HashSet<Position>(); + transitions.put(TopicWord.ANY_WORD,dest); + } + dest.add(pos); + } + + final int nextPos = pos._position + 1; + Position nextPosition = nextPos == positions.length ? ERROR_POSITION : positions[nextPos]; + + Set<Position> dest = transitions.get(pos._word); + if(dest == null) + { + dest = new HashSet<Position>(); + transitions.put(pos._word,dest); + } + dest.add(nextPosition); + + } + + Set<Position> anyWordTransitions = transitions.get(TopicWord.ANY_WORD); + if(anyWordTransitions != null) + { + for(Set<Position> dest : transitions.values()) + { + dest.addAll(anyWordTransitions); + } + } + + state._nextState = new HashMap<TopicWord, SimpleState>(); + + for(Map.Entry<TopicWord,Set<Position>> dest : transitions.entrySet()) + { + + if(dest.getValue().size()>1) + { + dest.getValue().remove(ERROR_POSITION); + } + Position loopingTerminal = null; + for(Position destPos : dest.getValue()) + { + if(destPos._selfTransition && destPos._endState) + { + loopingTerminal = destPos; + break; + } + } + + if(loopingTerminal!=null) + { + dest.setValue(Collections.singleton(loopingTerminal)); + } + else + { + Position anyLoop = null; + for(Position destPos : dest.getValue()) + { + if(destPos._followedByAnyLoop) + { + if(anyLoop == null || anyLoop._position<destPos._position) + { + anyLoop = destPos; + } + } + } + if(anyLoop != null) + { + Collection<Position> removals = new ArrayList<Position>(); + for(Position destPos : dest.getValue()) + { + if(destPos._position < anyLoop._position) + { + removals.add(destPos); + } + } + dest.getValue().removeAll(removals); + } + } + + SimpleState stateForEntry = stateMap.get(dest.getValue()); + if(stateForEntry == null) + { + stateForEntry = new SimpleState(); + stateForEntry._positions = dest.getValue(); + stateMap.put(dest.getValue(),stateForEntry); + calculateNextStates(stateForEntry, + stateMap, + positions); + } + state._nextState.put(dest.getKey(),stateForEntry); + + + + } + + // remove redundant transitions + SimpleState anyWordState = state._nextState.get(TopicWord.ANY_WORD); + if(anyWordState != null) + { + List<TopicWord> removeList = new ArrayList<TopicWord>(); + for(Map.Entry<TopicWord,SimpleState> entry : state._nextState.entrySet()) + { + if(entry.getValue() == anyWordState && entry.getKey() != TopicWord.ANY_WORD) + { + removeList.add(entry.getKey()); + } + } + for(TopicWord removeKey : removeList) + { + state._nextState.remove(removeKey); + } + } + + + } + + private List<TopicWord> createTopicWordList(final AMQShortString bindingKey) + { + AMQShortStringTokenizer tokens = bindingKey.tokenize(TOPIC_DELIMITER); + TopicWord previousWord = null; + + List<TopicWord> wordList = new ArrayList<TopicWord>(); + + while(tokens.hasMoreTokens()) + { + TopicWord nextWord = _dictionary.getOrCreateWord(tokens.nextToken()); + if(previousWord == TopicWord.WILDCARD_WORD) + { + + if(nextWord == TopicWord.WILDCARD_WORD) + { + // consecutive wildcards can be merged + // i.e. subsequent wildcards can be discarded + continue; + } + else if(nextWord == TopicWord.ANY_WORD) + { + // wildcard and anyword can be reordered to always put anyword first + wordList.set(wordList.size()-1,TopicWord.ANY_WORD); + nextWord = TopicWord.WILDCARD_WORD; + } + } + wordList.add(nextWord); + previousWord = nextWord; + + } + return wordList; + } + + + public static void main(String[] args) + { + + printMatches("#.b.*.*.*.*.*.h.#.j.*.*.*.*.*.*.q.#.r.*.*.*.*.*.*.*.*","a.b.c.d.e.f.g.h.i.j.k.l.m.n.o.p.q.r.s.t.u.v.w.x.y.z"); + printMatches(new String[]{ + "#.a.#", + "#.b.#", + "#.c.#", + "#.d.#", + "#.e.#", + "#.f.#", + "#.g.#", + "#.h.#", + "#.i.#", + "#.j.#", + "#.k.#", + "#.l.#", + "#.m.#", + "#.n.#", + "#.o.#", + "#.p.#", + "#.q.#" + + }, "a.b.c.d.e.f.g.h.i.j.k.l.m.n.o.p.q.r.s.t.u.v.w.x.y.z"); +/* + printMatches(new String[]{ + "#.a.#", + "#.b.#", + "#.c.#", + "#.d.#", + "#.e.#", + "#.f.#", + "#.g.#", + "#.h.#", + "#.i.#", + "#.j.#", + "#.k.#", + "#.l.#", + "#.m.#", + "#.n.#", + "#.o.#", + "#.p.#", + "#.q.#", + "#.r.#", + "#.s.#", + "#.t.#", + "#.u.#", + "#.v.#", + "#.w.#", + "#.x.#", + "#.y.#", + "#.z.#" + + + },"a.b"); + + printMatches("#.b.*.*.*.*.*.h.#.j.*.*.*.*.*.p.#.r.*.*.*.*.*","a.b.c.d.e.f.g.h.i.j.k.l.m.n.o.p.q.r.s.t.u.v.w.x.y.z"); + printMatches("#.b.*.*.*.*.*.h.#.j.*.*.*.*.*.p.#.r.*.*.*.*.*.*.*.*","a.b.c.d.e.f.g.h.i.j.k.l.m.n.o.p.q.r.s.t.u.v.w.x.y.z"); + printMatches("a.#.b.#","a.b.b.b.b.b.b.b.c"); + +*/ + + printMatches("",""); + printMatches("a","a"); + printMatches("a",""); + printMatches("","a"); + printMatches("a.b","a.b"); + printMatches("a","a.b"); + printMatches("a.b","a"); + printMatches("*","a"); + printMatches("*.b","a.b"); + printMatches("*.*","a.b"); + printMatches("a.*","a.b"); + printMatches("a.*.#","a.b"); + printMatches("a.#.b","a.b"); + + printMatches("#.b","a"); + printMatches("#.b","a.b"); + printMatches("#.a.b","a.b"); + + + printMatches("#",""); + printMatches("#","a"); + printMatches("#","a.b"); + printMatches("#.#","a.b"); + printMatches("#.*","a.b"); + + printMatches("#.a.b","a.b"); + printMatches("a.b.#","a.b"); + printMatches("a.#","a.b"); + printMatches("#.*.#","a.b"); + printMatches("#.*.b.#","a.b"); + printMatches("#.a.*.#","a.b"); + printMatches("#.a.#.b.#","a.b"); + printMatches("#.*.#.*.#","a.b"); + printMatches("*.#.*.#","a.b"); + printMatches("#.*.#.*","a.b"); + + + printMatches(new String[]{"a.#.b.#","a.*.#.b.#"},"a.b.b.b.b.b.b.b.c"); + + + printMatches(new String[]{"a.b", "a.c"},"a.b"); + printMatches(new String[]{"a.#", "a.c", "#.b"},"a.b"); + printMatches(new String[]{"a.#", "a.c", "#.b", "#", "*.*"},"a.b"); + + printMatches(new String[]{"a.b.c.d.e.#", "a.b.c.d.#", "a.b.c.d.*", "a.b.c.#", "#.e", "a.*.c.d.e","#.c.*.#.*.*"},"a.b.c.d.e"); + printMatches(new String[]{"a.b.c.d.e.#", "a.b.c.d.#", "a.b.c.d.*", "a.b.c.#", "#.e", "a.*.c.d.e","#.c.*.#.*.*"},"a.b.c.d.f.g"); + + + + + } + + private static void printMatches(final String[] bindingKeys, final String routingKey) + { + TopicMatcherDFAState sm = null; + Map<TopicMatcherResult, String> resultMap = new HashMap<TopicMatcherResult, String>(); + + TopicParser parser = new TopicParser(); + + long start = System.currentTimeMillis(); + for(int i = 0; i < bindingKeys.length; i++) + { + System.out.println((System.currentTimeMillis() - start) + ":\t" + bindingKeys[i]); + TopicMatcherResult r = new TopicMatcherResult(){}; + resultMap.put(r, bindingKeys[i]); + AMQShortString bindingKeyShortString = new AMQShortString(bindingKeys[i]); + + System.err.println("====================================================="); + System.err.println("Adding binding key: " + bindingKeyShortString); + System.err.println("-----------------------------------------------------"); + + + if(i==0) + { + sm = parser.createStateMachine(bindingKeyShortString, r); + } + else + { + sm = sm.mergeStateMachines(parser.createStateMachine(bindingKeyShortString, r)); + } + System.err.println(sm.reachableStates()); + System.err.println("====================================================="); + try + { + System.in.read(); + } + catch (IOException e) + { + e.printStackTrace(); //To change body of catch statement use File | Settings | File Templates. + } + } + AMQShortString routingKeyShortString = new AMQShortString(routingKey); + + Collection<TopicMatcherResult> results = sm.parse(parser._dictionary, routingKeyShortString); + Collection<String> resultStrings = new ArrayList<String>(); + + for(TopicMatcherResult result : results) + { + resultStrings.add(resultMap.get(result)); + } + + final ArrayList<String> nonMatches = new ArrayList<String>(Arrays.asList(bindingKeys)); + nonMatches.removeAll(resultStrings); + System.out.println("\""+routingKeyShortString+"\" matched with " + resultStrings + " DID NOT MATCH with " + nonMatches); + + + } + + private static void printMatches(String bindingKey, String routingKey) + { + printMatches(new String[] { bindingKey }, routingKey); + } + + + private static boolean matches(String bindingKey, String routingKey) + { + AMQShortString bindingKeyShortString = new AMQShortString(bindingKey); + AMQShortString routingKeyShortString = new AMQShortString(routingKey); + TopicParser parser = new TopicParser(); + + final TopicMatcherResult result = new TopicMatcherResult(){}; + + TopicMatcherDFAState sm = parser.createStateMachine(bindingKeyShortString, result); + return !sm.parse(parser._dictionary,routingKeyShortString).isEmpty(); + + } + +} diff --git a/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicWord.java b/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicWord.java new file mode 100644 index 0000000000..f14d70f8a1 --- /dev/null +++ b/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicWord.java @@ -0,0 +1,54 @@ +package org.apache.qpid.server.exchange.topic; + +import org.apache.qpid.framing.AMQShortString; + +import java.util.Map; +import java.util.HashMap; +import java.util.concurrent.ConcurrentHashMap; + +/* +* +* Licensed to the Apache Software Foundation (ASF) under one +* 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. +* +*/ +public final class TopicWord +{ + public static final TopicWord ANY_WORD = new TopicWord("*"); + public static final TopicWord WILDCARD_WORD = new TopicWord("#"); + private String _word; + + public TopicWord() + { + + } + + public TopicWord(String s) + { + _word = s; + } + + public TopicWord(final AMQShortString name) + { + _word = name.toString(); + } + + public String toString() + { + return _word; + } +} diff --git a/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicWordDictionary.java b/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicWordDictionary.java new file mode 100644 index 0000000000..65a0cd3107 --- /dev/null +++ b/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicWordDictionary.java @@ -0,0 +1,63 @@ +package org.apache.qpid.server.exchange.topic; + +import org.apache.qpid.framing.AMQShortString; + +import java.util.concurrent.ConcurrentHashMap; + +/* +* +* 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. +* +*/ +public class TopicWordDictionary +{ + private final ConcurrentHashMap<AMQShortString,TopicWord> _dictionary = + new ConcurrentHashMap<AMQShortString,TopicWord>(); + + + + public TopicWordDictionary() + { + _dictionary.put(new AMQShortString("*"), TopicWord.ANY_WORD); + _dictionary.put(new AMQShortString("#"), TopicWord.WILDCARD_WORD); + } + + + + + public TopicWord getOrCreateWord(AMQShortString name) + { + TopicWord word = _dictionary.putIfAbsent(name, new TopicWord(name)); + if(word == null) + { + word = _dictionary.get(name); + } + return word; + } + + + public TopicWord getWord(AMQShortString name) + { + TopicWord word = _dictionary.get(name); + if(word == null) + { + word = TopicWord.ANY_WORD; + } + return word; + } +} |