diff options
Diffstat (limited to 'java')
15 files changed, 340 insertions, 21 deletions
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 0c73e0f9f0..8603bc34aa 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 @@ -25,6 +25,8 @@ import org.apache.qpid.AMQException; import java.util.HashMap; import java.util.Map; +import org.apache.qpid.AMQUnknownExchangeType; +import org.apache.qpid.exchange.ExchangeDefaults; public class DefaultExchangeFactory implements ExchangeFactory { @@ -34,9 +36,10 @@ public class DefaultExchangeFactory implements ExchangeFactory public DefaultExchangeFactory() { - _exchangeClassMap.put("direct", org.apache.qpid.server.exchange.DestNameExchange.class); - _exchangeClassMap.put("topic", org.apache.qpid.server.exchange.DestWildExchange.class); - _exchangeClassMap.put("headers", org.apache.qpid.server.exchange.HeadersExchange.class); + _exchangeClassMap.put(ExchangeDefaults.DIRECT_EXCHANGE_CLASS, org.apache.qpid.server.exchange.DestNameExchange.class); + _exchangeClassMap.put(ExchangeDefaults.TOPIC_EXCHANGE_CLASS, org.apache.qpid.server.exchange.DestWildExchange.class); + _exchangeClassMap.put(ExchangeDefaults.HEADERS_EXCHANGE_CLASS, org.apache.qpid.server.exchange.HeadersExchange.class); + _exchangeClassMap.put(ExchangeDefaults.FANOUT_EXCHANGE_CLASS, org.apache.qpid.server.exchange.FanoutExchange.class); } public Exchange createExchange(String exchange, String type, boolean durable, boolean autoDelete, @@ -46,7 +49,7 @@ public class DefaultExchangeFactory implements ExchangeFactory Class<? extends Exchange> exchClass = _exchangeClassMap.get(type); if (exchClass == null) { - throw new AMQException(_logger, "Unknown exchange type: " + type); + throw new AMQUnknownExchangeType("Unknown exchange type: " + type); } try { 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 ef94918e59..02a8dbda6e 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 @@ -37,6 +37,8 @@ public class DefaultExchangeRegistry implements ExchangeRegistry */ private ConcurrentMap<String, Exchange> _exchangeMap = new ConcurrentHashMap<String, Exchange>(); + private Exchange _defaultExchange; + public DefaultExchangeRegistry(ExchangeFactory exchangeFactory) { //create 'standard' exchanges: @@ -52,9 +54,18 @@ public class DefaultExchangeRegistry implements ExchangeRegistry public void registerExchange(Exchange exchange) { + if(_defaultExchange == null) + { + setDefaultExchange(exchange); + } _exchangeMap.put(exchange.getName(), exchange); } + public void setDefaultExchange(Exchange exchange) + { + _defaultExchange = exchange; + } + public void unregisterExchange(String name, boolean inUse) throws AMQException { // TODO: check inUse argument @@ -71,7 +82,16 @@ public class DefaultExchangeRegistry implements ExchangeRegistry public Exchange getExchange(String name) { - return _exchangeMap.get(name); + + if(name == null || name.length() == 0) + { + return _defaultExchange; + } + else + { + return _exchangeMap.get(name); + } + } /** @@ -82,7 +102,7 @@ public class DefaultExchangeRegistry implements ExchangeRegistry public void routeContent(AMQMessage payload) throws AMQException { final String exchange = payload.getTransferBody().destination; - final Exchange exch = _exchangeMap.get(exchange); + final Exchange exch = getExchange(exchange); // there is a small window of opportunity for the exchange to be deleted in between // the JmsPublish being received (where the exchange is validated) and the final // content body being received (which triggers this method) diff --git a/java/broker/src/main/java/org/apache/qpid/server/exchange/DestNameExchange.java b/java/broker/src/main/java/org/apache/qpid/server/exchange/DestNameExchange.java index b777ae7d82..872cde56e2 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/exchange/DestNameExchange.java +++ b/java/broker/src/main/java/org/apache/qpid/server/exchange/DestNameExchange.java @@ -29,6 +29,7 @@ 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.registry.ApplicationRegistry; +import org.apache.qpid.exchange.ExchangeDefaults; import javax.management.JMException; import javax.management.MBeanException; @@ -228,4 +229,9 @@ public class DestNameExchange extends AbstractExchange { return !_index.getBindingsMap().isEmpty(); } + + public String getType() + { + return ExchangeDefaults.DIRECT_EXCHANGE_CLASS; + } } diff --git a/java/broker/src/main/java/org/apache/qpid/server/exchange/DestWildExchange.java b/java/broker/src/main/java/org/apache/qpid/server/exchange/DestWildExchange.java index 932632cde3..3064a7eabc 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/exchange/DestWildExchange.java +++ b/java/broker/src/main/java/org/apache/qpid/server/exchange/DestWildExchange.java @@ -29,6 +29,7 @@ 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.registry.ApplicationRegistry; +import org.apache.qpid.exchange.ExchangeDefaults; import javax.management.JMException; import javax.management.MBeanException; @@ -241,4 +242,10 @@ public class DestWildExchange extends AbstractExchange throw new AMQException("Exception occured in creating the topic exchenge mbean", ex); } } + + + public String getType() + { + return ExchangeDefaults.TOPIC_EXCHANGE_CLASS; + } } 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 824e85dc5c..9fcbe2a871 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 @@ -28,6 +28,7 @@ import org.apache.qpid.server.queue.AMQMessage; public interface Exchange { String getName(); + String getType(); void initialise(String name, boolean durable, int ticket, boolean autoDelete) throws AMQException; diff --git a/java/broker/src/main/java/org/apache/qpid/server/exchange/ExchangeRegistry.java b/java/broker/src/main/java/org/apache/qpid/server/exchange/ExchangeRegistry.java index 4a0a6a0ee1..2fc64135a2 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/exchange/ExchangeRegistry.java +++ b/java/broker/src/main/java/org/apache/qpid/server/exchange/ExchangeRegistry.java @@ -37,4 +37,6 @@ public interface ExchangeRegistry extends MessageRouter void unregisterExchange(String name, boolean inUse) throws ExchangeInUseException, AMQException; Exchange getExchange(String name); + + void setDefaultExchange(Exchange exchange); } 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 new file mode 100644 index 0000000000..11b8061105 --- /dev/null +++ b/java/broker/src/main/java/org/apache/qpid/server/exchange/FanoutExchange.java @@ -0,0 +1,205 @@ +package org.apache.qpid.server.exchange; + +import org.apache.log4j.Logger; +import org.apache.qpid.server.management.MBeanDescription; +import org.apache.qpid.server.management.MBeanConstructor; +import org.apache.qpid.server.queue.AMQQueue; +import org.apache.qpid.server.queue.AMQMessage; +import org.apache.qpid.server.registry.ApplicationRegistry; +import org.apache.qpid.framing.FieldTable; +import org.apache.qpid.framing.MessageTransferBody; +import org.apache.qpid.AMQException; +import org.apache.qpid.exchange.ExchangeDefaults; + +import javax.management.openmbean.*; +import javax.management.JMException; +import javax.management.MBeanException; +import java.util.List; +import java.util.Map; +import java.util.ArrayList; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.CopyOnWriteArraySet; + +public class FanoutExchange extends AbstractExchange +{ + private static final Logger _logger = Logger.getLogger(FanoutExchange.class); + + /** + * Maps from queue name to queue instances + */ + private final CopyOnWriteArraySet<AMQQueue> _queues = new CopyOnWriteArraySet<AMQQueue>(); + + /** + * MBean class implementing the management interfaces. + */ + @MBeanDescription("Management Bean for Fanout Exchange") + private final class FanoutExchangeMBean extends ExchangeMBean + { + // open mbean data types for representing exchange bindings + private String[] _bindingItemNames = {"Routing Key", "Queue Names"}; + private String[] _bindingItemIndexNames = {_bindingItemNames[0]}; + private OpenType[] _bindingItemTypes = new OpenType[2]; + private CompositeType _bindingDataType = null; + private TabularType _bindinglistDataType = null; + private TabularDataSupport _bindingList = null; + + @MBeanConstructor("Creates an MBean for AMQ fanout exchange") + public FanoutExchangeMBean() throws JMException + { + super(); + _exchangeType = "fanout"; + init(); + } + + /** + * initialises the OpenType objects. + */ + private void init() throws OpenDataException + { + _bindingItemTypes[0] = SimpleType.STRING; + _bindingItemTypes[1] = new ArrayType(1, SimpleType.STRING); + _bindingDataType = new CompositeType("Exchange Binding", "Routing key and Queue names", + _bindingItemNames, _bindingItemNames, _bindingItemTypes); + _bindinglistDataType = new TabularType("Exchange Bindings", "Exchange Bindings for " + getName(), + _bindingDataType, _bindingItemIndexNames); + } + + public TabularData bindings() throws OpenDataException + { + + _bindingList = new TabularDataSupport(_bindinglistDataType); + + for (AMQQueue queue : _queues) + { + String queueName = queue.getName().toString(); + + + + Object[] bindingItemValues = {queueName, new String[] {queueName}}; + CompositeData bindingData = new CompositeDataSupport(_bindingDataType, _bindingItemNames, bindingItemValues); + _bindingList.put(bindingData); + } + + return _bindingList; + } + + public void createNewBinding(String queueName, String binding) throws JMException + { + AMQQueue queue = ApplicationRegistry.getInstance().getQueueRegistry().getQueue(queueName); + if (queue == null) + { + throw new JMException("Queue \"" + queueName + "\" is not registered with the exchange."); + } + + try + { + registerQueue(binding, queue, null); + queue.bind(binding, FanoutExchange.this); + } + catch (AMQException ex) + { + throw new MBeanException(ex); + } + } + + }// End of MBean class + + + protected ExchangeMBean createMBean() throws AMQException + { + try + { + return new FanoutExchange.FanoutExchangeMBean(); + } + catch (JMException ex) + { + _logger.error("Exception occured in creating the direct exchange mbean", ex); + throw new AMQException("Exception occured in creating the direct exchange mbean", ex); + } + } + + public String getType() + { + return ExchangeDefaults.FANOUT_EXCHANGE_CLASS; + } + + public void registerQueue(String routingKey, AMQQueue queue, FieldTable args) throws AMQException + { + assert queue != null; + + if (_queues.contains(queue)) + { + _logger.debug("Queue " + queue + " is already registered"); + } + else + { + _queues.add(queue); + _logger.debug("Binding queue " + queue + " with routing key " + routingKey + " to exchange " + this); + } + } + + public void deregisterQueue(String routingKey, AMQQueue queue) throws AMQException + { + assert queue != null; + assert routingKey != null; + + if (!_queues.remove(queue)) + { + throw new AMQException("Queue " + queue + " was not registered with exchange " + this.getName() + + ". "); + } + } + + public void route(AMQMessage payload) throws AMQException + { + MessageTransferBody transferBody = payload.getTransferBody(); + if (_queues == null || _queues.isEmpty()) + { + String msg = "No queues bound to " + this; + // XXX + /*if (publishBody.mandatory) + { + throw new NoRouteException(msg, payload); + } + else*/ + { + _logger.warn(msg); + } + } + else + { + if (_logger.isDebugEnabled()) + { + _logger.debug("Publishing message to queue " + _queues); + } + + for (AMQQueue q : _queues) + { + q.deliver(payload); + } + } + } + + public boolean isBound(String routingKey, AMQQueue queue) throws AMQException + { + return _queues.contains(queue); + } + + public boolean isBound(String routingKey) throws AMQException + { + + return _queues != null && !_queues.isEmpty(); + } + + public boolean isBound(AMQQueue queue) throws AMQException + { + + + return _queues.contains(queue); + } + + public boolean hasBindings() throws AMQException + { + return !_queues.isEmpty(); + } +} 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 5534f9d227..60ec059336 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 @@ -28,6 +28,7 @@ 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.registry.ApplicationRegistry; +import org.apache.qpid.exchange.ExchangeDefaults; import javax.management.JMException; import javax.management.openmbean.*; @@ -285,4 +286,9 @@ public class HeadersExchange extends AbstractExchange return o instanceof Registration && ((Registration) o).queue.equals(queue); } } + + public String getType() + { + return ExchangeDefaults.HEADERS_EXCHANGE_CLASS; + } } diff --git a/java/broker/src/main/java/org/apache/qpid/server/handler/ExchangeDeclareHandler.java b/java/broker/src/main/java/org/apache/qpid/server/handler/ExchangeDeclareHandler.java index 273c3881ae..99d0df82cd 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/handler/ExchangeDeclareHandler.java +++ b/java/broker/src/main/java/org/apache/qpid/server/handler/ExchangeDeclareHandler.java @@ -22,9 +22,12 @@ package org.apache.qpid.server.handler; import org.apache.log4j.Logger; import org.apache.qpid.AMQException; +import org.apache.qpid.AMQConnectionException; +import org.apache.qpid.AMQUnknownExchangeType; import org.apache.qpid.framing.AMQMethodBody; import org.apache.qpid.framing.ExchangeDeclareBody; import org.apache.qpid.framing.ExchangeDeclareOkBody; +import org.apache.qpid.protocol.AMQConstant; import org.apache.qpid.protocol.AMQMethodEvent; import org.apache.qpid.server.exchange.Exchange; import org.apache.qpid.server.exchange.ExchangeRegistry; @@ -68,9 +71,28 @@ public class ExchangeDeclareHandler implements StateAwareMethodListener<Exchange if (exchange == null) { - exchange = exchangeFactory.createExchange(body.exchange, body.type, body.durable, - body.passive, body.ticket); - exchangeRegistry.registerExchange(exchange); + if(body.passive && ((body.type == null) || body.type.length() ==0)) + { + throw body.getChannelException(AMQConstant.NOT_FOUND.getCode(), "Unknown exchange: " + body.exchange); + } + else + { + try + { + exchange = exchangeFactory.createExchange(body.exchange, body.type, body.durable, + body.passive, body.ticket); + exchangeRegistry.registerExchange(exchange); + } + catch(AMQUnknownExchangeType e) + { + throw body.getConnectionException(AMQConstant.COMMAND_INVALID.getCode(), "Unknown exchange type: " + body.type,e); + } + } + } + else if (!exchange.getType().equals(body.type)) + { + + throw new AMQConnectionException(AMQConstant.NOT_ALLOWED.getCode(), "Attempt to redeclare exchange: " + body.exchange + " of type " + exchange.getType() + " to " + body.type +".",body.getClazz(), body.getMethod(),body.getMajor(),body.getMinor()); } } if(!body.nowait) diff --git a/java/broker/src/main/java/org/apache/qpid/server/protocol/ExchangeInitialiser.java b/java/broker/src/main/java/org/apache/qpid/server/protocol/ExchangeInitialiser.java index d3ec70456f..d1d0bc7e17 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/protocol/ExchangeInitialiser.java +++ b/java/broker/src/main/java/org/apache/qpid/server/protocol/ExchangeInitialiser.java @@ -31,6 +31,9 @@ public class ExchangeInitialiser define(registry, factory, ExchangeDefaults.DIRECT_EXCHANGE_NAME, ExchangeDefaults.DIRECT_EXCHANGE_CLASS); define(registry, factory, ExchangeDefaults.TOPIC_EXCHANGE_NAME, ExchangeDefaults.TOPIC_EXCHANGE_CLASS); define(registry, factory, ExchangeDefaults.HEADERS_EXCHANGE_NAME, ExchangeDefaults.HEADERS_EXCHANGE_CLASS); + define(registry, factory, ExchangeDefaults.FANOUT_EXCHANGE_NAME, ExchangeDefaults.FANOUT_EXCHANGE_CLASS); + + registry.setDefaultExchange(registry.getExchange(ExchangeDefaults.DIRECT_EXCHANGE_NAME)); } private void define(ExchangeRegistry r, ExchangeFactory f, diff --git a/java/broker/src/main/java/org/apache/qpid/server/queue/ExchangeBindings.java b/java/broker/src/main/java/org/apache/qpid/server/queue/ExchangeBindings.java index 684e312fa3..57ca5f0f6e 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/queue/ExchangeBindings.java +++ b/java/broker/src/main/java/org/apache/qpid/server/queue/ExchangeBindings.java @@ -62,7 +62,7 @@ class ExchangeBindings public int hashCode() { - return exchange.hashCode() + routingKey.hashCode(); + return (exchange == null ? 0 : exchange.hashCode()) + (routingKey == null ? 0 : routingKey.hashCode()); } public boolean equals(Object o) diff --git a/java/common/src/main/java/org/apache/qpid/AMQUnknownExchangeType.java b/java/common/src/main/java/org/apache/qpid/AMQUnknownExchangeType.java new file mode 100644 index 0000000000..70f333a580 --- /dev/null +++ b/java/common/src/main/java/org/apache/qpid/AMQUnknownExchangeType.java @@ -0,0 +1,9 @@ +package org.apache.qpid; + +public class AMQUnknownExchangeType extends AMQException +{ + public AMQUnknownExchangeType(String message) + { + super(message); + } +} diff --git a/java/common/src/main/java/org/apache/qpid/exchange/ExchangeDefaults.java b/java/common/src/main/java/org/apache/qpid/exchange/ExchangeDefaults.java index e67a5ba7fe..dca8075f7f 100644 --- a/java/common/src/main/java/org/apache/qpid/exchange/ExchangeDefaults.java +++ b/java/common/src/main/java/org/apache/qpid/exchange/ExchangeDefaults.java @@ -33,4 +33,8 @@ public class ExchangeDefaults public final static String HEADERS_EXCHANGE_NAME = "amq.match"; public final static String HEADERS_EXCHANGE_CLASS = "headers"; + + public final static String FANOUT_EXCHANGE_NAME = "amq.fanout"; + + public final static String FANOUT_EXCHANGE_CLASS = "fanout"; } diff --git a/java/common/src/main/java/org/apache/qpid/framing/AMQType.java b/java/common/src/main/java/org/apache/qpid/framing/AMQType.java index ad07634554..23c1929205 100644 --- a/java/common/src/main/java/org/apache/qpid/framing/AMQType.java +++ b/java/common/src/main/java/org/apache/qpid/framing/AMQType.java @@ -65,27 +65,46 @@ public enum AMQType public int getEncodingSize(Object value)
{
- // TODO : fixme
- throw new UnsupportedOperationException();
+ return EncodingUtils.unsignedIntegerLength();
}
-
- public Object toNativeValue(Object value)
+ public Long toNativeValue(Object value)
{
- // TODO : fixme
- throw new UnsupportedOperationException();
+ if (value instanceof Long)
+ {
+ return (Long) value;
+ }
+ else if (value instanceof Integer)
+ {
+ return ((Integer) value).longValue();
+ }
+ else if (value instanceof Short)
+ {
+ return ((Short) value).longValue();
+ }
+ else if (value instanceof Byte)
+ {
+ return ((Byte) value).longValue();
+ }
+ else if ((value instanceof String) || (value == null))
+ {
+ return Long.valueOf((String)value);
+ }
+ else
+ {
+ throw new NumberFormatException("Cannot convert: " + value + "(" +
+ value.getClass().getName() + ") to int.");
+ }
}
public void writeValueImpl(Object value, ByteBuffer buffer)
{
- // TODO : fixme
- throw new UnsupportedOperationException();
+ EncodingUtils.writeUnsignedInteger(buffer, (Long) value);
}
public Object readValueFromBuffer(ByteBuffer buffer)
{
- // TODO : fixme
- throw new UnsupportedOperationException();
+ return EncodingUtils.readUnsignedInteger(buffer);
}
},
diff --git a/java/common/src/main/java/org/apache/qpid/framing/EncodingUtils.java b/java/common/src/main/java/org/apache/qpid/framing/EncodingUtils.java index 1d7dd33c76..c88c5d3bd3 100644 --- a/java/common/src/main/java/org/apache/qpid/framing/EncodingUtils.java +++ b/java/common/src/main/java/org/apache/qpid/framing/EncodingUtils.java @@ -199,7 +199,7 @@ public class EncodingUtils } - public static long unsignedIntegerLength() + public static int unsignedIntegerLength() { return 4; } @@ -627,4 +627,16 @@ public class EncodingUtils writeByte(buffer, (byte) character); } + public static long readUnsignedInteger(ByteBuffer buffer) + { + long l = 0xFF & buffer.get(); + l <<=8; + l = l | (0xFF & buffer.get()); + l <<=8; + l = l | (0xFF & buffer.get()); + l <<=8; + l = l | (0xFF & buffer.get()); + + return l; + } } |