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 2012/03/01 16:42:53 UTC
svn commit: r1295627 [3/12] - in
/qpid/branches/rg-amqp-1-0-sandbox/qpid/java: ./ bdbstore/ bdbstore/src/main/
bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/
bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/
bdbsto...
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/qmf/QMFMethodRequestCommand.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/qmf/QMFMethodRequestCommand.java?rev=1295627&r1=1295626&r2=1295627&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/qmf/QMFMethodRequestCommand.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/qmf/QMFMethodRequestCommand.java Thu Mar 1 15:42:44 2012
@@ -20,6 +20,7 @@
*/
package org.apache.qpid.qmf;
+import org.apache.log4j.Logger;
import org.apache.qpid.transport.codec.BBDecoder;
import org.apache.qpid.server.virtualhost.VirtualHost;
import org.apache.qpid.server.message.ServerMessage;
@@ -27,11 +28,14 @@ import org.apache.qpid.server.exchange.E
import org.apache.qpid.server.queue.BaseQueue;
import org.apache.qpid.AMQException;
+import java.util.List;
import java.util.UUID;
import java.util.ArrayList;
public class QMFMethodRequestCommand extends QMFCommand
{
+ private static final Logger _qmfLogger = Logger.getLogger("qpid.qmf");
+
private QMFMethodInvocation _methodInstance;
private QMFObject _object;
@@ -58,6 +62,9 @@ public class QMFMethodRequestCommand ext
String queueName = message.getMessageHeader().getReplyToRoutingKey();
QMFCommand[] commands = new QMFCommand[2];
+
+ _qmfLogger.debug("Execute: " + _methodInstance + " on " + _object);
+
commands[0] = _methodInstance.execute(_object, this);
commands[1] = new QMFCommandCompletionCommand(this);
@@ -68,7 +75,7 @@ public class QMFMethodRequestCommand ext
QMFMessage responseMessage = new QMFMessage(queueName, cmd);
- ArrayList<? extends BaseQueue> queues = exchange.route(responseMessage);
+ List<? extends BaseQueue> queues = exchange.route(responseMessage);
for(BaseQueue q : queues)
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/qmf/QMFObject.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/qmf/QMFObject.java?rev=1295627&r1=1295626&r2=1295627&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/qmf/QMFObject.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/qmf/QMFObject.java Thu Mar 1 15:42:44 2012
@@ -73,4 +73,9 @@ public abstract class QMFObject<C extend
abstract public QMFCommand asInstrumentInfoCmd(long sampleTime);
abstract public QMFCommand asGetQueryResponseCmd(final QMFGetQueryCommand queryCommand, long sampleTime);
+ @Override
+ public String toString()
+ {
+ return _delegate.toString();
+ }
}
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/qmf/QMFPackageQueryCommand.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/qmf/QMFPackageQueryCommand.java?rev=1295627&r1=1295626&r2=1295627&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/qmf/QMFPackageQueryCommand.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/qmf/QMFPackageQueryCommand.java Thu Mar 1 15:42:44 2012
@@ -21,6 +21,7 @@
package org.apache.qpid.qmf;
+import org.apache.log4j.Logger;
import org.apache.qpid.transport.codec.BBDecoder;
import org.apache.qpid.server.virtualhost.VirtualHost;
import org.apache.qpid.server.message.ServerMessage;
@@ -31,9 +32,13 @@ import org.apache.qpid.AMQException;
import java.util.ArrayList;
import java.util.Collection;
+import java.util.List;
public class QMFPackageQueryCommand extends QMFCommand
{
+
+ private static final Logger _qmfLogger = Logger.getLogger("qpid.qmf");
+
public QMFPackageQueryCommand(QMFCommandHeader header, BBDecoder decoder)
{
super(header);
@@ -52,6 +57,8 @@ public class QMFPackageQueryCommand exte
QMFCommand[] commands = new QMFCommand[ supportedSchemas.size() + 1 ];
+ _qmfLogger.debug("Exectuting " + this);
+
int i = 0;
for(QMFPackage p : supportedSchemas)
{
@@ -67,7 +74,7 @@ public class QMFPackageQueryCommand exte
Exchange exchange = virtualHost.getExchangeRegistry().getExchange(exchangeName);
- ArrayList<? extends BaseQueue> queues = exchange.route(responseMessage);
+ List<? extends BaseQueue> queues = exchange.route(responseMessage);
for(BaseQueue q : queues)
@@ -83,4 +90,9 @@ public class QMFPackageQueryCommand exte
}
}
}
+
+ public String toString()
+ {
+ return "QMFPackageQueryCommand";
+ }
}
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/qmf/QMFSchemaRequestCommand.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/qmf/QMFSchemaRequestCommand.java?rev=1295627&r1=1295626&r2=1295627&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/qmf/QMFSchemaRequestCommand.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/qmf/QMFSchemaRequestCommand.java Thu Mar 1 15:42:44 2012
@@ -21,6 +21,7 @@
package org.apache.qpid.qmf;
+import org.apache.log4j.Logger;
import org.apache.qpid.transport.codec.BBDecoder;
import org.apache.qpid.server.virtualhost.VirtualHost;
import org.apache.qpid.server.message.ServerMessage;
@@ -31,9 +32,12 @@ import org.apache.qpid.AMQException;
import java.util.Collection;
import java.util.ArrayList;
+import java.util.List;
public class QMFSchemaRequestCommand extends QMFCommand
{
+ private static final Logger _qmfLogger = Logger.getLogger("qpid.qmf");
+
private final String _packageName;
private final String _className;
private final byte[] _hash;
@@ -48,6 +52,8 @@ public class QMFSchemaRequestCommand ext
public void process(VirtualHost virtualHost, ServerMessage message)
{
+ _qmfLogger.debug("Execute: " + this);
+
String exchangeName = message.getMessageHeader().getReplyToExchange();
String routingKey = message.getMessageHeader().getReplyToRoutingKey();
@@ -70,7 +76,7 @@ public class QMFSchemaRequestCommand ext
QMFMessage responseMessage = new QMFMessage(routingKey, cmd);
- ArrayList<? extends BaseQueue> queues = exchange.route(responseMessage);
+ List<? extends BaseQueue> queues = exchange.route(responseMessage);
for(BaseQueue q : queues)
{
@@ -85,4 +91,13 @@ public class QMFSchemaRequestCommand ext
}
}
}
+
+ @Override
+ public String toString()
+ {
+ return "QMFSchemaRequestCommand{" +
+ " packageName='" + _packageName + '\'' +
+ ", className='" + _className + '\'' +
+ '}';
+ }
}
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/qmf/QMFService.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/qmf/QMFService.java?rev=1295627&r1=1295626&r2=1295627&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/qmf/QMFService.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/qmf/QMFService.java Thu Mar 1 15:42:44 2012
@@ -410,7 +410,10 @@ public class QMFService implements Confi
ConcurrentHashMap<UUID, QMFObject> map = _managedObjectsById.get(qmfclass);
if(map != null)
{
- return map.get(id);
+
+ UUID key = new UUID(id.getMostSignificantBits() & (0xFFFl << 48), id.getLeastSignificantBits());
+ return map.get(key);
+
}
else
{
@@ -604,6 +607,11 @@ public class QMFService implements Confi
{
return _obj.getCreateTime();
}
+
+ public String toString()
+ {
+ return _obj.toString();
+ }
}
private class BrokerDelegate implements BrokerSchema.BrokerDelegate
@@ -762,6 +770,11 @@ public class QMFService implements Confi
{
return _obj.getCreateTime();
}
+
+ public String toString()
+ {
+ return _obj.toString();
+ }
}
private class VhostDelegate implements BrokerSchema.VhostDelegate
@@ -797,6 +810,11 @@ public class QMFService implements Confi
{
return _obj.getCreateTime();
}
+
+ public String toString()
+ {
+ return _obj.toString();
+ }
}
private class ExchangeDelegate implements BrokerSchema.ExchangeDelegate
@@ -923,6 +941,11 @@ public class QMFService implements Confi
{
return _obj.getCreateTime();
}
+
+ public String toString()
+ {
+ return _obj.toString();
+ }
}
private class QueueDelegate implements BrokerSchema.QueueDelegate
@@ -1163,6 +1186,11 @@ public class QMFService implements Confi
{
return _obj.getCreateTime();
}
+
+ public String toString()
+ {
+ return _obj.toString();
+ }
}
private class BindingDelegate implements BrokerSchema.BindingDelegate
@@ -1214,6 +1242,11 @@ public class QMFService implements Confi
{
return _obj.getCreateTime();
}
+
+ public String toString()
+ {
+ return _obj.toString();
+ }
}
private class ConnectionDelegate implements BrokerSchema.ConnectionDelegate
@@ -1352,6 +1385,12 @@ public class QMFService implements Confi
// TODO
return 0;
}
+
+
+ public String toString()
+ {
+ return _obj.toString();
+ }
}
private class SessionDelegate implements BrokerSchema.SessionDelegate
@@ -1476,6 +1515,11 @@ public class QMFService implements Confi
{
return _obj.getCreateTime();
}
+
+ public String toString()
+ {
+ return _obj.toString();
+ }
}
private class SubscriptionDelegate implements BrokerSchema.SubscriptionDelegate
@@ -1542,93 +1586,103 @@ public class QMFService implements Confi
{
return _obj.getCreateTime();
}
- }
- private class BridgeDelegate implements BrokerSchema.BridgeDelegate
+ public String toString()
{
- private final BridgeConfig _obj;
+ return _obj.toString();
+ }
+ }
- private BridgeDelegate(final BridgeConfig obj)
- {
- _obj = obj;
- }
+ private class BridgeDelegate implements BrokerSchema.BridgeDelegate
+ {
+ private final BridgeConfig _obj;
- public BrokerSchema.LinkObject getLinkRef()
- {
- return (BrokerSchema.LinkObject) adapt(_obj.getLink());
- }
+ private BridgeDelegate(final BridgeConfig obj)
+ {
+ _obj = obj;
+ }
- public Integer getChannelId()
- {
- return _obj.getChannelId();
- }
+ public BrokerSchema.LinkObject getLinkRef()
+ {
+ return (BrokerSchema.LinkObject) adapt(_obj.getLink());
+ }
- public Boolean getDurable()
- {
- return _obj.isDurable();
- }
+ public Integer getChannelId()
+ {
+ return _obj.getChannelId();
+ }
- public String getSrc()
- {
- return _obj.getSource();
- }
+ public Boolean getDurable()
+ {
+ return _obj.isDurable();
+ }
- public String getDest()
- {
- return _obj.getDestination();
- }
+ public String getSrc()
+ {
+ return _obj.getSource();
+ }
- public String getKey()
- {
- return _obj.getKey();
- }
+ public String getDest()
+ {
+ return _obj.getDestination();
+ }
- public Boolean getSrcIsQueue()
- {
- return _obj.isQueueBridge();
- }
+ public String getKey()
+ {
+ return _obj.getKey();
+ }
- public Boolean getSrcIsLocal()
- {
- return _obj.isLocalSource();
- }
+ public Boolean getSrcIsQueue()
+ {
+ return _obj.isQueueBridge();
+ }
- public String getTag()
- {
- return _obj.getTag();
- }
+ public Boolean getSrcIsLocal()
+ {
+ return _obj.isLocalSource();
+ }
- public String getExcludes()
- {
- return _obj.getExcludes();
- }
+ public String getTag()
+ {
+ return _obj.getTag();
+ }
- public Boolean getDynamic()
- {
- return _obj.isDynamic();
- }
+ public String getExcludes()
+ {
+ return _obj.getExcludes();
+ }
- public Integer getSync()
- {
- return _obj.getAckBatching();
- }
+ public Boolean getDynamic()
+ {
+ return _obj.isDynamic();
+ }
- public BrokerSchema.BridgeClass.CloseMethodResponseCommand close(final BrokerSchema.BridgeClass.CloseMethodResponseCommandFactory factory)
- {
- return null;
- }
+ public Integer getSync()
+ {
+ return _obj.getAckBatching();
+ }
- public UUID getId()
- {
- return _obj.getId();
- }
+ public BrokerSchema.BridgeClass.CloseMethodResponseCommand close(final BrokerSchema.BridgeClass.CloseMethodResponseCommandFactory factory)
+ {
+ return null;
+ }
- public long getCreateTime()
- {
- return _obj.getCreateTime();
- }
+ public UUID getId()
+ {
+ return _obj.getId();
}
+ public long getCreateTime()
+ {
+ return _obj.getCreateTime();
+ }
+
+ public String toString()
+ {
+ return _obj.toString();
+ }
+ }
+
private class LinkDelegate implements BrokerSchema.LinkDelegate
{
private final LinkConfig _obj;
@@ -1665,14 +1719,12 @@ public class QMFService implements Confi
public String getState()
{
- // TODO
- return "";
+ return _obj.getState();
}
public String getLastError()
{
- // TODO
- return "";
+ return _obj.getLastError();
}
public BrokerSchema.LinkClass.CloseMethodResponseCommand close(final BrokerSchema.LinkClass.CloseMethodResponseCommandFactory factory)
@@ -1706,6 +1758,12 @@ public class QMFService implements Confi
{
return _obj.getCreateTime();
}
+
+ @Override
+ public String toString()
+ {
+ return _obj.toString();
+ }
}
}
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java?rev=1295627&r1=1295626&r2=1295627&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java Thu Mar 1 15:42:44 2012
@@ -96,7 +96,7 @@ import java.util.concurrent.atomic.Atomi
public class AMQChannel implements SessionConfig, AMQSessionModel
{
- public static final int DEFAULT_PREFETCH = 5000;
+ public static final int DEFAULT_PREFETCH = 4096;
private static final Logger _logger = Logger.getLogger(AMQChannel.class);
@@ -167,6 +167,8 @@ public class AMQChannel implements Sessi
private final UUID _id;
private long _createTime = System.currentTimeMillis();
+ private final ClientDeliveryMethod _clientDeliveryMethod;
+
public AMQChannel(AMQProtocolSession session, int channelId, MessageStore messageStore)
throws AMQException
{
@@ -184,6 +186,8 @@ public class AMQChannel implements Sessi
// by default the session is non-transactional
_transaction = new AutoCommitTransaction(_messageStore);
+
+ _clientDeliveryMethod = session.createDeliveryMethod(_channelId);
}
public ConfigStore getConfigStore()
@@ -206,6 +210,11 @@ public class AMQChannel implements Sessi
return !(_transaction instanceof AutoCommitTransaction);
}
+ public void receivedComplete()
+ {
+ }
+
+
public boolean inTransaction()
{
return isTransactional() && _txnUpdateTime.get() > 0 && _transaction.getTransactionStartTime() > 0;
@@ -285,7 +294,7 @@ public class AMQChannel implements Sessi
_currentMessage.setExpiration();
- MessageMetaData mmd = _currentMessage.headersReceived();
+ MessageMetaData mmd = _currentMessage.headersReceived(getProtocolSession().getLastReceivedTime());
final StoredMessage<MessageMetaData> handle = _messageStore.addMessage(mmd);
_currentMessage.setStoredMessage(handle);
@@ -317,8 +326,7 @@ public class AMQChannel implements Sessi
{
try
{
- _currentMessage.getStoredMessage().flushToStore();
- final ArrayList<? extends BaseQueue> destinationQueues = _currentMessage.getDestinationQueues();
+ final List<? extends BaseQueue> destinationQueues = _currentMessage.getDestinationQueues();
if(!checkMessageUserId(_currentMessage.getContentHeader()))
{
@@ -340,11 +348,13 @@ public class AMQChannel implements Sessi
}
else
{
- _transaction.enqueue(destinationQueues, _currentMessage, new MessageDeliveryAction(_currentMessage, destinationQueues, isTransactional()));
+ _transaction.enqueue(destinationQueues, _currentMessage, new MessageDeliveryAction(_currentMessage, destinationQueues), getProtocolSession().getLastReceivedTime());
incrementOutstandingTxnsIfNecessary();
updateTransactionalActivity();
}
}
+ _currentMessage.getStoredMessage().flushToStore();
+
}
finally
{
@@ -858,10 +868,8 @@ public class AMQChannel implements Sessi
private Collection<QueueEntry> getAckedMessages(long deliveryTag, boolean multiple)
{
- Map<Long, QueueEntry> ackedMessageMap = new LinkedHashMap<Long,QueueEntry>();
- _unacknowledgedMessageMap.collect(deliveryTag, multiple, ackedMessageMap);
- _unacknowledgedMessageMap.remove(ackedMessageMap);
- return ackedMessageMap.values();
+ return _unacknowledgedMessageMap.acknowledge(deliveryTag, multiple);
+
}
/**
@@ -950,12 +958,17 @@ public class AMQChannel implements Sessi
public void commit() throws AMQException
{
+ commit(null);
+ }
+ public void commit(Runnable immediateAction) throws AMQException
+ {
+
if (!isTransactional())
{
throw new AMQException("Fatal error: commit called on non-transactional channel");
}
- _transaction.commit();
+ _transaction.commit(immediateAction);
_txnCommits.incrementAndGet();
_txnStarts.incrementAndGet();
@@ -1034,7 +1047,7 @@ public class AMQChannel implements Sessi
{
if (isTransactional())
{
- _txnUpdateTime.set(System.currentTimeMillis());
+ _txnUpdateTime.set(getProtocolSession().getLastReceivedTime());
}
}
@@ -1080,21 +1093,6 @@ public class AMQChannel implements Sessi
return _messageStore;
}
- private final ClientDeliveryMethod _clientDeliveryMethod = new ClientDeliveryMethod()
- {
-
- public void deliverToClient(final Subscription sub, final QueueEntry entry, final long deliveryTag)
- throws AMQException
- {
- _session.registerMessageDelivered(entry.getMessage().getSize());
- getProtocolSession().getProtocolOutputConverter().writeDeliver(entry, getChannelId(),
- deliveryTag,
- ((SubscriptionImpl)sub).getConsumerTag());
- entry.incrementDeliveryCount();
- }
-
- };
-
public ClientDeliveryMethod getClientDeliveryMethod()
{
return _clientDeliveryMethod;
@@ -1160,11 +1158,10 @@ public class AMQChannel implements Sessi
private class MessageDeliveryAction implements ServerTransaction.Action
{
private IncomingMessage _incommingMessage;
- private ArrayList<? extends BaseQueue> _destinationQueues;
+ private List<? extends BaseQueue> _destinationQueues;
public MessageDeliveryAction(IncomingMessage currentMessage,
- ArrayList<? extends BaseQueue> destinationQueues,
- boolean transactional)
+ List<? extends BaseQueue> destinationQueues)
{
_incommingMessage = currentMessage;
_destinationQueues = destinationQueues;
@@ -1179,8 +1176,10 @@ public class AMQChannel implements Sessi
final AMQMessage amqMessage = createAMQMessage(_incommingMessage);
MessageReference ref = amqMessage.newReference();
- for(final BaseQueue queue : _destinationQueues)
+ for(int i = 0; i < _destinationQueues.size(); i++)
{
+ BaseQueue queue = _destinationQueues.get(i);
+
BaseQueue.PostEnqueueAction action;
if(immediate)
@@ -1192,7 +1191,7 @@ public class AMQChannel implements Sessi
action = null;
}
- queue.enqueue(amqMessage, action);
+ queue.enqueue(amqMessage, isTransactional(), action);
if(queue instanceof AMQQueue)
{
@@ -1200,6 +1199,8 @@ public class AMQChannel implements Sessi
}
}
+
+ _incommingMessage.getStoredMessage().flushToStore();
ref.release();
}
catch (AMQException e)
@@ -1541,7 +1542,7 @@ public class AMQChannel implements Sessi
final InboundMessage m = new InboundMessageAdapter(rejectedQueueEntry);
- final ArrayList<? extends BaseQueue> destinationQueues = altExchange.route(m);
+ final List<? extends BaseQueue> destinationQueues = altExchange.route(m);
if (destinationQueues == null || destinationQueues.isEmpty())
{
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/ExtractResendAndRequeue.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/ExtractResendAndRequeue.java?rev=1295627&r1=1295626&r2=1295627&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/ExtractResendAndRequeue.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/ExtractResendAndRequeue.java Thu Mar 1 15:42:44 2012
@@ -22,8 +22,8 @@ package org.apache.qpid.server;
import org.apache.qpid.server.ack.UnacknowledgedMessageMap;
import org.apache.qpid.server.queue.QueueEntry;
+import org.apache.qpid.server.store.MessageStore;
import org.apache.qpid.server.subscription.Subscription;
-import org.apache.qpid.server.store.TransactionLog;
import org.apache.qpid.server.txn.ServerTransaction;
import org.apache.qpid.server.txn.AutoCommitTransaction;
import org.apache.qpid.AMQException;
@@ -39,13 +39,13 @@ public class ExtractResendAndRequeue imp
private final Map<Long, QueueEntry> _msgToResend;
private final boolean _requeueIfUnabletoResend;
private final UnacknowledgedMessageMap _unacknowledgedMessageMap;
- private final TransactionLog _transactionLog;
+ private final MessageStore _transactionLog;
public ExtractResendAndRequeue(UnacknowledgedMessageMap unacknowledgedMessageMap,
Map<Long, QueueEntry> msgToRequeue,
Map<Long, QueueEntry> msgToResend,
boolean requeueIfUnabletoResend,
- TransactionLog txnLog)
+ MessageStore txnLog)
{
_unacknowledgedMessageMap = unacknowledgedMessageMap;
_msgToRequeue = msgToRequeue;
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/Main.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/Main.java?rev=1295627&r1=1295626&r2=1295627&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/Main.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/Main.java Thu Mar 1 15:42:44 2012
@@ -27,6 +27,7 @@ import org.apache.commons.cli.OptionBuil
import org.apache.commons.cli.Options;
import org.apache.commons.cli.ParseException;
import org.apache.commons.cli.PosixParser;
+import org.apache.log4j.Logger;
import org.apache.qpid.server.Broker.InitException;
import org.apache.qpid.server.registry.ApplicationRegistry;
@@ -230,11 +231,77 @@ public class Main
{
parsePortArray(options, commandLine.getOptionValues(pe.getExcludeName()), pe);
}
- }
+ }
+
+ setExceptionHandler();
startBroker(options);
}
+ protected void setExceptionHandler()
+ {
+ Thread.UncaughtExceptionHandler handler = null;
+ String handlerClass = System.getProperty("qpid.broker.exceptionHandler");
+ if(handlerClass != null)
+ {
+ try
+ {
+ handler = (Thread.UncaughtExceptionHandler) Class.forName(handlerClass).newInstance();
+ }
+ catch (ClassNotFoundException e)
+ {
+
+ }
+ catch (InstantiationException e)
+ {
+
+ }
+ catch (IllegalAccessException e)
+ {
+
+ }
+ catch (ClassCastException e)
+ {
+
+ }
+ }
+
+ if(handler == null)
+ {
+ handler =
+ new Thread.UncaughtExceptionHandler()
+ {
+ public void uncaughtException(final Thread t, final Throwable e)
+ {
+ try
+ {
+ System.err.println("########################################################################");
+ System.err.println("#");
+ System.err.print("# Unhandled Exception ");
+ System.err.print(e.toString());
+ System.err.print(" in Thread ");
+ System.err.println(t.getName());
+ System.err.println("#");
+ System.err.println("# Exiting");
+ System.err.println("#");
+ System.err.println("########################################################################");
+ e.printStackTrace(System.err);
+
+ Logger logger = Logger.getLogger("org.apache.qpid.server.Main");
+ logger.error("Uncaught exception, shutting down.", e);
+ }
+ finally
+ {
+ Runtime.getRuntime().halt(1);
+ }
+
+ }
+ };
+
+ Thread.setDefaultUncaughtExceptionHandler(handler);
+ }
+ }
+
protected void startBroker(final BrokerOptions options) throws Exception
{
Broker broker = new Broker();
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/ack/UnacknowledgedMessageMap.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/ack/UnacknowledgedMessageMap.java?rev=1295627&r1=1295626&r2=1295627&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/ack/UnacknowledgedMessageMap.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/ack/UnacknowledgedMessageMap.java Thu Mar 1 15:42:44 2012
@@ -46,10 +46,6 @@ public interface UnacknowledgedMessageMa
void add(long deliveryTag, QueueEntry message);
- void collect(long deliveryTag, boolean multiple, Map<Long, QueueEntry> msgs);
-
- void remove(Map<Long,QueueEntry> msgs);
-
QueueEntry remove(long deliveryTag);
Collection<QueueEntry> cancelAllMessages();
@@ -67,6 +63,8 @@ public interface UnacknowledgedMessageMa
*/
Set<Long> getDeliveryTags();
+ Collection<QueueEntry> acknowledge(long deliveryTag, boolean multiple);
+
}
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/ack/UnacknowledgedMessageMapImpl.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/ack/UnacknowledgedMessageMapImpl.java?rev=1295627&r1=1295626&r2=1295627&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/ack/UnacknowledgedMessageMapImpl.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/ack/UnacknowledgedMessageMapImpl.java Thu Mar 1 15:42:44 2012
@@ -157,6 +157,14 @@ public class UnacknowledgedMessageMapImp
}
}
+ public Collection<QueueEntry> acknowledge(long deliveryTag, boolean multiple)
+ {
+ Map<Long, QueueEntry> ackedMessageMap = new LinkedHashMap<Long,QueueEntry>();
+ collect(deliveryTag, multiple, ackedMessageMap);
+ remove(ackedMessageMap);
+ return ackedMessageMap.values();
+ }
+
private void collect(long key, Map<Long, QueueEntry> msgs)
{
synchronized (_lock)
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/binding/Binding.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/binding/Binding.java?rev=1295627&r1=1295626&r2=1295627&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/binding/Binding.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/binding/Binding.java Thu Mar 1 15:42:44 2012
@@ -115,4 +115,9 @@ public class Binding
return result;
}
+ public String toString()
+ {
+ return "Binding{bindingKey="+_bindingKey+", exchange="+_exchange+", queue="+_queue+"}";
+ }
+
}
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/ConfigStore.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/ConfigStore.java?rev=1295627&r1=1295626&r2=1295627&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/ConfigStore.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/ConfigStore.java Thu Mar 1 15:42:44 2012
@@ -40,7 +40,10 @@ public class ConfigStore
private AtomicReference<SystemConfig> _root = new AtomicReference<SystemConfig>(null);
private final AtomicLong _objectIdSource = new AtomicLong(0l);
+ private final AtomicLong _persistentObjectIdSource = new AtomicLong(0l);
+ // TODO - should load/increment this on broker startup
+ private long _sequenceNumber = 1L;
public enum Event
{
@@ -167,9 +170,23 @@ public class ConfigStore
public UUID createId()
{
- return new UUID(0l, _objectIdSource.getAndIncrement());
+ return new UUID(((_sequenceNumber & 0xFFFl)<<48), _objectIdSource.incrementAndGet());
}
+ public UUID createPersistentId()
+ {
+ return new UUID(0L, _persistentObjectIdSource.incrementAndGet());
+ }
+
+ public void persistentIdInUse(UUID id)
+ {
+ long lsb = id.getLeastSignificantBits();
+ long currentId;
+ while((currentId = _persistentObjectIdSource.get()) < lsb)
+ {
+ _persistentObjectIdSource.compareAndSet(currentId, lsb);
+ }
+ }
public SystemConfig getRoot()
{
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/LinkConfig.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/LinkConfig.java?rev=1295627&r1=1295626&r2=1295627&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/LinkConfig.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/LinkConfig.java Thu Mar 1 15:42:44 2012
@@ -54,4 +54,8 @@ public interface LinkConfig extends Conf
String src,
String dest,
String key, String tag, String excludes);
+
+ String getState();
+
+ String getLastError();
}
\ No newline at end of file
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/ServerConfiguration.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/ServerConfiguration.java?rev=1295627&r1=1295626&r2=1295627&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/ServerConfiguration.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/ServerConfiguration.java Thu Mar 1 15:42:44 2012
@@ -89,7 +89,6 @@ public class ServerConfiguration extends
envVarMap.put("QPID_PORT", "connector.port");
envVarMap.put("QPID_ENABLEDIRECTBUFFERS", "advanced.enableDirectBuffers");
envVarMap.put("QPID_SSLPORT", "connector.ssl.port");
- envVarMap.put("QPID_WRITEBIASED", "advanced.useWriteBiasedPool");
envVarMap.put("QPID_JMXPORT_REGISTRYSERVER", MGMT_JMXPORT_REGISTRYSERVER);
envVarMap.put("QPID_JMXPORT_CONNECTORSERVER", MGMT_JMXPORT_CONNECTORSERVER);
envVarMap.put("QPID_FRAMESIZE", "advanced.framesize");
@@ -736,11 +735,6 @@ public class ServerConfiguration extends
return getStringValue("connector.ssl.certType", "SunX509");
}
- public boolean getUseBiasedWrites()
- {
- return getBooleanValue("advanced.useWriteBiasedPool");
- }
-
public String getDefaultVirtualHost()
{
return getStringValue("virtualhosts.default");
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/AbstractExchange.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/AbstractExchange.java?rev=1295627&r1=1295626&r2=1295627&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/AbstractExchange.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/AbstractExchange.java Thu Mar 1 15:42:44 2012
@@ -351,11 +351,11 @@ public abstract class AbstractExchange i
- public final ArrayList<? extends BaseQueue> route(final InboundMessage message)
+ public final List<? extends BaseQueue> route(final InboundMessage message)
{
_receivedMessageCount.incrementAndGet();
_receivedMessageSize.addAndGet(message.getSize());
- final ArrayList<? extends BaseQueue> queues = doRoute(message);
+ final List<? extends BaseQueue> queues = doRoute(message);
if(!queues.isEmpty())
{
_routedMessageCount.incrementAndGet();
@@ -364,7 +364,7 @@ public abstract class AbstractExchange i
return queues;
}
- protected abstract ArrayList<? extends BaseQueue> doRoute(final InboundMessage message);
+ protected abstract List<? extends BaseQueue> doRoute(final InboundMessage message);
public long getMsgReceives()
{
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/DirectExchange.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/DirectExchange.java?rev=1295627&r1=1295626&r2=1295627&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/DirectExchange.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/DirectExchange.java Thu Mar 1 15:42:44 2012
@@ -34,6 +34,8 @@ import org.apache.qpid.server.virtualhos
import javax.management.JMException;
import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArraySet;
@@ -41,8 +43,52 @@ public class DirectExchange extends Abst
{
private static final Logger _logger = Logger.getLogger(DirectExchange.class);
- private final ConcurrentHashMap<String, CopyOnWriteArraySet<Binding>> _bindingsByKey =
- new ConcurrentHashMap<String, CopyOnWriteArraySet<Binding>>();
+ private static final class BindingSet
+ {
+ private CopyOnWriteArraySet<Binding> _bindings = new CopyOnWriteArraySet<Binding>();
+ private List<BaseQueue> _queues = new ArrayList<BaseQueue>();
+
+ public synchronized void addBinding(Binding binding)
+ {
+ _bindings.add(binding);
+ recalculateQueues();
+ }
+
+
+ public synchronized void removeBinding(Binding binding)
+ {
+ _bindings.remove(binding);
+ recalculateQueues();
+ }
+
+ private void recalculateQueues()
+ {
+ List<BaseQueue> queues = new ArrayList<BaseQueue>(_bindings.size());
+
+ for(Binding b : _bindings)
+ {
+ if(!queues.contains(b.getQueue()))
+ {
+ queues.add(b.getQueue());
+ }
+ }
+ _queues = queues;
+ }
+
+
+ public List<BaseQueue> getQueues()
+ {
+ return _queues;
+ }
+
+ public CopyOnWriteArraySet<Binding> getBindings()
+ {
+ return _bindings;
+ }
+ }
+
+ private final ConcurrentHashMap<String, BindingSet> _bindingsByKey =
+ new ConcurrentHashMap<String, BindingSet>();
public static final ExchangeType<DirectExchange> TYPE = new ExchangeType<DirectExchange>()
{
@@ -91,33 +137,20 @@ public class DirectExchange extends Abst
}
- public ArrayList<? extends BaseQueue> doRoute(InboundMessage payload)
+ public List<? extends BaseQueue> doRoute(InboundMessage payload)
{
final String routingKey = payload.getRoutingKey();
- CopyOnWriteArraySet<Binding> bindings = _bindingsByKey.get(routingKey == null ? "" : routingKey);
+ BindingSet bindings = _bindingsByKey.get(routingKey == null ? "" : routingKey);
if(bindings != null)
{
- final ArrayList<BaseQueue> queues = new ArrayList<BaseQueue>(bindings.size());
-
- for(Binding binding : bindings)
- {
- queues.add(binding.getQueue());
- binding.incrementMatches();
- }
-
- if (_logger.isDebugEnabled())
- {
- _logger.debug("Publishing message to queue " + queues);
- }
-
- return queues;
+ return bindings.getQueues();
}
else
{
- return new ArrayList<BaseQueue>(0);
+ return Collections.emptyList();
}
@@ -132,16 +165,10 @@ public class DirectExchange extends Abst
public boolean isBound(AMQShortString routingKey, AMQQueue queue)
{
String bindingKey = (routingKey == null) ? "" : routingKey.toString();
- CopyOnWriteArraySet<Binding> bindings = _bindingsByKey.get(bindingKey);
+ BindingSet bindings = _bindingsByKey.get(bindingKey);
if(bindings != null)
{
- for(Binding binding : bindings)
- {
- if(binding.getQueue().equals(queue))
- {
- return true;
- }
- }
+ return bindings.getQueues().contains(queue);
}
return false;
@@ -150,22 +177,20 @@ public class DirectExchange extends Abst
public boolean isBound(AMQShortString routingKey)
{
String bindingKey = (routingKey == null) ? "" : routingKey.toString();
- CopyOnWriteArraySet<Binding> bindings = _bindingsByKey.get(bindingKey);
- return bindings != null && !bindings.isEmpty();
+ BindingSet bindings = _bindingsByKey.get(bindingKey);
+ return bindings != null && !bindings.getQueues().isEmpty();
}
public boolean isBound(AMQQueue queue)
{
- for (CopyOnWriteArraySet<Binding> bindings : _bindingsByKey.values())
+ for (BindingSet bindings : _bindingsByKey.values())
{
- for(Binding binding : bindings)
+ if(bindings.getQueues().contains(queue))
{
- if(binding.getQueue().equals(queue))
- {
- return true;
- }
+ return true;
}
+
}
return false;
}
@@ -184,19 +209,19 @@ public class DirectExchange extends Abst
assert queue != null;
assert routingKey != null;
- CopyOnWriteArraySet<Binding> bindings = _bindingsByKey.get(bindingKey);
+ BindingSet bindings = _bindingsByKey.get(bindingKey);
if(bindings == null)
{
- bindings = new CopyOnWriteArraySet<Binding>();
- CopyOnWriteArraySet<Binding> newBindings;
+ bindings = new BindingSet();
+ BindingSet newBindings;
if((newBindings = _bindingsByKey.putIfAbsent(bindingKey, bindings)) != null)
{
bindings = newBindings;
}
}
- bindings.add(binding);
+ bindings.addBinding(binding);
}
@@ -204,10 +229,10 @@ public class DirectExchange extends Abst
{
assert binding != null;
- CopyOnWriteArraySet<Binding> bindings = _bindingsByKey.get(binding.getBindingKey());
+ BindingSet bindings = _bindingsByKey.get(binding.getBindingKey());
if(bindings != null)
{
- bindings.remove(binding);
+ bindings.removeBinding(binding);
}
}
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/Exchange.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/Exchange.java?rev=1295627&r1=1295626&r2=1295627&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/Exchange.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/Exchange.java Thu Mar 1 15:42:44 2012
@@ -36,6 +36,7 @@ import org.apache.qpid.server.configurat
import javax.management.JMException;
import java.util.ArrayList;
import java.util.Collection;
+import java.util.List;
public interface Exchange extends ExchangeReferrer, ExchangeConfig
{
@@ -70,7 +71,7 @@ public interface Exchange extends Exchan
*
* @return list of queues to which to route the message.
*/
- ArrayList<? extends BaseQueue> route(InboundMessage message);
+ List<? extends BaseQueue> route(InboundMessage message);
/**
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/TopicExchange.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/TopicExchange.java?rev=1295627&r1=1295626&r2=1295627&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/TopicExchange.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/TopicExchange.java Thu Mar 1 15:42:44 2012
@@ -37,8 +37,11 @@ import org.apache.qpid.server.filter.JMS
import org.apache.qpid.server.message.InboundMessage;
import javax.management.JMException;
+import java.sql.Array;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
+
import java.lang.ref.WeakReference;
public class TopicExchange extends AbstractExchange
@@ -77,8 +80,6 @@ public class TopicExchange extends Abstr
private static final Logger _logger = Logger.getLogger(TopicExchange.class);
-
-
private final TopicParser _parser = new TopicParser();
private final Map<AMQShortString, TopicExchangeResult> _topicExchangeResults =
@@ -175,7 +176,6 @@ public class TopicExchange extends Abstr
_bindings.put(binding, args);
}
-
}
private JMSSelectorFilter createSelectorFilter(final FieldTable args) throws AMQInvalidArgumentException
@@ -201,14 +201,23 @@ public class TopicExchange extends Abstr
public ArrayList<BaseQueue> doRoute(InboundMessage payload)
{
- final AMQShortString routingKey = payload.getRoutingKey() == null
+ final AMQShortString routingKey = payload.getRoutingKeyShortString() == null
? AMQShortString.EMPTY_STRING
- : new AMQShortString(payload.getRoutingKey());
+ : payload.getRoutingKeyShortString();
+
+ final Collection<AMQQueue> matchedQueues = getMatchedQueues(payload, routingKey);
- // The copy here is unfortunate, but not too bad relevant to the amount of
- // things created and copied in getMatchedQueues
- ArrayList<BaseQueue> queues = new ArrayList<BaseQueue>();
- queues.addAll(getMatchedQueues(payload, routingKey));
+ ArrayList<BaseQueue> queues;
+
+ if(matchedQueues.getClass() == ArrayList.class)
+ {
+ queues = (ArrayList) matchedQueues;
+ }
+ else
+ {
+ queues = new ArrayList<BaseQueue>();
+ queues.addAll(matchedQueues);
+ }
if(queues == null || queues.isEmpty())
{
@@ -325,25 +334,28 @@ public class TopicExchange extends Abstr
{
Collection<TopicMatcherResult> results = _parser.parse(routingKey);
- if(results.isEmpty())
- {
- return Collections.EMPTY_SET;
- }
- else
+ switch(results.size())
{
- Collection<AMQQueue> queues = results.size() == 1 ? null : new HashSet<AMQQueue>();
- for(TopicMatcherResult result : results)
- {
- TopicExchangeResult res = (TopicExchangeResult)result;
+ case 0:
+ return Collections.EMPTY_SET;
+ case 1:
+ TopicMatcherResult[] resultQueues = new TopicMatcherResult[1];
+ results.toArray(resultQueues);
+ return ((TopicExchangeResult)resultQueues[0]).processMessage(message, null);
+ default:
+ Collection<AMQQueue> queues = new HashSet<AMQQueue>();
+ for(TopicMatcherResult result : results)
+ {
+ TopicExchangeResult res = (TopicExchangeResult)result;
+
+ for(Binding b : res.getBindings())
+ {
+ b.incrementMatches();
+ }
- for(Binding b : res.getBindings())
- {
- b.incrementMatches();
+ queues = res.processMessage(message, queues);
}
-
- queues = res.processMessage(message, queues);
- }
- return queues;
+ return queues;
}
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicExchangeResult.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicExchangeResult.java?rev=1295627&r1=1295626&r2=1295627&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicExchangeResult.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicExchangeResult.java Thu Mar 1 15:42:44 2012
@@ -39,6 +39,7 @@ public final class TopicExchangeResult i
private final List<Binding> _bindings = new CopyOnWriteArrayList<Binding>();
private final Map<AMQQueue, Integer> _unfilteredQueues = new ConcurrentHashMap<AMQQueue, Integer>();
private final ConcurrentHashMap<AMQQueue, Map<MessageFilter,Integer>> _filteredQueues = new ConcurrentHashMap<AMQQueue, Map<MessageFilter, Integer>>();
+ private volatile ArrayList<AMQQueue> _unfilteredQueueList = new ArrayList<AMQQueue>(0);
public void addUnfilteredQueue(AMQQueue queue)
{
@@ -46,6 +47,9 @@ public final class TopicExchangeResult i
if(instances == null)
{
_unfilteredQueues.put(queue, 1);
+ ArrayList<AMQQueue> newList = new ArrayList<AMQQueue>(_unfilteredQueueList);
+ newList.add(queue);
+ _unfilteredQueueList = newList;
}
else
{
@@ -59,6 +63,10 @@ public final class TopicExchangeResult i
if(instances == 1)
{
_unfilteredQueues.remove(queue);
+ ArrayList<AMQQueue> newList = new ArrayList<AMQQueue>(_unfilteredQueueList);
+ newList.remove(queue);
+ _unfilteredQueueList = newList;
+
}
else
{
@@ -166,7 +174,7 @@ public final class TopicExchangeResult i
{
if(_filteredQueues.isEmpty())
{
- return new ArrayList<AMQQueue>(_unfilteredQueues.keySet());
+ return _unfilteredQueueList;
}
else
{
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicMatcherDFAState.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicMatcherDFAState.java?rev=1295627&r1=1295626&r2=1295627&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicMatcherDFAState.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicMatcherDFAState.java Thu Mar 1 15:42:44 2012
@@ -77,7 +77,7 @@ public class TopicMatcherDFAState
}
if(nextState == null)
{
- return Collections.EMPTY_SET;
+ return Collections.EMPTY_LIST;
}
// Shortcut if we are at a looping terminal state
if((nextState == this) && (_nextStateMap.size() == 1) && _nextStateMap.containsKey(TopicWord.ANY_WORD))
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/federation/Bridge.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/federation/Bridge.java?rev=1295627&r1=1295626&r2=1295627&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/federation/Bridge.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/federation/Bridge.java Thu Mar 1 15:42:44 2012
@@ -21,6 +21,7 @@
package org.apache.qpid.server.federation;
import org.apache.qpid.AMQException;
+import org.apache.qpid.AMQStoreException;
import org.apache.qpid.server.binding.Binding;
import org.apache.qpid.server.configuration.BridgeConfig;
import org.apache.qpid.server.configuration.BridgeConfigType;
@@ -44,32 +45,23 @@ import org.apache.qpid.server.transport.
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 org.apache.qpid.transport.*;
+
+import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
public class Bridge implements BridgeConfig
{
+ private static final String DURABLE = "durable";
+ private static final String DYNAMIC = "dynamic";
+ private static final String SRC_IS_QUEUE = "srcIsQueue";
+ private static final String SRC_IS_LOCAL = "srcIsLocal";
+ private static final String SOURCE = "source";
+ private static final String DESTINATION = "destination";
+ private static final String KEY = "key";
+ private static final String TAG = "tag";
+ private static final String EXCLUDES = "excludes";
private final boolean _durable;
private final boolean _dynamic;
private final boolean _queueBridge;
@@ -113,19 +105,36 @@ public class Bridge implements BridgeCon
_key = key;
_tag = tag;
_excludes = excludes;
- _id = brokerLink.getConfigStore().createId();
+ _id = durable ? brokerLink.getConfigStore().createPersistentId() : brokerLink.getConfigStore().createId();
_transaction = new AutoCommitTransaction(getVirtualHost().getMessageStore());
- if(dynamic)
+ if(durable)
{
- if(srcIsLocal)
+ try
+ {
+ brokerLink.getVirtualHost().getDurableConfigurationStore().createBridge(this);
+ }
+ catch (AMQStoreException e)
+ {
+ throw new RuntimeException(e);
+ }
+ }
+
+ createDelegate();
+ }
+
+ private void createDelegate()
+ {
+ if(_dynamic)
+ {
+ if(_localSource)
{
// TODO
}
else
{
- if(srcIsQueue)
+ if(_queueBridge)
{
// TODO
}
@@ -137,9 +146,9 @@ public class Bridge implements BridgeCon
}
else
{
- if(srcIsLocal)
+ if(_localSource)
{
- if(srcIsQueue)
+ if(_queueBridge)
{
_delegate = new StaticQueuePushBridge();
}
@@ -150,7 +159,7 @@ public class Bridge implements BridgeCon
}
else
{
- if(srcIsQueue)
+ if(_queueBridge)
{
_delegate = new StaticQueuePullBridge();
}
@@ -162,6 +171,65 @@ public class Bridge implements BridgeCon
}
}
+ public Bridge(final BrokerLink brokerLink,
+ final int bridgeNo,
+ final UUID id,
+ final long createTime,
+ final Map<String, String> arguments)
+ {
+ _link = brokerLink;
+ _bridgeNo = bridgeNo;
+ _id = id;
+ brokerLink.getConfigStore().persistentIdInUse(id);
+ _createTime = createTime;
+
+ _durable = Boolean.valueOf(arguments.get(DURABLE));
+ _dynamic = Boolean.valueOf(arguments.get(DYNAMIC));
+ _queueBridge = Boolean.valueOf(arguments.get(SRC_IS_QUEUE));
+ _localSource = Boolean.valueOf(arguments.get(SRC_IS_LOCAL));
+ _source = arguments.get(SOURCE);
+ _destination = arguments.get(DESTINATION);
+ _key = arguments.get(KEY);
+ _tag = arguments.get(TAG);
+ _excludes = arguments.get(EXCLUDES);
+
+ //TODO.
+ _transaction = new AutoCommitTransaction(getVirtualHost().getMessageStore());
+
+
+ if(_durable)
+ {
+ try
+ {
+ brokerLink.getVirtualHost().getDurableConfigurationStore().createBridge(this);
+ }
+ catch (AMQStoreException e)
+ {
+ throw new RuntimeException(e);
+ }
+ }
+
+ createDelegate();
+ }
+
+
+ public Map<String,String> getArguments()
+ {
+ Map<String,String> arguments = new HashMap<String, String>();
+
+ arguments.put(DURABLE, String.valueOf(_durable));
+ arguments.put(DYNAMIC, String.valueOf(_dynamic));
+ arguments.put(SRC_IS_QUEUE, String.valueOf(_queueBridge));
+ arguments.put(SRC_IS_LOCAL, String.valueOf(_localSource));
+ arguments.put(SOURCE, _source);
+ arguments.put(DESTINATION, _destination);
+ arguments.put(KEY, _key);
+ arguments.put(TAG, _tag);
+ arguments.put(EXCLUDES, _excludes);
+
+ return Collections.unmodifiableMap(arguments);
+ }
+
public UUID getId()
{
return _id;
@@ -336,6 +404,7 @@ public class Bridge implements BridgeCon
}
+
private interface BridgeImpl
{
void setSession(Session session);
@@ -365,7 +434,8 @@ public class Bridge implements BridgeCon
// TODO - deal with exchange not existing
DeliveryProperties delvProps = null;
- if(xfr.getHeader() != null && (delvProps = xfr.getHeader().get(DeliveryProperties.class)) != null && delvProps.hasTtl() && !delvProps.hasExpiration())
+ if(xfr.getHeader() != null && (delvProps = xfr.getHeader().getDeliveryProperties()) != null && delvProps.hasTtl() &&
+ !delvProps.hasExpiration())
{
delvProps.setExpiration(System.currentTimeMillis() + delvProps.getTtl());
}
@@ -377,7 +447,7 @@ public class Bridge implements BridgeCon
storeMessage.flushToStore();
MessageTransferMessage message = new MessageTransferMessage(storeMessage, ((ServerSession)_session).getReference());
- ArrayList<? extends BaseQueue> queues = exchange.route(message);
+ List<? extends BaseQueue> queues = exchange.route(message);
@@ -391,7 +461,7 @@ public class Bridge implements BridgeCon
{
if(xfr.getAcceptMode() == MessageAcceptMode.EXPLICIT)
{
- RangeSet rejects = new RangeSet();
+ RangeSet rejects = RangeSetFactory.createRangeSet();
rejects.add(xfr.getId());
MessageReject reject = new MessageReject(rejects, MessageRejectCode.UNROUTABLE, "Unroutable");
ssn.invoke(reject);
@@ -428,7 +498,7 @@ public class Bridge implements BridgeCon
}
- private void enqueue(final ServerMessage message, final ArrayList<? extends BaseQueue> queues)
+ private void enqueue(final ServerMessage message, final List<? extends BaseQueue> queues)
{
_transaction.enqueue(queues,message, new ServerTransaction.Action()
{
@@ -456,8 +526,7 @@ public class Bridge implements BridgeCon
{
// NO-OP
}
- });
-
+ }, 0L);
}
public void exception(final Session session, final SessionException exception)
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/federation/BrokerLink.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/federation/BrokerLink.java?rev=1295627&r1=1295626&r2=1295627&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/federation/BrokerLink.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/federation/BrokerLink.java Thu Mar 1 15:42:44 2012
@@ -20,6 +20,7 @@
*/
package org.apache.qpid.server.federation;
+import org.apache.qpid.AMQStoreException;
import org.apache.qpid.common.ServerPropertyNames;
import org.apache.qpid.server.configuration.ConfigStore;
import org.apache.qpid.server.configuration.ConfiguredObject;
@@ -29,16 +30,19 @@ import org.apache.qpid.server.configurat
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 org.apache.qpid.transport.*;
+import org.apache.qpid.util.Strings;
-import java.util.Map;
-import java.util.UUID;
+import javax.security.auth.callback.Callback;
+import javax.security.auth.callback.CallbackHandler;
+import javax.security.auth.callback.NameCallback;
+import javax.security.auth.callback.PasswordCallback;
+import javax.security.auth.callback.UnsupportedCallbackException;
+import javax.security.sasl.Sasl;
+import javax.security.sasl.SaslClient;
+import javax.security.sasl.SaslException;
+import java.io.IOException;
+import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ScheduledThreadPoolExecutor;
@@ -55,6 +59,14 @@ public class BrokerLink implements LinkC
private static final ScheduledThreadPoolExecutor _threadPool =
new ScheduledThreadPoolExecutor(CORE_POOL_SIZE);
+ private static final String TRANSPORT = "transport";
+ private static final String HOST = "host";
+ private static final String PORT = "port";
+ private static final String REMOTE_VHOST = "remoteVhost";
+ private static final String DURABLE = "durable";
+ private static final String AUTH_MECHANISM = "authMechanism";
+ private static final String USERNAME = "username";
+ private static final String PASSWORD = "password";
private final String _transport;
@@ -68,7 +80,7 @@ public class BrokerLink implements LinkC
private final VirtualHost _virtualHost;
private UUID _id;
private AtomicBoolean _closing = new AtomicBoolean();
- private final long _createTime = System.currentTimeMillis();
+ private final long _createTime;
private Connection _qpidConnection;
private AtomicReference<Thread> _executor = new AtomicReference<Thread>();
private AtomicInteger _bridgeId = new AtomicInteger();
@@ -88,8 +100,10 @@ public class BrokerLink implements LinkC
{
doMakeConnection();
}
- };;
- ;
+ };
+
+
+
public static enum State
{
@@ -205,6 +219,44 @@ public class BrokerLink implements LinkC
}
};
+ public BrokerLink(final VirtualHost virtualHost, UUID id, long createTime, Map<String, String> arguments)
+ {
+ _virtualHost = virtualHost;
+ _id = id;
+ virtualHost.getConfigStore().persistentIdInUse(id);
+ _createTime = createTime;
+ _transport = arguments.get(TRANSPORT);
+
+ _host = arguments.get(HOST);
+ _port = Integer.parseInt(arguments.get(PORT));
+ _remoteVhost = arguments.get(REMOTE_VHOST);
+ _durable = Boolean.parseBoolean(arguments.get(DURABLE));
+ _authMechanism = arguments.get("authMechanism");
+ _username = arguments.get("username");
+ _password = arguments.get("password");
+
+ if(_durable)
+ {
+ try
+ {
+ _virtualHost.getDurableConfigurationStore().createBrokerLink(this);
+ }
+ catch (AMQStoreException e)
+ {
+ throw new RuntimeException(e);
+ }
+ }
+
+
+ _qpidConnection = new Connection();
+ _connectionConfig = new ConnectionConfigAdapter();
+ _qpidConnection.addConnectionListener(this);
+
+
+ makeConnection();
+
+ }
+
public BrokerLink(final VirtualHost virtualHost,
final String transport,
@@ -212,10 +264,13 @@ public class BrokerLink implements LinkC
final int port,
final String remoteVhost,
final boolean durable,
- final String authMechanism, final String username, final String password)
+ final String authMechanism,
+ final String username,
+ final String password)
{
_virtualHost = virtualHost;
_transport = transport;
+ _createTime = System.currentTimeMillis();
_host = host;
_port = port;
_remoteVhost = remoteVhost;
@@ -223,15 +278,42 @@ public class BrokerLink implements LinkC
_authMechanism = authMechanism;
_username = username;
_password = password;
- _id = virtualHost.getConfigStore().createId();
+ _id = durable ? virtualHost.getConfigStore().createPersistentId() : virtualHost.getConfigStore().createId();
+
+ if(durable)
+ {
+ try
+ {
+ _virtualHost.getDurableConfigurationStore().createBrokerLink(this);
+ }
+ catch (AMQStoreException e)
+ {
+ throw new RuntimeException(e);
+ }
+ }
_qpidConnection = new Connection();
_connectionConfig = new ConnectionConfigAdapter();
_qpidConnection.addConnectionListener(this);
-
makeConnection();
}
+ public Map<String,String> getArguments()
+ {
+ Map<String,String> arguments = new HashMap<String, String>();
+
+ arguments.put(TRANSPORT, _transport);
+ arguments.put(HOST, _host);
+ arguments.put(PORT, String.valueOf(_port));
+ arguments.put(REMOTE_VHOST, _remoteVhost);
+ arguments.put(DURABLE, String.valueOf(_durable));
+ arguments.put(AUTH_MECHANISM, _authMechanism);
+ arguments.put(USERNAME, _username);
+ arguments.put(PASSWORD, _password);
+
+ return Collections.unmodifiableMap(arguments);
+ }
+
private final boolean updateState(State expected, State newState)
{
return _stateUpdater.compareAndSet(this,expected,newState);
@@ -250,9 +332,50 @@ public class BrokerLink implements LinkC
{
try
{
+ _qpidConnection.setConnectionDelegate(new ClientDelegate(new ConnectionSettings())
+ {
+ protected SaslClient createSaslClient(List<Object> brokerMechs) throws ConnectionException,
+ SaslException
+ {
+ Map<String,Object> saslProps = new HashMap<String,Object>();
+
+
+ CallbackHandler cbh = new CallbackHandler()
+ {
+ public void handle(final Callback[] callbacks)
+ throws IOException, UnsupportedCallbackException
+ {
+ for (int i = 0; i < callbacks.length; i++)
+ {
+ Callback cb = callbacks[i];
+ if (cb instanceof NameCallback)
+ {
+ ((NameCallback)cb).setName(_username);
+ }
+ else if (cb instanceof PasswordCallback)
+ {
+ ((PasswordCallback)cb).setPassword(_password.toCharArray());
+ }
+ else
+ {
+ throw new UnsupportedCallbackException(cb);
+ }
+ }
+
+ }
+ };
+ final SaslClient sc = Sasl.createSaslClient(new String[] {"PLAIN"}, null,
+ _conSettings.getSaslProtocol(),
+ _conSettings.getSaslServerName(),
+ saslProps, cbh);
+
+ return sc;
+ }});
+
_qpidConnection.connect(_host, _port, _remoteVhost, _username, _password, "ssl".equals(_transport), _authMechanism);
final Map<String,Object> serverProps = _qpidConnection.getServerProperties();
+
_remoteFederationTag = (String) serverProps.get(ServerPropertyNames.FEDERATION_TAG);
if(_remoteFederationTag == null)
{
@@ -445,6 +568,20 @@ public class BrokerLink implements LinkC
}
+ public void createBridge(final UUID id, final long createTime, final Map<String, String> arguments)
+ {
+ if(!_closing.get())
+ {
+ Bridge bridge = new Bridge(this, _bridgeId.incrementAndGet(), id, createTime, arguments);
+ if(_bridges.putIfAbsent(bridge, bridge) == null)
+ {
+
+ addBridge(bridge);
+ }
+ }
+ }
+
+
private void addBridge(final Bridge bridge)
{
getConfigStore().addConfiguredObject(bridge);
@@ -509,4 +646,34 @@ public class BrokerLink implements LinkC
{
return _remoteFederationTag;
}
+
+ public String getState()
+ {
+ return _state.name();
+ }
+
+ public String getLastError()
+ {
+ return _lastErrorMessage;
+ }
+
+ @Override
+ public String toString()
+ {
+ return "BrokerLink{" +
+ " _id=" + _id +
+ ", _transport='" + _transport + '\'' +
+ ", _host='" + _host + '\'' +
+ ", _port=" + _port +
+ ", _remoteVhost='" + _remoteVhost + '\'' +
+ ", _durable=" + _durable +
+ ", _authMechanism='" + _authMechanism + '\'' +
+ ", _username='" + _username + '\'' +
+ ", _password='" + _password + '\'' +
+ ", _virtualHost=" + _virtualHost +
+ ", _createTime=" + _createTime +
+ ", _remoteFederationTag='" + _remoteFederationTag + '\'' +
+ ", _state=" + _state +
+ '}';
+ }
}
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/flow/AbstractFlowCreditManager.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/flow/AbstractFlowCreditManager.java?rev=1295627&r1=1295626&r2=1295627&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/flow/AbstractFlowCreditManager.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/flow/AbstractFlowCreditManager.java Thu Mar 1 15:42:44 2012
@@ -20,6 +20,7 @@
*/
package org.apache.qpid.server.flow;
+import java.util.ArrayList;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.Set;
import java.util.HashSet;
@@ -27,13 +28,16 @@ import java.util.HashSet;
public abstract class AbstractFlowCreditManager implements FlowCreditManager
{
protected final AtomicBoolean _suspended = new AtomicBoolean(false);
- private final Set<FlowCreditManagerListener> _listeners = new HashSet<FlowCreditManagerListener>();
+ private final ArrayList<FlowCreditManagerListener> _listeners = new ArrayList<FlowCreditManagerListener>();
public final void addStateListener(FlowCreditManagerListener listener)
{
synchronized(_listeners)
{
- _listeners.add(listener);
+ if(!_listeners.contains(listener))
+ {
+ _listeners.add(listener);
+ }
}
}
@@ -49,9 +53,10 @@ public abstract class AbstractFlowCredit
{
synchronized(_listeners)
{
- for(FlowCreditManagerListener listener : _listeners)
+ final int size = _listeners.size();
+ for(int i = 0; i<size; i++)
{
- listener.creditStateChanged(!suspended);
+ _listeners.get(i).creditStateChanged(!suspended);
}
}
}
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicConsumeMethodHandler.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicConsumeMethodHandler.java?rev=1295627&r1=1295626&r2=1295627&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicConsumeMethodHandler.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicConsumeMethodHandler.java Thu Mar 1 15:42:44 2012
@@ -52,7 +52,6 @@ public class BasicConsumeMethodHandler i
AMQProtocolSession protocolConnection = stateManager.getProtocolSession();
AMQChannel channel = protocolConnection.getChannel(channelId);
-
VirtualHost vHost = protocolConnection.getVirtualHost();
if (channel == null)
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ChannelCloseHandler.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ChannelCloseHandler.java?rev=1295627&r1=1295626&r2=1295627&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ChannelCloseHandler.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ChannelCloseHandler.java Thu Mar 1 15:42:44 2012
@@ -65,7 +65,6 @@ public class ChannelCloseHandler impleme
{
throw body.getConnectionException(AMQConstant.CHANNEL_ERROR, "Trying to close unknown channel");
}
-
session.closeChannel(channelId);
// Client requested closure so we don't wait for ok we send it
stateManager.getProtocolSession().closeChannelOk(channelId);
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ChannelFlowHandler.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ChannelFlowHandler.java?rev=1295627&r1=1295626&r2=1295627&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ChannelFlowHandler.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ChannelFlowHandler.java Thu Mar 1 15:42:44 2012
@@ -55,7 +55,6 @@ public class ChannelFlowHandler implemen
{
throw body.getChannelNotFoundException(channelId);
}
-
channel.setSuspended(!body.getActive());
_logger.debug("Channel.Flow for channel " + channelId + ", active=" + body.getActive());
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionCloseMethodHandler.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionCloseMethodHandler.java?rev=1295627&r1=1295626&r2=1295627&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionCloseMethodHandler.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionCloseMethodHandler.java Thu Mar 1 15:42:44 2012
@@ -49,7 +49,6 @@ public class ConnectionCloseMethodHandle
public void methodReceived(AMQStateManager stateManager, ConnectionCloseBody body, int channelId) throws AMQException
{
AMQProtocolSession session = stateManager.getProtocolSession();
-
if (_logger.isInfoEnabled())
{
_logger.info("ConnectionClose received with reply code/reply text " + body.getReplyCode() + "/" +
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ExchangeBoundHandler.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ExchangeBoundHandler.java?rev=1295627&r1=1295626&r2=1295627&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ExchangeBoundHandler.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ExchangeBoundHandler.java Thu Mar 1 15:42:44 2012
@@ -65,7 +65,6 @@ public class ExchangeBoundHandler implem
public void methodReceived(AMQStateManager stateManager, ExchangeBoundBody body, int channelId) throws AMQException
{
AMQProtocolSession session = stateManager.getProtocolSession();
-
VirtualHost virtualHost = session.getVirtualHost();
QueueRegistry queueRegistry = virtualHost.getQueueRegistry();
MethodRegistry methodRegistry = session.getMethodRegistry();
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeclareHandler.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeclareHandler.java?rev=1295627&r1=1295626&r2=1295627&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeclareHandler.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeclareHandler.java Thu Mar 1 15:42:44 2012
@@ -85,6 +85,13 @@ public class QueueDeclareHandler impleme
//TODO: do we need to check that the queue already exists with exactly the same "configuration"?
+ AMQChannel channel = protocolConnection.getChannel(channelId);
+
+ if (channel == null)
+ {
+ throw body.getChannelNotFoundException(channelId);
+ }
+
synchronized (queueRegistry)
{
queue = queueRegistry.getQueue(queueName);
@@ -183,12 +190,6 @@ public class QueueDeclareHandler impleme
}
- AMQChannel channel = protocolConnection.getChannel(channelId);
-
- if (channel == null)
- {
- throw body.getChannelNotFoundException(channelId);
- }
//set this as the default queue on the channel:
channel.setDefaultQueue(queue);
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeleteHandler.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeleteHandler.java?rev=1295627&r1=1295626&r2=1295627&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeleteHandler.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeleteHandler.java Thu Mar 1 15:42:44 2012
@@ -64,15 +64,17 @@ public class QueueDeleteHandler implemen
QueueRegistry queueRegistry = virtualHost.getQueueRegistry();
DurableConfigurationStore store = virtualHost.getDurableConfigurationStore();
+
+ AMQChannel channel = protocolConnection.getChannel(channelId);
+
+ if (channel == null)
+ {
+ throw body.getChannelNotFoundException(channelId);
+ }
+
AMQQueue queue;
if (body.getQueue() == null)
{
- AMQChannel channel = protocolConnection.getChannel(channelId);
-
- if (channel == null)
- {
- throw body.getChannelNotFoundException(channelId);
- }
//get the default queue on the channel:
queue = channel.getDefaultQueue();
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueuePurgeHandler.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueuePurgeHandler.java?rev=1295627&r1=1295626&r2=1295627&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueuePurgeHandler.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueuePurgeHandler.java Thu Mar 1 15:42:44 2012
@@ -63,17 +63,14 @@ public class QueuePurgeHandler implement
QueueRegistry queueRegistry = virtualHost.getQueueRegistry();
AMQChannel channel = protocolConnection.getChannel(channelId);
-
-
+ if (channel == null)
+ {
+ throw body.getChannelNotFoundException(channelId);
+ }
AMQQueue queue;
if(body.getQueue() == null)
{
- if (channel == null)
- {
- throw body.getChannelNotFoundException(channelId);
- }
-
//get the default queue on the channel:
queue = channel.getDefaultQueue();
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueueUnbindHandler.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueueUnbindHandler.java?rev=1295627&r1=1295626&r2=1295627&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueueUnbindHandler.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueueUnbindHandler.java Thu Mar 1 15:42:44 2012
@@ -66,14 +66,15 @@ public class QueueUnbindHandler implemen
final AMQQueue queue;
final AMQShortString routingKey;
- if (body.getQueue() == null)
+
+ AMQChannel channel = session.getChannel(channelId);
+ if (channel == null)
{
- AMQChannel channel = session.getChannel(channelId);
+ throw body.getChannelNotFoundException(channelId);
+ }
- if (channel == null)
- {
- throw body.getChannelNotFoundException(channelId);
- }
+ if (body.getQueue() == null)
+ {
queue = channel.getDefaultQueue();
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org