You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by rg...@apache.org on 2010/01/31 01:31:57 UTC
svn commit: r904934 [6/11] - in /qpid/trunk/qpid/java:
broker-plugins/src/main/java/org/apache/qpid/extras/exchanges/diagnostic/
broker-plugins/src/main/java/org/apache/qpid/extras/exchanges/example/
broker/ broker/src/main/java/org/apache/qpid/qmf/ br...
Added: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/TopicExchangeMBean.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/TopicExchangeMBean.java?rev=904934&view=auto
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/TopicExchangeMBean.java (added)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/TopicExchangeMBean.java Sun Jan 31 00:31:49 2010
@@ -0,0 +1,74 @@
+/*
+ *
+ * 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.qpid.management.common.mbeans.annotations.MBeanDescription;
+import org.apache.qpid.management.common.mbeans.annotations.MBeanConstructor;
+import org.apache.qpid.server.binding.Binding;
+
+import javax.management.JMException;
+import javax.management.openmbean.*;
+import java.util.List;
+import java.util.Map;
+import java.util.HashMap;
+import java.util.ArrayList;
+
+/** TopicExchangeMBean class implements the management interface for the Topic exchanges. */
+@MBeanDescription("Management Bean for Topic Exchange")
+final class TopicExchangeMBean extends AbstractExchangeMBean<TopicExchange>
+{
+ private TopicExchange _topicExchange;
+
+ @MBeanConstructor("Creates an MBean for AMQ topic exchange")
+ public TopicExchangeMBean(final TopicExchange topicExchange) throws JMException
+ {
+ super(topicExchange);
+ init();
+ }
+
+ /** returns exchange bindings in tabular form */
+ public TabularData bindings() throws OpenDataException
+ {
+ TabularDataSupport bindingList = new TabularDataSupport(_bindinglistDataType);
+ Map<String, List<String>> bindingData = new HashMap<String, List<String>>();
+ for (Binding binding : getExchange().getBindings())
+ {
+ String key = binding.getBindingKey();
+ List<String> queueNames = bindingData.get(key);
+ if(queueNames == null)
+ {
+ queueNames = new ArrayList<String>();
+ bindingData.put(key, queueNames);
+ }
+ queueNames.add(binding.getQueue().getNameShortString().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, COMPOSITE_ITEM_NAMES, bindingItemValues);
+ bindingList.put(bindingCompositeData);
+ }
+
+ return bindingList;
+ }
+
+} // End of MBean class
Added: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicBinding.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicBinding.java?rev=904934&view=auto
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicBinding.java (added)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicBinding.java Sun Jan 31 00:31:49 2010
@@ -0,0 +1,70 @@
+/*
+ *
+ * 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.topic;
+
+import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.framing.FieldTable;
+import org.apache.qpid.server.queue.AMQQueue;
+import org.apache.qpid.server.exchange.TopicExchange;
+
+public class TopicBinding
+{
+ private final AMQShortString _bindingKey;
+ private final AMQQueue _queue;
+ private final FieldTable _args;
+
+ public TopicBinding(AMQShortString bindingKey, AMQQueue queue, FieldTable args)
+ {
+ _bindingKey = bindingKey;
+ _queue = queue;
+ _args = args;
+ }
+
+ 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 TopicBinding)
+ {
+ TopicBinding other = (TopicBinding) o;
+ return (_queue == other._queue)
+ && ((_bindingKey == null) ? other._bindingKey == null : _bindingKey.equals(other._bindingKey));
+ }
+ return false;
+ }
+}
Added: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicExchangeResult.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicExchangeResult.java?rev=904934&view=auto
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicExchangeResult.java (added)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicExchangeResult.java Sun Jan 31 00:31:49 2010
@@ -0,0 +1,179 @@
+/*
+ *
+ * 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.topic;
+
+import org.apache.qpid.server.queue.AMQQueue;
+import org.apache.qpid.server.filter.MessageFilter;
+import org.apache.qpid.server.message.InboundMessage;
+
+import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
+
+public final class TopicExchangeResult implements TopicMatcherResult
+{
+ private final Map<AMQQueue, Integer> _unfilteredQueues = new ConcurrentHashMap<AMQQueue, Integer>();
+ private final ConcurrentHashMap<AMQQueue, Map<MessageFilter,Integer>> _filteredQueues = new ConcurrentHashMap<AMQQueue, Map<MessageFilter, 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 Collection<AMQQueue> getUnfilteredQueues()
+ {
+ return _unfilteredQueues.keySet();
+ }
+
+
+ public void addFilteredQueue(AMQQueue queue, MessageFilter filter)
+ {
+ Map<MessageFilter,Integer> filters = _filteredQueues.get(queue);
+ if(filters == null)
+ {
+ filters = new ConcurrentHashMap<MessageFilter,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 filter)
+ {
+ Map<MessageFilter,Integer> filters = _filteredQueues.get(queue);
+ if(filters != null)
+ {
+ Integer instances = filters.get(filter);
+ if(instances != null)
+ {
+ if(instances == 1)
+ {
+ filters.remove(filter);
+ if(filters.isEmpty())
+ {
+ _filteredQueues.remove(queue);
+ }
+ }
+ else
+ {
+ filters.put(filter, instances - 1);
+ }
+ }
+
+ }
+
+ }
+
+ public void replaceQueueFilter(AMQQueue queue,
+ MessageFilter oldFilter,
+ MessageFilter newFilter)
+ {
+ Map<MessageFilter,Integer> filters = _filteredQueues.get(queue);
+ Map<MessageFilter,Integer> newFilters = new ConcurrentHashMap<MessageFilter,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(InboundMessage 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, Integer>> entry : _filteredQueues.entrySet())
+ {
+ if(!queues.contains(entry.getKey()))
+ {
+ for(MessageFilter filter : entry.getValue().keySet())
+ {
+ if(filter.matches(msg))
+ {
+ queues.add(entry.getKey());
+ }
+ }
+ }
+ }
+ }
+ return queues;
+ }
+
+}
Added: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicNormalizer.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicNormalizer.java?rev=904934&view=auto
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicNormalizer.java (added)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicNormalizer.java Sun Jan 31 00:31:49 2010
@@ -0,0 +1,96 @@
+/*
+ *
+ * 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.topic;
+
+import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.framing.AMQShortStringTokenizer;
+import org.apache.qpid.server.exchange.TopicExchange;
+
+import java.util.List;
+import java.util.ArrayList;
+
+public class TopicNormalizer
+{
+ private static final byte TOPIC_SEPARATOR = (byte)'.';
+ private static final byte HASH_BYTE = (byte)'#';
+ private static final byte STAR_BYTE = (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("#");
+
+ public static AMQShortString normalize(AMQShortString routingKey)
+ {
+ if(routingKey == null)
+ {
+ return AMQShortString.EMPTY_STRING;
+ }
+ else if(!(routingKey.contains(HASH_BYTE) || routingKey.contains(STAR_BYTE)))
+ {
+ return routingKey;
+ }
+ else
+ {
+ 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;
+ }
+ }
+}
Added: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/federation/Bridge.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/federation/Bridge.java?rev=904934&view=auto
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/federation/Bridge.java (added)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/federation/Bridge.java Sun Jan 31 00:31:49 2010
@@ -0,0 +1,807 @@
+/*
+ *
+ * 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.federation;
+
+import org.apache.qpid.AMQException;
+import org.apache.qpid.server.binding.Binding;
+import org.apache.qpid.server.configuration.BridgeConfig;
+import org.apache.qpid.server.configuration.BridgeConfigType;
+import org.apache.qpid.server.configuration.ConfiguredObject;
+import org.apache.qpid.server.exchange.Exchange;
+import org.apache.qpid.server.exchange.ExchangeRegistry;
+import org.apache.qpid.server.flow.FlowCreditManager_0_10;
+import org.apache.qpid.server.flow.WindowCreditManager;
+import org.apache.qpid.server.message.MessageMetaData_0_10;
+import org.apache.qpid.server.message.MessageTransferMessage;
+import org.apache.qpid.server.message.ServerMessage;
+import org.apache.qpid.server.queue.AMQQueue;
+import org.apache.qpid.server.queue.AMQQueueFactory;
+import org.apache.qpid.server.queue.BaseQueue;
+import org.apache.qpid.server.queue.QueueRegistry;
+import org.apache.qpid.server.store.MessageStore;
+import org.apache.qpid.server.store.StoredMessage;
+import org.apache.qpid.server.subscription.Subscription_0_10;
+import org.apache.qpid.server.transport.ServerSession;
+import org.apache.qpid.server.txn.AutoCommitTransaction;
+import org.apache.qpid.server.txn.ServerTransaction;
+import org.apache.qpid.server.virtualhost.VirtualHost;
+import org.apache.qpid.transport.DeliveryProperties;
+import org.apache.qpid.transport.MessageAcceptMode;
+import org.apache.qpid.transport.MessageAcquireMode;
+import org.apache.qpid.transport.MessageCreditUnit;
+import org.apache.qpid.transport.MessageFlowMode;
+import org.apache.qpid.transport.MessageReject;
+import org.apache.qpid.transport.MessageRejectCode;
+import org.apache.qpid.transport.MessageTransfer;
+import org.apache.qpid.transport.Option;
+import org.apache.qpid.transport.RangeSet;
+import org.apache.qpid.transport.Session;
+import org.apache.qpid.transport.SessionException;
+import org.apache.qpid.transport.SessionListener;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
+public class Bridge implements BridgeConfig
+{
+ private final boolean _durable;
+ private final boolean _dynamic;
+ private final boolean _queueBridge;
+ private final boolean _localSource;
+ private final String _source;
+ private final String _destination;
+ private final String _key;
+ private final String _tag;
+ private final String _excludes;
+ private final BrokerLink _link;
+ private UUID _id;
+ private long _createTime = System.currentTimeMillis();
+
+ private Session _session;
+
+ private BridgeImpl _delegate;
+
+ private final int _bridgeNo;
+ private AutoCommitTransaction _transaction;
+
+ public Bridge(final BrokerLink brokerLink,
+ final int bridgeNo,
+ final boolean durable,
+ final boolean dynamic,
+ final boolean srcIsQueue,
+ final boolean srcIsLocal,
+ final String src,
+ final String dest,
+ final String key,
+ final String tag,
+ final String excludes)
+ {
+ _link = brokerLink;
+ _bridgeNo = bridgeNo;
+ _durable = durable;
+ _dynamic = dynamic;
+ _queueBridge = srcIsQueue;
+ _localSource = srcIsLocal;
+ _source = src;
+ _destination = dest;
+ _key = key;
+ _tag = tag;
+ _excludes = excludes;
+ _id = brokerLink.getConfigStore().createId();
+
+ _transaction = new AutoCommitTransaction(getVirtualHost().getMessageStore());
+
+ if(dynamic)
+ {
+ if(srcIsLocal)
+ {
+ // TODO
+ }
+ else
+ {
+ if(srcIsQueue)
+ {
+ // TODO
+ }
+ else
+ {
+ _delegate = new DynamicExchangeBridge();
+ }
+ }
+ }
+ else
+ {
+ if(srcIsLocal)
+ {
+ if(srcIsQueue)
+ {
+ _delegate = new StaticQueuePushBridge();
+ }
+ else
+ {
+ _delegate = new StaticExchangePushBridge();
+ }
+ }
+ else
+ {
+ if(srcIsQueue)
+ {
+ _delegate = new StaticQueuePullBridge();
+ }
+ else
+ {
+ _delegate = new StaticExchangePullBridge();
+ }
+ }
+ }
+ }
+
+ public UUID getId()
+ {
+ return _id;
+ }
+
+ public BridgeConfigType getConfigType()
+ {
+ return BridgeConfigType.getInstance();
+ }
+
+ public ConfiguredObject getParent()
+ {
+ return getLink();
+ }
+
+ public boolean isDurable()
+ {
+ return _durable;
+ }
+
+ public boolean isDynamic()
+ {
+ return _dynamic;
+ }
+
+ public boolean isQueueBridge()
+ {
+ return _queueBridge;
+ }
+
+ public boolean isLocalSource()
+ {
+ return _localSource;
+ }
+
+ public String getSource()
+ {
+ return _source;
+ }
+
+ public String getDestination()
+ {
+ return _destination;
+ }
+
+ public String getKey()
+ {
+ return _key;
+ }
+
+ public String getTag()
+ {
+ return _tag;
+ }
+
+ public String getExcludes()
+ {
+ return _excludes;
+ }
+
+ public BrokerLink getLink()
+ {
+ return _link;
+ }
+
+ public Integer getChannelId()
+ {
+ return (_session == null) ? 0 : _session.getChannel();
+ }
+
+ public int getAckBatching()
+ {
+ return 0;
+ }
+
+ public long getCreateTime()
+ {
+ return _createTime;
+ }
+
+ @Override
+ public boolean equals(final Object o)
+ {
+ if (this == o)
+ {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass())
+ {
+ return false;
+ }
+
+ final Bridge bridge = (Bridge) o;
+
+ if (_durable != bridge._durable)
+ {
+ return false;
+ }
+ if (_dynamic != bridge._dynamic)
+ {
+ return false;
+ }
+ if (_localSource != bridge._localSource)
+ {
+ return false;
+ }
+ if (_queueBridge != bridge._queueBridge)
+ {
+ return false;
+ }
+ if (_destination != null ? !_destination.equals(bridge._destination) : bridge._destination != null)
+ {
+ return false;
+ }
+ if (_excludes != null ? !_excludes.equals(bridge._excludes) : bridge._excludes != null)
+ {
+ return false;
+ }
+ if (_key != null ? !_key.equals(bridge._key) : bridge._key != null)
+ {
+ return false;
+ }
+ if (_source != null ? !_source.equals(bridge._source) : bridge._source != null)
+ {
+ return false;
+ }
+ if (_tag != null ? !_tag.equals(bridge._tag) : bridge._tag != null)
+ {
+ return false;
+ }
+
+ return true;
+ }
+
+ @Override
+ public int hashCode()
+ {
+ int result = (_durable ? 1 : 0);
+ result = 31 * result + (_dynamic ? 1 : 0);
+ result = 31 * result + (_queueBridge ? 1 : 0);
+ result = 31 * result + (_localSource ? 1 : 0);
+ result = 31 * result + (_source != null ? _source.hashCode() : 0);
+ result = 31 * result + (_destination != null ? _destination.hashCode() : 0);
+ result = 31 * result + (_key != null ? _key.hashCode() : 0);
+ result = 31 * result + (_tag != null ? _tag.hashCode() : 0);
+ result = 31 * result + (_excludes != null ? _excludes.hashCode() : 0);
+ return result;
+ }
+
+ public void setSession(final Session session)
+ {
+ _session = session;
+ _delegate.setSession(session);
+ }
+
+ private long getMessageWindowSize()
+ {
+ return 10l;
+ }
+
+
+ VirtualHost getVirtualHost()
+ {
+ return _link.getVirtualHost();
+ }
+
+ public void close()
+ {
+ // TODO
+ _delegate.close();
+ _session = null;
+ }
+
+
+ private interface BridgeImpl
+ {
+ void setSession(Session session);
+
+ void close();
+ }
+
+ private abstract class AbstractPullBridge implements BridgeImpl, SessionListener
+ {
+ public final void setSession(final Session session)
+ {
+ session.setSessionListener(this);
+ onSession();
+
+ }
+
+ abstract void onSession();
+
+
+
+ public void message(final Session ssn, final MessageTransfer xfr)
+ {
+ ExchangeRegistry exchangeRegistry = getVirtualHost().getExchangeRegistry();
+
+ Exchange exchange = exchangeRegistry.getExchange(_destination);
+
+ // TODO - deal with exchange not existing
+
+ DeliveryProperties delvProps = null;
+ if(xfr.getHeader() != null && (delvProps = xfr.getHeader().get(DeliveryProperties.class)) != null && delvProps.hasTtl() && !delvProps.hasExpiration())
+ {
+ delvProps.setExpiration(System.currentTimeMillis() + delvProps.getTtl());
+ }
+
+ MessageMetaData_0_10 messageMetaData = new MessageMetaData_0_10(xfr);
+ final MessageStore store = getVirtualHost().getMessageStore();
+ StoredMessage<MessageMetaData_0_10> storeMessage = store.addMessage(messageMetaData);
+ storeMessage.addContent(0,xfr.getBody());
+ storeMessage.flushToStore();
+ MessageTransferMessage message = new MessageTransferMessage(storeMessage, ((ServerSession)_session).getReference());
+
+ ArrayList<? extends BaseQueue> queues = exchange.route(message);
+
+
+
+ if(queues != null && queues.size() != 0)
+ {
+ enqueue(message, queues);
+ }
+ else
+ {
+ if(delvProps == null || !delvProps.hasDiscardUnroutable() || !delvProps.getDiscardUnroutable())
+ {
+ if(xfr.getAcceptMode() == MessageAcceptMode.EXPLICIT)
+ {
+ RangeSet rejects = new RangeSet();
+ rejects.add(xfr.getId());
+ MessageReject reject = new MessageReject(rejects, MessageRejectCode.UNROUTABLE, "Unroutable");
+ ssn.invoke(reject);
+ }
+ else
+ {
+ Exchange alternate = exchange.getAlternateExchange();
+ if(alternate != null)
+ {
+ queues = alternate.route(message);
+ if(queues != null && queues.size() != 0)
+ {
+ enqueue(message, queues);
+ }
+ else
+ {
+ //TODO - log the message discard
+ }
+ }
+ else
+ {
+ //TODO - log the message discard
+ }
+
+
+ }
+ }
+
+
+ }
+
+ ssn.processed(xfr);
+
+ }
+
+
+ private void enqueue(final ServerMessage message, final ArrayList<? extends BaseQueue> queues)
+ {
+ _transaction.enqueue(queues,message, new ServerTransaction.Action()
+ {
+
+ BaseQueue[] _queues = queues.toArray(new BaseQueue[queues.size()]);
+
+ public void postCommit()
+ {
+ for(int i = 0; i < _queues.length; i++)
+ {
+ try
+ {
+ _queues[i].enqueue(message);
+ }
+ catch (AMQException e)
+ {
+ // TODO
+
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ public void onRollback()
+ {
+ // NO-OP
+ }
+ });
+
+ }
+
+ public void exception(final Session session, final SessionException exception)
+ {
+ // TODO - Handle exceptions
+ }
+
+ public void closed(final Session session)
+ {
+ // TODO - handle close
+ }
+
+ public void opened(final Session session)
+ {
+ // this method never called
+ }
+
+ public void resumed(final Session session)
+ {
+ // will never resume these sessions
+ }
+
+
+
+ }
+
+ private final class StaticExchangePullBridge extends AbstractPullBridge
+ {
+ private final String _tmpQueueName = "bridge_queue_" + _bridgeNo + "_" + _link.getFederationTag();;
+
+ public void onSession()
+ {
+
+ final HashMap<String, Object> options = new HashMap<String, Object>();
+ options.put("qpid.trace.exclude", _link.getFederationTag());
+ options.put("qpid.trace.id",_link.getRemoteFederationTag());
+ _session.queueDeclare(_tmpQueueName,null, options, Option.AUTO_DELETE, Option.EXCLUSIVE);
+ _session.sync();
+ // todo check exception
+ final Map<String,Object> bindingArgs = new HashMap<String,Object>();
+ _session.exchangeBind(_tmpQueueName, _source, _key, bindingArgs);
+ _session.sync();
+ // todo check exception
+
+ final Map<String,Object> subscribeOptions = Collections.EMPTY_MAP;
+ final String subName = String.valueOf(_bridgeNo);
+ _session.messageSubscribe(_tmpQueueName,
+ subName,MessageAcceptMode.NONE,MessageAcquireMode.PRE_ACQUIRED,null,0l, subscribeOptions);
+ _session.sync();
+ // todo check exception
+
+ _session.messageSetFlowMode(subName,MessageFlowMode.WINDOW);
+ _session.messageFlow(subName, MessageCreditUnit.MESSAGE, getMessageWindowSize());
+ _session.messageFlow(subName, MessageCreditUnit.BYTE, 0xFFFFFFFF);
+
+ }
+
+ public void close()
+ {
+ // TODO
+ }
+ }
+
+ private final class StaticQueuePullBridge extends AbstractPullBridge
+ {
+
+ public void onSession()
+ {
+
+ final Map<String,Object> subscribeOptions = Collections.EMPTY_MAP;
+ final String subName = String.valueOf(_bridgeNo);
+ _session.messageSubscribe(_source,
+ subName,MessageAcceptMode.NONE,MessageAcquireMode.PRE_ACQUIRED,null,0l, subscribeOptions);
+ _session.sync();
+ // todo check exception
+
+ _session.messageSetFlowMode(subName,MessageFlowMode.WINDOW);
+ _session.messageFlow(subName, MessageCreditUnit.MESSAGE, getMessageWindowSize());
+ _session.messageFlow(subName, MessageCreditUnit.BYTE, 0xFFFFFFFF);
+
+ }
+
+ public void close()
+ {
+ // TODO
+ }
+ }
+
+ private final class DynamicExchangeBridge extends AbstractPullBridge implements Exchange.BindingListener
+ {
+ private final String _tmpQueueName = "bridge_queue_" + _bridgeNo + "_" + _link.getFederationTag();
+
+ private final ConcurrentMap<Binding,Binding> _bindings = new ConcurrentHashMap<Binding,Binding>();
+
+
+ void onSession()
+ {
+
+
+ final HashMap<String, Object> options = new HashMap<String, Object>();
+ options.put("qpid.trace.exclude", _link.getFederationTag());
+ options.put("qpid.trace.id",_link.getRemoteFederationTag());
+ _session.queueDeclare(_tmpQueueName,null, options, Option.AUTO_DELETE, Option.EXCLUSIVE);
+ _session.sync();
+ // todo - check exception
+
+ final Map<String,Object> subscribeOptions = Collections.EMPTY_MAP;
+ final String subName = String.valueOf(_bridgeNo);
+ _session.messageSubscribe(_tmpQueueName,
+ subName,MessageAcceptMode.NONE,MessageAcquireMode.PRE_ACQUIRED,null,0l, subscribeOptions);
+ _session.sync();
+ // todo check exception
+ _session.messageSetFlowMode(subName,MessageFlowMode.WINDOW);
+ _session.messageFlow(subName, MessageCreditUnit.MESSAGE, getMessageWindowSize());
+ _session.messageFlow(subName, MessageCreditUnit.BYTE, 0xFFFFFFFF);
+ _session.sync();
+ // todo check exception
+
+
+ ExchangeRegistry exchangeRegistry = getVirtualHost().getExchangeRegistry();
+
+ Exchange exchange = exchangeRegistry.getExchange(_destination);
+
+ // TODO - check null
+
+ exchange.addBindingListener(this);
+
+ Collection<Binding> bindings = exchange.getBindings();
+ for(Binding binding : bindings)
+ {
+ propogateBinding(binding);
+ }
+
+ }
+
+ private void propogateBinding(final Binding binding)
+ {
+ if(_bindings.putIfAbsent(binding,binding)== null)
+ {
+ Map<String,Object> arguments = new HashMap<String,Object>(binding.getArguments());
+
+ if(arguments.get("qpid.fed.origin") == null)
+ {
+ arguments.put("qpid.fed.op","");
+ arguments.put("qpid.fed.origin",_link.getFederationTag());
+ arguments.put("qpid.fed.tags",_link.getFederationTag());
+ }
+ else
+ {
+ String tags = (String) arguments.get("qpid.fed.tags");
+ if(tags == null)
+ {
+ tags = _link.getFederationTag();
+ }
+ else
+ {
+ if(Arrays.asList(tags.split(",")).contains(_link.getFederationTag()))
+ {
+ return;
+ }
+ tags += "," + _link.getFederationTag();
+ }
+ arguments.put("qpid.fed.tags", tags);
+ }
+
+ _session.exchangeBind(_tmpQueueName, _source, binding.getBindingKey(), arguments);
+ _session.sync();
+ // TODO - check exception?
+
+ }
+ }
+
+ private void propogateBindingRemoval(final Binding binding)
+ {
+ if(_bindings.remove(binding) != null)
+ {
+ // TODO - this is wrong!!!!
+ _session.exchangeUnbind(_tmpQueueName, _source, binding.getBindingKey());
+ }
+ }
+
+
+ public void bindingAdded(final Exchange exchange, final Binding binding)
+ {
+ propogateBinding(binding);
+ }
+
+ public void bindingRemoved(final Exchange exchange, final Binding binding)
+ {
+ propogateBindingRemoval(binding);
+ }
+
+ public void close()
+ {
+ // TODO
+ }
+ }
+
+ private class StaticExchangePushBridge implements BridgeImpl, SessionListener
+ {
+ private final String _tmpQueueName = "bridge_queue_" + _bridgeNo + "_" + _link.getFederationTag();
+ private AMQQueue _queue;
+
+ public void setSession(final Session session)
+ {
+ session.setSessionListener(this);
+
+ ExchangeRegistry exchangeRegistry = getVirtualHost().getExchangeRegistry();
+
+ Exchange exchange = exchangeRegistry.getExchange(_source);
+
+ // TODO - Check null
+
+ final HashMap<String, Object> options = new HashMap<String, Object>();
+ options.put("qpid.trace.exclude", _link.getFederationTag());
+ options.put("qpid.trace.id",_link.getRemoteFederationTag());
+
+ _queue = AMQQueueFactory.createAMQQueueImpl(_tmpQueueName,
+ isDurable(),
+ _link.getFederationTag(),
+ false,
+ getVirtualHost(),
+ options);
+
+ FlowCreditManager_0_10 creditManager = new WindowCreditManager(0xFFFFFFFF,getMessageWindowSize());
+
+ Subscription_0_10 sub = new Subscription_0_10((ServerSession)session,
+ _destination,
+ MessageAcceptMode.NONE,
+ MessageAcquireMode.PRE_ACQUIRED,
+ MessageFlowMode.WINDOW,
+ creditManager, null);
+
+ ((ServerSession)session).register(_destination, sub);
+
+ try
+ {
+ _queue.registerSubscription(sub, true);
+ }
+ catch (AMQException e)
+ {
+ // TODO
+ throw new RuntimeException(e);
+ }
+
+ getVirtualHost().getBindingFactory().addBinding(_key, _queue, exchange, Collections.EMPTY_MAP);
+ }
+
+ public void close()
+ {
+ // TODO
+ }
+
+ public void opened(final Session session)
+ {
+ // this method never called
+ }
+
+ public void resumed(final Session session)
+ {
+ // this session will never be resumed
+ }
+
+ public void message(final Session ssn, final MessageTransfer xfr)
+ {
+ // messages should not be sent ... should probably log error
+ }
+
+ public void exception(final Session session, final SessionException exception)
+ {
+ // TODO
+ }
+
+ public void closed(final Session session)
+ {
+ // TODO
+ }
+ }
+
+ private class StaticQueuePushBridge implements BridgeImpl, SessionListener
+ {
+ private AMQQueue _queue;
+
+ public void setSession(final Session session)
+ {
+ session.setSessionListener(this);
+
+ QueueRegistry queueRegistry = getVirtualHost().getQueueRegistry();
+
+ _queue = queueRegistry.getQueue(_source);
+
+ // TODO - null check
+
+ FlowCreditManager_0_10 creditManager = new WindowCreditManager(0xFFFFFFFF,getMessageWindowSize());
+
+ Subscription_0_10 sub = new Subscription_0_10((ServerSession)session,
+ _destination,
+ MessageAcceptMode.NONE,
+ MessageAcquireMode.PRE_ACQUIRED,
+ MessageFlowMode.WINDOW,
+ creditManager, null);
+
+ ((ServerSession)session).register(_destination, sub);
+
+ try
+ {
+ _queue.registerSubscription(sub, false);
+ }
+ catch (AMQException e)
+ {
+ // TODO
+ throw new RuntimeException(e);
+ }
+
+ }
+
+ public void close()
+ {
+ // TODO
+ }
+
+ public void opened(final Session session)
+ {
+ // never called
+ }
+
+ public void resumed(final Session session)
+ {
+ // session will not resume
+ }
+
+ public void message(final Session ssn, final MessageTransfer xfr)
+ {
+ // should never be called ... should probably log error
+ }
+
+ public void exception(final Session session, final SessionException exception)
+ {
+ // TODO
+ }
+
+ public void closed(final Session session)
+ {
+ // TODO
+ }
+ }
+
+}
Added: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/federation/BrokerLink.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/federation/BrokerLink.java?rev=904934&view=auto
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/federation/BrokerLink.java (added)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/federation/BrokerLink.java Sun Jan 31 00:31:49 2010
@@ -0,0 +1,497 @@
+/*
+ *
+ * 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.federation;
+
+import org.apache.qpid.server.configuration.ConfigStore;
+import org.apache.qpid.server.configuration.ConfiguredObject;
+import org.apache.qpid.server.configuration.ConnectionConfig;
+import org.apache.qpid.server.configuration.ConnectionConfigType;
+import org.apache.qpid.server.configuration.LinkConfig;
+import org.apache.qpid.server.configuration.LinkConfigType;
+import org.apache.qpid.server.transport.ServerSession;
+import org.apache.qpid.server.virtualhost.VirtualHost;
+import org.apache.qpid.transport.Binary;
+import org.apache.qpid.transport.Connection;
+import org.apache.qpid.transport.ConnectionException;
+import org.apache.qpid.transport.ConnectionListener;
+import org.apache.qpid.transport.Session;
+import org.apache.qpid.transport.SessionDelegate;
+import org.apache.qpid.transport.TransportException;
+
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
+
+public class BrokerLink implements LinkConfig, ConnectionListener
+{
+
+ private static final int CORE_POOL_SIZE = 4;
+
+ private static final ScheduledThreadPoolExecutor _threadPool =
+ new ScheduledThreadPoolExecutor(CORE_POOL_SIZE);
+
+
+ private final String _transport;
+ private final String _host;
+ private final int _port;
+ private final String _remoteVhost;
+ private final boolean _durable;
+ private final String _authMechanism;
+ private final String _username;
+ private final String _password;
+ private final VirtualHost _virtualHost;
+ private UUID _id;
+ private AtomicBoolean _closing = new AtomicBoolean();
+ private final long _createTime = System.currentTimeMillis();
+ private Connection _qpidConnection;
+ private AtomicReference<Thread> _executor = new AtomicReference<Thread>();
+ private AtomicInteger _bridgeId = new AtomicInteger();
+
+ private final ConcurrentHashMap<Bridge,Bridge> _bridges = new ConcurrentHashMap<Bridge,Bridge>();
+ private final ConcurrentHashMap<Bridge,Bridge> _activeBridges = new ConcurrentHashMap<Bridge,Bridge>();
+ private final ConcurrentLinkedQueue<Bridge> _pendingBridges = new ConcurrentLinkedQueue<Bridge>();
+ private String _remoteFederationTag;
+
+ private ConnectionConfig _connectionConfig;
+ private ConnectionException _exception;
+ private String _lastErrorMessage;
+ private int _retryDelay = 1;
+ private final Runnable _makeConnectionTask = new Runnable()
+ {
+ public void run()
+ {
+ doMakeConnection();
+ }
+ };;
+ ;
+
+ public static enum State
+ {
+ OPERATIONAL,
+ DOWN,
+ ESTABLISHING,
+ DELETED
+ }
+
+
+ private volatile State _state = State.DOWN;
+
+ private static final AtomicReferenceFieldUpdater<BrokerLink, State> _stateUpdater =
+ AtomicReferenceFieldUpdater.newUpdater(BrokerLink.class, State.class, "_state");
+
+ private class ConnectionConfigAdapter implements ConnectionConfig
+ {
+
+ private UUID _id = BrokerLink.this.getConfigStore().createId();
+
+ public VirtualHost getVirtualHost()
+ {
+ return BrokerLink.this.getVirtualHost();
+ }
+
+ public String getAddress()
+ {
+ return _host+":"+_port;
+ }
+
+ public Boolean isIncoming()
+ {
+ return false;
+ }
+
+ public Boolean isSystemConnection()
+ {
+ return true;
+ }
+
+ public Boolean isFederationLink()
+ {
+ return true;
+ }
+
+ public String getAuthId()
+ {
+ return _username;
+ }
+
+ public String getRemoteProcessName()
+ {
+ return null;
+ }
+
+ public Integer getRemotePID()
+ {
+ return null;
+ }
+
+ public Integer getRemoteParentPID()
+ {
+ return null;
+ }
+
+ public ConfigStore getConfigStore()
+ {
+ return getVirtualHost().getConfigStore();
+ }
+
+ public UUID getId()
+ {
+ return _id;
+ }
+
+ public ConnectionConfigType getConfigType()
+ {
+ return ConnectionConfigType.getInstance();
+ }
+
+ public ConfiguredObject getParent()
+ {
+ return getVirtualHost();
+ }
+
+ public boolean isDurable()
+ {
+ return false;
+ }
+ }
+
+ private class SessionFactory implements Connection.SessionFactory
+ {
+
+ public Session newSession(final Connection conn, final Binary name, final long expiry)
+ {
+ return new ServerSession(conn, new SessionDelegate(), name, expiry, _connectionConfig);
+ }
+ };
+
+
+ public BrokerLink(final VirtualHost virtualHost,
+ final String transport,
+ final String host,
+ final int port,
+ final String remoteVhost,
+ final boolean durable,
+ final String authMechanism, final String username, final String password)
+ {
+ _virtualHost = virtualHost;
+ _transport = transport;
+ _host = host;
+ _port = port;
+ _remoteVhost = remoteVhost;
+ _durable = durable;
+ _authMechanism = authMechanism;
+ _username = username;
+ _password = password;
+ _id = virtualHost.getConfigStore().createId();
+ _qpidConnection = new Connection();
+ _connectionConfig = new ConnectionConfigAdapter();
+ _qpidConnection.addConnectionListener(this);
+
+
+ makeConnection();
+ }
+
+ private final boolean updateState(State expected, State newState)
+ {
+ return _stateUpdater.compareAndSet(this,expected,newState);
+ }
+
+ private void makeConnection()
+ {
+ _threadPool.execute(_makeConnectionTask);
+ }
+
+
+
+ private void doMakeConnection()
+ {
+ if(updateState(State.DOWN, State.ESTABLISHING))
+ {
+ try
+ {
+ _qpidConnection.connect(_host, _port, _remoteVhost, _username, _password, "ssl".equals(_transport), _authMechanism);
+
+ final Map<String,Object> serverProps = _qpidConnection.getServerProperties();
+ _remoteFederationTag = (String) serverProps.get("qpid.federation_tag");
+ if(_remoteFederationTag == null)
+ {
+ _remoteFederationTag = UUID.fromString(_transport+":"+_host+":"+_port).toString();
+ }
+ _qpidConnection.setSessionFactory(new SessionFactory());
+ _qpidConnection.setAuthorizationID(_username == null ? "" : _username);
+
+ updateState(State.ESTABLISHING, State.OPERATIONAL);
+
+ _retryDelay = 1;
+
+ for(Bridge bridge : _bridges.values())
+ {
+ if(_state != State.OPERATIONAL)
+ {
+ break;
+ }
+ addBridge(bridge);
+ }
+
+
+ }
+ catch (TransportException e)
+ {
+ _lastErrorMessage = e.getMessage();
+ if(_retryDelay < 60)
+ {
+ _retryDelay <<= 1;
+ }
+
+ updateState(State.ESTABLISHING, State.DOWN);
+ _activeBridges.clear();
+ scheduleConnectionRetry();
+ }
+ }
+ }
+
+ private void scheduleConnectionRetry()
+ {
+ if(_state != State.DELETED)
+ {
+ _threadPool.schedule(_makeConnectionTask, _retryDelay, TimeUnit.SECONDS);
+ }
+ }
+
+ public VirtualHost getVirtualHost()
+ {
+ return _virtualHost;
+ }
+
+ public String getTransport()
+ {
+ return _transport;
+ }
+
+ public String getHost()
+ {
+ return _host;
+ }
+
+ public int getPort()
+ {
+ return _port;
+ }
+
+ public String getRemoteVhost()
+ {
+ return _remoteVhost;
+ }
+
+ public UUID getId()
+ {
+ return _id;
+ }
+
+ public LinkConfigType getConfigType()
+ {
+ return LinkConfigType.getInstance();
+ }
+
+ public ConfiguredObject getParent()
+ {
+ return getVirtualHost();
+ }
+
+ public boolean isDurable()
+ {
+ return _durable;
+ }
+
+ public String getAuthMechanism()
+ {
+ return _authMechanism;
+ }
+
+ public String getUsername()
+ {
+ return _username;
+ }
+
+ public String getPassword()
+ {
+ return _password;
+ }
+
+ @Override
+ public boolean equals(final Object o)
+ {
+ if (this == o)
+ {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass())
+ {
+ return false;
+ }
+
+ final BrokerLink that = (BrokerLink) o;
+
+ if (_port != that._port)
+ {
+ return false;
+ }
+ if (_host != null ? !_host.equals(that._host) : that._host != null)
+ {
+ return false;
+ }
+ if (_remoteVhost != null ? !_remoteVhost.equals(that._remoteVhost) : that._remoteVhost != null)
+ {
+ return false;
+ }
+ if (_transport != null ? !_transport.equals(that._transport) : that._transport != null)
+ {
+ return false;
+ }
+
+ return true;
+ }
+
+ @Override
+ public int hashCode()
+ {
+ int result = _transport != null ? _transport.hashCode() : 0;
+ result = 31 * result + (_host != null ? _host.hashCode() : 0);
+ result = 31 * result + _port;
+ result = 31 * result + (_remoteVhost != null ? _remoteVhost.hashCode() : 0);
+ return result;
+ }
+
+ public void close()
+ {
+ if(_closing.compareAndSet(false,true))
+ {
+ // TODO - close connection
+ for(Bridge bridge : _bridges.values())
+ {
+ bridge.close();
+ }
+ _bridges.clear();
+
+ _virtualHost.removeBrokerConnection(this);
+ }
+ }
+
+ public long getCreateTime()
+ {
+ return _createTime;
+ }
+
+ public void createBridge(final boolean durable,
+ final boolean dynamic,
+ final boolean srcIsQueue,
+ final boolean srcIsLocal,
+ final String src,
+ final String dest,
+ final String key,
+ final String tag,
+ final String excludes)
+ {
+ if(!_closing.get())
+ {
+ Bridge bridge = new Bridge(this, _bridgeId.incrementAndGet(), durable,dynamic,srcIsQueue,srcIsLocal,src,dest,key,tag,excludes);
+ if(_bridges.putIfAbsent(bridge, bridge) == null)
+ {
+
+ addBridge(bridge);
+ }
+ }
+
+
+ }
+
+ private void addBridge(final Bridge bridge)
+ {
+ getConfigStore().addConfiguredObject(bridge);
+
+ if(_state == State.OPERATIONAL && (_activeBridges.putIfAbsent(bridge,bridge) == null))
+ {
+
+
+ Session session = _qpidConnection.createSession("Bridge("
+ + (bridge.isDurable() ? "durable" : "transient")
+ + "," + (bridge.isDynamic() ? "dynamic" : "static")
+ + "," + (bridge.isQueueBridge() ? "queue" : "exchange")
+ + "," + (bridge.isLocalSource() ? "local-src" : "remote-src")
+ + ",[Source: '" + bridge.getSource() + "']"
+ + ",[Destination: '" + bridge.getDestination() + "']"
+ + ",[Key: '" + bridge.getKey() + "']"
+ + ",[Tag: '" + bridge.getTag() + "']"
+ + ".[Excludes: '" + bridge.getExcludes() + "'])");
+ bridge.setSession(session);
+
+
+ if(_closing.get())
+ {
+ bridge.close();
+ }
+ }
+
+ }
+
+ public void opened(final Connection connection)
+ {
+ // this method not called
+ }
+
+ public void exception(final Connection connection, final ConnectionException exception)
+ {
+ _exception = exception;
+ _lastErrorMessage = exception.getMessage();
+
+ }
+
+ public void closed(final Connection connection)
+ {
+ State currentState = _state;
+ if(currentState != State.DOWN && currentState != State.DELETED && updateState(currentState, State.DOWN))
+ {
+ scheduleConnectionRetry();
+ }
+ }
+
+ public ConfigStore getConfigStore()
+ {
+ return getVirtualHost().getConfigStore();
+ }
+
+ public String getFederationTag()
+ {
+ return getVirtualHost().getFederationTag();
+ }
+
+ public String getRemoteFederationTag()
+ {
+ return _remoteFederationTag;
+ }
+}
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/JMSSelectorFilter.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/JMSSelectorFilter.java?rev=904934&r1=904933&r2=904934&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/JMSSelectorFilter.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/JMSSelectorFilter.java Sun Jan 31 00:31:49 2010
@@ -22,6 +22,7 @@
import org.apache.log4j.Logger;
import org.apache.qpid.AMQException;
+import org.apache.qpid.AMQInvalidArgumentException;
import org.apache.qpid.server.filter.jms.selector.SelectorParser;
import org.apache.qpid.server.queue.Filterable;
@@ -33,7 +34,7 @@
private String _selector;
private BooleanExpression _matcher;
- public JMSSelectorFilter(String selector) throws AMQException
+ public JMSSelectorFilter(String selector) throws AMQInvalidArgumentException
{
_selector = selector;
_matcher = new SelectorParser().parse(selector);
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicConsumeMethodHandler.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicConsumeMethodHandler.java?rev=904934&r1=904933&r2=904934&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicConsumeMethodHandler.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicConsumeMethodHandler.java Sun Jan 31 00:31:49 2010
@@ -27,7 +27,6 @@
import org.apache.qpid.server.AMQChannel;
import org.apache.qpid.server.protocol.AMQProtocolSession;
import org.apache.qpid.server.queue.AMQQueue;
-import org.apache.qpid.server.security.access.Permission;
import org.apache.qpid.server.state.AMQStateManager;
import org.apache.qpid.server.state.StateAwareMethodListener;
import org.apache.qpid.server.virtualhost.VirtualHost;
@@ -102,11 +101,11 @@
body.getNoLocal(), body.getNowait(), queue))
{
throw body.getConnectionException(AMQConstant.ACCESS_REFUSED, "Permission denied");
- }
+ }
else if (queue.isExclusive() && !queue.isDurable() && queue.getExclusiveOwner() != session)
{
throw body.getConnectionException(AMQConstant.NOT_ALLOWED,
- "Queue " + queue.getName() + " is exclusive, but not created on this Connection.");
+ "Queue " + queue.getNameShortString() + " is exclusive, but not created on this Connection.");
}
if (body.getConsumerTag() != null)
@@ -163,14 +162,14 @@
{
throw body.getChannelException(AMQConstant.ACCESS_REFUSED,
"Cannot subscribe to queue "
- + queue.getName()
+ + queue.getNameShortString()
+ " as it already has an existing exclusive consumer");
}
catch (AMQQueue.ExistingSubscriptionPreventsExclusive e)
{
throw body.getChannelException(AMQConstant.ACCESS_REFUSED,
"Cannot subscribe to queue "
- + queue.getName()
+ + queue.getNameShortString()
+ " exclusively as it already has a consumer");
}
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ExchangeDeclareHandler.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ExchangeDeclareHandler.java?rev=904934&r1=904933&r2=904934&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ExchangeDeclareHandler.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ExchangeDeclareHandler.java Sun Jan 31 00:31:49 2010
@@ -30,7 +30,6 @@
import org.apache.qpid.server.exchange.ExchangeFactory;
import org.apache.qpid.server.exchange.ExchangeRegistry;
import org.apache.qpid.server.protocol.AMQProtocolSession;
-import org.apache.qpid.server.security.access.Permission;
import org.apache.qpid.server.state.AMQStateManager;
import org.apache.qpid.server.state.StateAwareMethodListener;
import org.apache.qpid.server.virtualhost.VirtualHost;
@@ -109,10 +108,10 @@
}
}
}
- else if (!exchange.getType().equals(body.getType()))
+ else if (!exchange.getTypeShortString().equals(body.getType()))
{
- throw new AMQConnectionException(AMQConstant.NOT_ALLOWED, "Attempt to redeclare exchange: " + body.getExchange() + " of type " + exchange.getType() + " to " + body.getType() +".",body.getClazz(), body.getMethod(),body.getMajor(),body.getMinor(),null);
+ throw new AMQConnectionException(AMQConstant.NOT_ALLOWED, "Attempt to redeclare exchange: " + body.getExchange() + " of type " + exchange.getTypeShortString() + " to " + body.getType() +".",body.getClazz(), body.getMethod(),body.getMajor(),body.getMinor(),null);
}
}
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueueBindHandler.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueueBindHandler.java?rev=904934&r1=904933&r2=904934&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueueBindHandler.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueueBindHandler.java Sun Jan 31 00:31:49 2010
@@ -7,9 +7,9 @@
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
- *
+ *
* http://www.apache.org/licenses/LICENSE-2.0
- *
+ *
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
@@ -21,21 +21,27 @@
package org.apache.qpid.server.handler;
import org.apache.log4j.Logger;
+
import org.apache.qpid.AMQException;
-import org.apache.qpid.AMQInvalidRoutingKeyException;
-import org.apache.qpid.framing.*;
+import org.apache.qpid.framing.AMQMethodBody;
+import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.framing.FieldTable;
+import org.apache.qpid.framing.MethodRegistry;
+import org.apache.qpid.framing.QueueBindBody;
import org.apache.qpid.protocol.AMQConstant;
import org.apache.qpid.server.AMQChannel;
+import org.apache.qpid.server.binding.Binding;
import org.apache.qpid.server.exchange.Exchange;
import org.apache.qpid.server.exchange.ExchangeRegistry;
import org.apache.qpid.server.protocol.AMQProtocolSession;
import org.apache.qpid.server.queue.AMQQueue;
import org.apache.qpid.server.queue.QueueRegistry;
-import org.apache.qpid.server.security.access.Permission;
import org.apache.qpid.server.state.AMQStateManager;
import org.apache.qpid.server.state.StateAwareMethodListener;
import org.apache.qpid.server.virtualhost.VirtualHost;
+import java.util.Map;
+
public class QueueBindHandler implements StateAwareMethodListener<QueueBindBody>
{
private static final Logger _log = Logger.getLogger(QueueBindHandler.class);
@@ -80,7 +86,7 @@
if (body.getRoutingKey() == null)
{
- routingKey = queue.getName();
+ routingKey = queue.getNameShortString();
}
else
{
@@ -116,18 +122,26 @@
else if (queue.isExclusive() && !queue.isDurable() && queue.getExclusiveOwner() != session)
{
throw body.getConnectionException(AMQConstant.NOT_ALLOWED,
- "Queue " + queue.getName() + " is exclusive, but not created on this Connection.");
+ "Queue " + queue.getNameShortString() + " is exclusive, but not created on this Connection.");
}
if (!exch.isBound(routingKey, body.getArguments(), queue))
{
- queue.bind(exch, routingKey, body.getArguments());
+ String bindingKey = String.valueOf(routingKey);
+ Map<String,Object> arguments = FieldTable.convertToMap(body.getArguments());
+
+ if(!virtualHost.getBindingFactory().addBinding(bindingKey, queue, exch, arguments))
+ {
+ Binding oldBinding = virtualHost.getBindingFactory().getBinding(bindingKey, queue, exch, arguments);
+
+ Map<String, Object> oldArgs = oldBinding.getArguments();
+ if((oldArgs == null && !arguments.isEmpty()) || (oldArgs != null && !oldArgs.equals(arguments)))
+ {
+ virtualHost.getBindingFactory().replaceBinding(bindingKey, queue, exch, arguments);
+ }
+ }
}
}
- catch (AMQInvalidRoutingKeyException rke)
- {
- throw body.getChannelException(AMQConstant.INVALID_ROUTING_KEY, routingKey.toString());
- }
catch (AMQException e)
{
throw body.getChannelException(AMQConstant.CHANNEL_ERROR, e.toString());
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeclareHandler.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeclareHandler.java?rev=904934&r1=904933&r2=904934&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeclareHandler.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeclareHandler.java Sun Jan 31 00:31:49 2010
@@ -21,6 +21,7 @@
package org.apache.qpid.server.handler;
import java.util.UUID;
+import java.util.Collections;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.log4j.Logger;
@@ -150,21 +151,21 @@
{
Exchange defaultExchange = exchangeRegistry.getDefaultExchange();
- queue.bind(defaultExchange, queueName, null);
- _logger.info("Queue " + queueName + " bound to default exchange(" + defaultExchange.getName() + ")");
+ virtualHost.getBindingFactory().addBinding(String.valueOf(queueName), queue, defaultExchange, Collections.EMPTY_MAP);
+ _logger.info("Queue " + queueName + " bound to default exchange(" + defaultExchange.getNameShortString() + ")");
}
}
}
else if (queue.isExclusive() && !queue.isDurable() && queue.getExclusiveOwner() != session)
{
throw body.getConnectionException(AMQConstant.NOT_ALLOWED,
- "Queue " + queue.getName() + " is exclusive, but not created on this Connection.");
+ "Queue " + queue.getNameShortString() + " is exclusive, but not created on this Connection.");
}
else if(!body.getPassive() && ((queue.isExclusive()) != body.getExclusive()))
{
throw body.getChannelException(AMQConstant.ALREADY_EXISTS,
- "Cannot re-declare queue '" + queue.getName() + "' with different exclusivity (was: "
+ "Cannot re-declare queue '" + queue.getNameShortString() + "' with different exclusivity (was: "
+ queue.isExclusive() + " requested " + body.getExclusive() + ")");
}
else if (!body.getPassive() && body.getExclusive() && !queue.getExclusiveOwner().equals(queue.isDurable() ? session.getPrincipal().getName() : session))
@@ -178,13 +179,13 @@
else if(!body.getPassive() && queue.isAutoDelete() != body.getAutoDelete())
{
throw body.getChannelException(AMQConstant.ALREADY_EXISTS,
- "Cannot re-declare queue '" + queue.getName() + "' with different auto-delete (was: "
+ "Cannot re-declare queue '" + queue.getNameShortString() + "' with different auto-delete (was: "
+ queue.isAutoDelete() + " requested " + body.getAutoDelete() + ")");
}
else if(!body.getPassive() && queue.isDurable() != body.getDurable())
{
throw body.getChannelException(AMQConstant.ALREADY_EXISTS,
- "Cannot re-declare queue '" + queue.getName() + "' with different durability (was: "
+ "Cannot re-declare queue '" + queue.getNameShortString() + "' with different durability (was: "
+ queue.isDurable() + " requested " + body.getDurable() + ")");
}
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeleteHandler.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeleteHandler.java?rev=904934&r1=904933&r2=904934&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeleteHandler.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeleteHandler.java Sun Jan 31 00:31:49 2010
@@ -7,9 +7,9 @@
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
- *
+ *
* http://www.apache.org/licenses/LICENSE-2.0
- *
+ *
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
@@ -30,11 +30,9 @@
import org.apache.qpid.server.queue.AMQQueue;
import org.apache.qpid.server.state.AMQStateManager;
import org.apache.qpid.server.state.StateAwareMethodListener;
-import org.apache.qpid.server.store.MessageStore;
import org.apache.qpid.server.store.DurableConfigurationStore;
import org.apache.qpid.server.virtualhost.VirtualHost;
import org.apache.qpid.server.AMQChannel;
-import org.apache.qpid.server.security.access.Permission;
public class QueueDeleteHandler implements StateAwareMethodListener<QueueDeleteBody>
{
@@ -75,7 +73,7 @@
throw body.getChannelNotFoundException(channelId);
}
- //get the default queue on the channel:
+ //get the default queue on the channel:
queue = channel.getDefaultQueue();
}
else
@@ -104,7 +102,7 @@
}
else
{
-
+
//Perform ACLs
if (!virtualHost.getAccessManager().authoriseDelete(session, queue))
{
@@ -113,7 +111,7 @@
else if (queue.isExclusive() && !queue.isDurable() && queue.getExclusiveOwner() != session)
{
throw body.getConnectionException(AMQConstant.NOT_ALLOWED,
- "Queue " + queue.getName() + " is exclusive, but not created on this Connection.");
+ "Queue " + queue.getNameShortString() + " is exclusive, but not created on this Connection.");
}
int purged = queue.delete();
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueueUnbindHandler.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueueUnbindHandler.java?rev=904934&r1=904933&r2=904934&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueueUnbindHandler.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueueUnbindHandler.java Sun Jan 31 00:31:49 2010
@@ -106,23 +106,16 @@
throw body.getConnectionException(AMQConstant.ACCESS_REFUSED, "Permission denied");
}
- try
+ if(virtualHost.getBindingFactory().getBinding(String.valueOf(routingKey), queue, exch, FieldTable.convertToMap(body.getArguments())) == null)
{
- queue.unBind(exch, routingKey, body.getArguments());
+ throw body.getChannelException(AMQConstant.NOT_FOUND,"No such binding");
}
- catch (AMQInvalidRoutingKeyException rke)
+ else
{
- throw body.getChannelException(AMQConstant.INVALID_ROUTING_KEY, routingKey.toString());
- }
- catch (AMQException e)
- {
- if(e.getErrorCode() == AMQConstant.NOT_FOUND)
- {
- throw body.getChannelException(AMQConstant.NOT_FOUND,e.getMessage(),e);
- }
- throw body.getChannelException(AMQConstant.CHANNEL_ERROR, e.toString());
+ virtualHost.getBindingFactory().removeBinding(String.valueOf(routingKey), queue, exch, FieldTable.convertToMap(body.getArguments()));
}
+
if (_log.isInfoEnabled())
{
_log.info("Binding queue " + queue + " to exchange " + exch + " with routing key " + routingKey);
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/subjects/BindingLogSubject.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/subjects/BindingLogSubject.java?rev=904934&r1=904933&r2=904934&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/subjects/BindingLogSubject.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/subjects/BindingLogSubject.java Sun Jan 31 00:31:49 2010
@@ -20,7 +20,6 @@
*/
package org.apache.qpid.server.logging.subjects;
-import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.server.exchange.Exchange;
import org.apache.qpid.server.queue.AMQQueue;
@@ -49,13 +48,13 @@
* @param exchange
* @param queue
*/
- public BindingLogSubject(AMQShortString routingKey, Exchange exchange,
+ public BindingLogSubject(String routingKey, Exchange exchange,
AMQQueue queue)
{
setLogStringWithFormat(BINDING_FORMAT, queue.getVirtualHost().getName(),
- exchange.getType(),
- exchange.getName(),
- queue.getName(),
+ exchange.getTypeShortString(),
+ exchange.getNameShortString(),
+ queue.getNameShortString(),
routingKey);
}
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/subjects/ExchangeLogSubject.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/subjects/ExchangeLogSubject.java?rev=904934&r1=904933&r2=904934&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/subjects/ExchangeLogSubject.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/subjects/ExchangeLogSubject.java Sun Jan 31 00:31:49 2010
@@ -41,6 +41,6 @@
public ExchangeLogSubject(Exchange exchange, VirtualHost vhost)
{
setLogStringWithFormat(BINDING_FORMAT, vhost.getName(),
- exchange.getType(), exchange.getName());
+ exchange.getTypeShortString(), exchange.getNameShortString());
}
}
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/subjects/QueueLogSubject.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/subjects/QueueLogSubject.java?rev=904934&r1=904933&r2=904934&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/subjects/QueueLogSubject.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/subjects/QueueLogSubject.java Sun Jan 31 00:31:49 2010
@@ -40,6 +40,6 @@
{
setLogStringWithFormat(LOG_FORMAT,
queue.getVirtualHost().getName(),
- queue.getName());
+ queue.getNameShortString());
}
}
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/message/AMQMessageHeader.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/message/AMQMessageHeader.java?rev=904934&r1=904933&r2=904934&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/message/AMQMessageHeader.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/message/AMQMessageHeader.java Sun Jan 31 00:31:49 2010
@@ -42,6 +42,10 @@
String getReplyTo();
+ String getReplyToExchange();
+ String getReplyToRoutingKey();
+
+
Object getHeader(String name);
boolean containsHeaders(Set<String> names);
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/message/ContentHeaderBodyAdapter.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/message/ContentHeaderBodyAdapter.java?rev=904934&r1=904933&r2=904934&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/message/ContentHeaderBodyAdapter.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/message/ContentHeaderBodyAdapter.java Sun Jan 31 00:31:49 2010
@@ -85,6 +85,19 @@
return getProperties().getReplyToAsString();
}
+ public String getReplyToExchange()
+ {
+ // TODO
+ return getReplyTo();
+ }
+
+ public String getReplyToRoutingKey()
+ {
+ // TODO
+ return getReplyTo();
+
+ }
+
public Object getHeader(String name)
{
FieldTable ft = getProperties().getHeaders();
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/message/MessageMetaData.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/message/MessageMetaData.java?rev=904934&r1=904933&r2=904934&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/message/MessageMetaData.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/message/MessageMetaData.java Sun Jan 31 00:31:49 2010
@@ -278,6 +278,18 @@
return getProperties().getReplyToAsString();
}
+ public String getReplyToExchange()
+ {
+ // TODO
+ return getReplyTo();
+ }
+
+ public String getReplyToRoutingKey()
+ {
+ // TODO
+ return getReplyTo();
+ }
+
public Object getHeader(String name)
{
FieldTable ft = getProperties().getHeaders();
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/message/MessageTransferHeader.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/message/MessageTransferHeader.java?rev=904934&r1=904933&r2=904934&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/message/MessageTransferHeader.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/message/MessageTransferHeader.java Sun Jan 31 00:31:49 2010
@@ -105,6 +105,30 @@
}
}
+ public String getReplyToExchange()
+ {
+ if (_messageProps != null && _messageProps.getReplyTo() != null)
+ {
+ return _messageProps.getReplyTo().getExchange();
+ }
+ else
+ {
+ return null;
+ }
+ }
+
+ public String getReplyToRoutingKey()
+ {
+ if (_messageProps != null && _messageProps.getReplyTo() != null)
+ {
+ return _messageProps.getReplyTo().getRoutingKey();
+ }
+ else
+ {
+ return null;
+ }
+ }
+
public Object getHeader(String name)
{
Map<String, Object> appHeaders = _messageProps == null ? null : _messageProps.getApplicationHeaders();
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/message/MessageTransferMessage.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/message/MessageTransferMessage.java?rev=904934&r1=904933&r2=904934&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/message/MessageTransferMessage.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/message/MessageTransferMessage.java Sun Jan 31 00:31:49 2010
@@ -119,7 +119,7 @@
public ByteBuffer getBody()
{
ByteBuffer body = getMetaData().getBody();
- if(body == null)
+ if(body == null && getSize() != 0l)
{
final int size = (int) getSize();
int pos = 0;
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org