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/10 20:22:37 UTC
svn commit: r1299257 [3/26] - in
/qpid/branches/rg-amqp-1-0-sandbox/qpid/java: broker-plugins/
broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/config/
broker-plugins/access-control/src/main/java/org/apache/qpid/server...
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=1299257&r1=1299256&r2=1299257&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 Sat Mar 10 19:22:10 2012
@@ -22,16 +22,15 @@
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;
-import org.apache.qpid.server.registry.IApplicationRegistry;
+
+import org.apache.qpid.AMQException;
import org.apache.qpid.server.exchange.Exchange;
+import org.apache.qpid.server.message.ServerMessage;
import org.apache.qpid.server.queue.BaseQueue;
-import org.apache.qpid.AMQException;
+import org.apache.qpid.server.registry.IApplicationRegistry;
+import org.apache.qpid.server.virtualhost.VirtualHost;
+import org.apache.qpid.transport.codec.BBDecoder;
-import java.util.Collection;
-import java.util.ArrayList;
import java.util.List;
public class QMFSchemaRequestCommand extends QMFCommand
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/qmf/QMFSchemaResponseCommand.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/qmf/QMFSchemaResponseCommand.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/qmf/QMFSchemaResponseCommand.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/qmf/QMFSchemaResponseCommand.java Sat Mar 10 19:22:10 2012
@@ -21,8 +21,6 @@
package org.apache.qpid.qmf;
-import org.apache.qpid.server.virtualhost.VirtualHost;
-import org.apache.qpid.server.message.ServerMessage;
import org.apache.qpid.transport.codec.BBEncoder;
import java.util.Collection;
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=1299257&r1=1299256&r2=1299257&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 Sat Mar 10 19:22:10 2012
@@ -658,6 +658,11 @@ public class QMFService implements Confi
return _obj.getStagingThreshold();
}
+ public Boolean getMgmtPublish()
+ {
+ return true;
+ }
+
public Integer getMgmtPubInterval()
{
return _obj.getManagementPublishInterval();
@@ -678,6 +683,204 @@ public class QMFService implements Confi
return (System.currentTimeMillis() - _obj.getCreateTime()) * 1000000L;
}
+ public Long getQueueCount()
+ {
+ // TODO
+ return 0L;
+ }
+
+ public Long getMsgTotalEnqueues()
+ {
+ // TODO
+ return 0L;
+ }
+
+ public Long getMsgTotalDequeues()
+ {
+ // TODO
+ return 0L;
+ }
+
+ public Long getByteTotalEnqueues()
+ {
+ // TODO
+ return 0L;
+ }
+
+ public Long getByteTotalDequeues()
+ {
+ // TODO
+ return 0L;
+ }
+
+ public Long getMsgDepth()
+ {
+ // TODO
+ return 0L;
+ }
+
+ public Long getByteDepth()
+ {
+ // TODO
+ return 0L;
+ }
+
+ public Long getMsgPersistEnqueues()
+ {
+ // TODO
+ return 0L;
+ }
+
+ public Long getMsgPersistDequeues()
+ {
+ // TODO
+ return 0L;
+ }
+
+ public Long getBytePersistEnqueues()
+ {
+ // TODO
+ return 0L;
+ }
+
+ public Long getBytePersistDequeues()
+ {
+ // TODO
+ return 0L;
+ }
+
+ public Long getMsgTxnEnqueues()
+ {
+ // TODO
+ return 0L;
+ }
+
+ public Long getMsgTxnDequeues()
+ {
+ // TODO
+ return 0L;
+ }
+
+ public Long getByteTxnEnqueues()
+ {
+ // TODO
+ return 0L;
+ }
+
+ public Long getByteTxnDequeues()
+ {
+ // TODO
+ return 0L;
+ }
+
+ public Long getMsgFtdEnqueues()
+ {
+ // TODO
+ return 0L;
+ }
+
+ public Long getMsgFtdDequeues()
+ {
+ // TODO
+ return 0L;
+ }
+
+ public Long getByteFtdEnqueues()
+ {
+ // TODO
+ return 0L;
+ }
+
+ public Long getByteFtdDequeues()
+ {
+ // TODO
+ return 0L;
+ }
+
+ public Long getMsgFtdDepth()
+ {
+ // TODO
+ return 0L;
+ }
+
+ public Long getByteFtdDepth()
+ {
+ // TODO
+ return 0L;
+ }
+
+ public Long getReleases()
+ {
+ // TODO
+ return 0L;
+ }
+
+ public Long getAcquires()
+ {
+ // TODO
+ return 0L;
+ }
+
+ public Long getDiscardsNoRoute()
+ {
+ // TODO
+ return 0L;
+ }
+
+ public Long getDiscardsTtl()
+ {
+ // TODO
+ return 0L;
+ }
+
+ public Long getDiscardsRing()
+ {
+ // TODO
+ return 0L;
+ }
+
+ public Long getDiscardsLvq()
+ {
+ // TODO
+ return 0L;
+ }
+
+ public Long getDiscardsOverflow()
+ {
+ // TODO
+ return 0L;
+ }
+
+ public Long getDiscardsSubscriber()
+ {
+ // TODO
+ return 0L;
+ }
+
+ public Long getDiscardsPurge()
+ {
+ // TODO
+ return 0L;
+ }
+
+ public Long getReroutes()
+ {
+ // TODO
+ return 0L;
+ }
+
+ public Long getAbandoned()
+ {
+ // TODO
+ return 0L;
+ }
+
+ public Long getAbandonedViaAlt()
+ {
+ // TODO
+ return 0L;
+ }
+
public BrokerSchema.BrokerClass.EchoMethodResponseCommand echo(final BrokerSchema.BrokerClass.EchoMethodResponseCommandFactory factory,
final Long sequence,
final String body)
@@ -1064,6 +1267,96 @@ public class QMFService implements Confi
return _obj.getPersistentByteDequeues();
}
+ public Long getMsgFtdEnqueues()
+ {
+ // TODO
+ return 0L;
+ }
+
+ public Long getMsgFtdDequeues()
+ {
+ // TODO
+ return 0L;
+ }
+
+ public Long getByteFtdEnqueues()
+ {
+ // TODO
+ return 0L;
+ }
+
+ public Long getByteFtdDequeues()
+ {
+ // TODO
+ return 0L;
+ }
+
+ public Long getMsgFtdDepth()
+ {
+ // TODO
+ return 0L;
+ }
+
+ public Long getByteFtdDepth()
+ {
+ // TODO
+ return 0L;
+ }
+
+ public Long getReleases()
+ {
+ // TODO
+ return 0L;
+ }
+
+ public Long getAcquires()
+ {
+ // TODO
+ return 0L;
+ }
+
+ public Long getDiscardsTtl()
+ {
+ // TODO
+ return 0L;
+ }
+
+ public Long getDiscardsRing()
+ {
+ // TODO
+ return 0L;
+ }
+
+ public Long getDiscardsLvq()
+ {
+ // TODO
+ return 0L;
+ }
+
+ public Long getDiscardsOverflow()
+ {
+ // TODO
+ return 0L;
+ }
+
+ public Long getDiscardsSubscriber()
+ {
+ // TODO
+ return 0L;
+ }
+
+ public Long getDiscardsPurge()
+ {
+ // TODO
+ return 0L;
+ }
+
+ public Long getReroutes()
+ {
+ // TODO
+ return 0L;
+ }
+
public Long getConsumerCount()
{
return (long) _obj.getConsumerCount();
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQBrokerManagerMBean.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQBrokerManagerMBean.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQBrokerManagerMBean.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQBrokerManagerMBean.java Sat Mar 10 19:22:10 2012
@@ -18,16 +18,6 @@
*/
package org.apache.qpid.server;
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-
-import javax.management.JMException;
-import javax.management.MBeanException;
-import javax.management.MalformedObjectNameException;
-import javax.management.ObjectName;
-
import org.apache.qpid.AMQException;
import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.framing.FieldTable;
@@ -39,6 +29,8 @@ import org.apache.qpid.server.exchange.E
import org.apache.qpid.server.exchange.ExchangeFactory;
import org.apache.qpid.server.exchange.ExchangeRegistry;
import org.apache.qpid.server.exchange.ExchangeType;
+import org.apache.qpid.server.logging.actors.CurrentActor;
+import org.apache.qpid.server.logging.actors.ManagementActor;
import org.apache.qpid.server.management.AMQManagedObject;
import org.apache.qpid.server.management.ManagedObject;
import org.apache.qpid.server.queue.AMQQueue;
@@ -48,8 +40,15 @@ import org.apache.qpid.server.queue.Queu
import org.apache.qpid.server.store.DurableConfigurationStore;
import org.apache.qpid.server.virtualhost.VirtualHost;
import org.apache.qpid.server.virtualhost.VirtualHostImpl;
-import org.apache.qpid.server.logging.actors.CurrentActor;
-import org.apache.qpid.server.logging.actors.ManagementActor;
+
+import javax.management.JMException;
+import javax.management.MBeanException;
+import javax.management.MalformedObjectNameException;
+import javax.management.ObjectName;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
/**
* This MBean implements the broker management interface and exposes the
@@ -169,7 +168,7 @@ public class AMQBrokerManagerMBean exten
*/
public void createNewExchange(String exchangeName, String type, boolean durable) throws JMException, MBeanException
{
- CurrentActor.set(new ManagementActor(_logActor.getRootMessageLogger()));
+ CurrentActor.set(new ManagementActor(getLogActor().getRootMessageLogger()));
try
{
synchronized (_exchangeRegistry)
@@ -213,10 +212,10 @@ public class AMQBrokerManagerMBean exten
{
// TODO
// Check if the exchange is in use.
- // boolean inUse = false;
+
// Check if there are queue-bindings with the exchange and unregister
// when there are no bindings.
- CurrentActor.set(new ManagementActor(_logActor.getRootMessageLogger()));
+ CurrentActor.set(new ManagementActor(getLogActor().getRootMessageLogger()));
try
{
_exchangeRegistry.unregisterExchange(new AMQShortString(exchangeName), false);
@@ -256,7 +255,7 @@ public class AMQBrokerManagerMBean exten
throw new JMException("The queue \"" + queueName + "\" already exists.");
}
- CurrentActor.set(new ManagementActor(_logActor.getRootMessageLogger()));
+ CurrentActor.set(new ManagementActor(getLogActor().getRootMessageLogger()));
try
{
AMQShortString ownerShortString = null;
@@ -312,7 +311,7 @@ public class AMQBrokerManagerMBean exten
throw new JMException("The Queue " + queueName + " is not a registered queue.");
}
- CurrentActor.set(new ManagementActor(_logActor.getRootMessageLogger()));
+ CurrentActor.set(new ManagementActor(getLogActor().getRootMessageLogger()));
try
{
queue.delete();
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=1299257&r1=1299256&r2=1299257&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 Sat Mar 10 19:22:10 2012
@@ -20,8 +20,23 @@
*/
package org.apache.qpid.server;
-import org.apache.log4j.Logger;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.SortedSet;
+import java.util.TreeSet;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentSkipListSet;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+import org.apache.log4j.Logger;
import org.apache.qpid.AMQException;
import org.apache.qpid.AMQSecurityException;
import org.apache.qpid.framing.AMQMethodBody;
@@ -57,10 +72,10 @@ import org.apache.qpid.server.message.Me
import org.apache.qpid.server.message.MessageReference;
import org.apache.qpid.server.message.ServerMessage;
import org.apache.qpid.server.output.ProtocolOutputConverter;
+import org.apache.qpid.server.protocol.AMQConnectionModel;
import org.apache.qpid.server.protocol.AMQProtocolEngine;
import org.apache.qpid.server.protocol.AMQProtocolSession;
import org.apache.qpid.server.protocol.AMQSessionModel;
-import org.apache.qpid.server.protocol.AMQConnectionModel;
import org.apache.qpid.server.queue.AMQQueue;
import org.apache.qpid.server.queue.BaseQueue;
import org.apache.qpid.server.queue.InboundMessageAdapter;
@@ -75,18 +90,11 @@ import org.apache.qpid.server.subscripti
import org.apache.qpid.server.subscription.SubscriptionFactoryImpl;
import org.apache.qpid.server.subscription.SubscriptionImpl;
import org.apache.qpid.server.txn.AsyncAutoCommitTransaction;
-import org.apache.qpid.server.txn.AutoCommitTransaction;
import org.apache.qpid.server.txn.LocalTransaction;
import org.apache.qpid.server.txn.ServerTransaction;
import org.apache.qpid.server.virtualhost.VirtualHost;
import org.apache.qpid.transport.TransportException;
-import java.util.*;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicLong;
-
public class AMQChannel implements SessionConfig, AMQSessionModel, AsyncAutoCommitTransaction.FutureRecorder
{
public static final int DEFAULT_PREFETCH = 4096;
@@ -122,7 +130,7 @@ public class AMQChannel implements Sessi
private IncomingMessage _currentMessage;
/** Maps from consumer tag to subscription instance. Allows us to unsubscribe from a queue. */
- protected final Map<AMQShortString, Subscription> _tag2SubscriptionMap = new HashMap<AMQShortString, Subscription>();
+ private final Map<AMQShortString, Subscription> _tag2SubscriptionMap = new HashMap<AMQShortString, Subscription>();
private final MessageStore _messageStore;
@@ -148,7 +156,7 @@ public class AMQChannel implements Sessi
private final AMQProtocolSession _session;
private AtomicBoolean _closing = new AtomicBoolean(false);
- private final ConcurrentMap<AMQQueue, Boolean> _blockingQueues = new ConcurrentHashMap<AMQQueue, Boolean>();
+ private final Set<AMQQueue> _blockingQueues = new ConcurrentSkipListSet<AMQQueue>();
private final AtomicBoolean _blocking = new AtomicBoolean(false);
@@ -267,7 +275,7 @@ public class AMQChannel implements Sessi
{
throw new AMQSecurityException("Permission denied: " + e.getName());
}
- _currentMessage = new IncomingMessage(info);
+ _currentMessage = new IncomingMessage(info, getProtocolSession().getReference());
_currentMessage.setExchange(e);
}
@@ -289,26 +297,9 @@ public class AMQChannel implements Sessi
_currentMessage.setExpiration();
+ _currentMessage.headersReceived(getProtocolSession().getLastReceivedTime());
- MessageMetaData mmd = _currentMessage.headersReceived(getProtocolSession().getLastReceivedTime());
- final StoredMessage<MessageMetaData> handle = _messageStore.addMessage(mmd);
- _currentMessage.setStoredMessage(handle);
-
- routeCurrentMessage();
-
-
- _transaction.addPostTransactionAction(new ServerTransaction.Action()
- {
-
- public void postCommit()
- {
- }
-
- public void onRollback()
- {
- handle.remove();
- }
- });
+ _currentMessage.route();
deliverCurrentMessageIfComplete();
}
@@ -340,17 +331,41 @@ public class AMQChannel implements Sessi
{
_actor.message(ExchangeMessages.DISCARDMSG(_currentMessage.getExchange().asString(), _currentMessage.getRoutingKey()));
}
-
}
else
{
+ final StoredMessage<MessageMetaData> handle = _messageStore.addMessage(_currentMessage.getMessageMetaData());
+ _currentMessage.setStoredMessage(handle);
+ int bodyCount = _currentMessage.getBodyCount();
+ if(bodyCount > 0)
+ {
+ long bodyLengthReceived = 0;
+ for(int i = 0 ; i < bodyCount ; i++)
+ {
+ ContentChunk contentChunk = _currentMessage.getContentChunk(i);
+ handle.addContent((int)bodyLengthReceived, ByteBuffer.wrap(contentChunk.getData()));
+ bodyLengthReceived += contentChunk.getSize();
+ }
+ }
+
+ _transaction.addPostTransactionAction(new ServerTransaction.Action()
+ {
+ public void postCommit()
+ {
+ }
+
+ public void onRollback()
+ {
+ handle.remove();
+ }
+ });
+
_transaction.enqueue(destinationQueues, _currentMessage, new MessageDeliveryAction(_currentMessage, destinationQueues), getProtocolSession().getLastReceivedTime());
incrementOutstandingTxnsIfNecessary();
- updateTransactionalActivity();
+ updateTransactionalActivity();
+ _currentMessage.getStoredMessage().flushToStore();
}
}
- _currentMessage.getStoredMessage().flushToStore();
-
}
finally
{
@@ -377,9 +392,6 @@ public class AMQChannel implements Sessi
try
{
-
- // returns true iff the message was delivered (i.e. if all data was
- // received
final ContentChunk contentChunk =
_session.getMethodRegistry().getProtocolVersionMethodConverter().convertToContentChunk(contentBody);
@@ -403,11 +415,6 @@ public class AMQChannel implements Sessi
}
}
- protected void routeCurrentMessage() throws AMQException
- {
- _currentMessage.route();
- }
-
public long getNextDeliveryTag()
{
return ++_deliveryTag;
@@ -777,20 +784,6 @@ public class AMQChannel implements Sessi
AMQQueue queue = message.getQueue();
- // Our Java Client will always suspend the channel when resending!
- // If the client has requested the messages be resent then it is
- // their responsibility to ensure that thay are capable of receiving them
- // i.e. The channel hasn't been server side suspended.
- // if (isSuspended())
- // {
- // _logger.info("Channel is suspended so requeuing");
- // //move this message to requeue
- // msgToRequeue.add(message);
- // }
- // else
- // {
- // release to allow it to be delivered
-
// Without any details from the client about what has been processed we have to mark
// all messages in the unacked map as redelivered.
message.setRedelivered();
@@ -1116,7 +1109,7 @@ public class AMQChannel implements Sessi
AMQMessage message = new AMQMessage(incomingMessage.getStoredMessage());
message.setExpiration(incomingMessage.getExpiration());
- message.setClientIdentifier(_session);
+ message.setConnectionIdentifier(_session.getReference());
return message;
}
@@ -1370,7 +1363,7 @@ public class AMQChannel implements Sessi
public void block(AMQQueue queue)
{
- if(_blockingQueues.putIfAbsent(queue, Boolean.TRUE) == null)
+ if(_blockingQueues.add(queue))
{
if(_blocking.compareAndSet(false,true))
@@ -1394,6 +1387,16 @@ public class AMQChannel implements Sessi
}
}
+ public boolean onSameConnection(InboundMessage inbound)
+ {
+ if(inbound instanceof IncomingMessage)
+ {
+ IncomingMessage incoming = (IncomingMessage) inbound;
+ return getProtocolSession().getReference() == incoming.getConnectionReference();
+ }
+ return false;
+ }
+
private void flow(boolean flow)
{
MethodRegistry methodRegistry = _session.getMethodRegistry();
@@ -1623,4 +1626,8 @@ public class AMQChannel implements Sessi
}
}
+ public int compareTo(AMQSessionModel session)
+ {
+ return getId().toString().compareTo(session.getID().toString());
+ }
}
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/Broker.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/Broker.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/Broker.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/Broker.java Sat Mar 10 19:22:10 2012
@@ -20,24 +20,10 @@
*/
package org.apache.qpid.server;
-import static org.apache.qpid.transport.ConnectionSettings.WILDCARD_ADDRESS;
-
-import java.io.File;
-import java.io.IOException;
-import java.io.InputStream;
-import java.net.InetAddress;
-import java.net.InetSocketAddress;
-import java.util.EnumSet;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Properties;
-import java.util.Set;
-import java.util.logging.*;
-
-import javax.net.ssl.SSLContext;
-
+import org.apache.log4j.Logger;
import org.apache.log4j.PropertyConfigurator;
import org.apache.log4j.xml.QpidLog4JConfigurator;
+
import org.apache.qpid.server.configuration.ServerConfiguration;
import org.apache.qpid.server.configuration.ServerNetworkTransportConfiguration;
import org.apache.qpid.server.configuration.management.ConfigurationManagementMBean;
@@ -58,10 +44,33 @@ import org.apache.qpid.transport.Network
import org.apache.qpid.transport.network.IncomingNetworkTransport;
import org.apache.qpid.transport.network.Transport;
+import static org.apache.qpid.transport.ConnectionSettings.WILDCARD_ADDRESS;
+
+import javax.net.ssl.SSLContext;
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.util.EnumSet;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Properties;
+import java.util.Set;
+import java.util.logging.ConsoleHandler;
+import java.util.logging.FileHandler;
+import java.util.logging.Formatter;
+import java.util.logging.Handler;
+import java.util.logging.Level;
+import java.util.logging.LogRecord;
+
public class Broker
{
+ private static final Logger LOGGER = Logger.getLogger(Broker.class);
+
private static final int IPV4_ADDRESS_LENGTH = 4;
private static final char IPV4_LITERAL_SEPARATOR = '.';
+ private volatile Thread _shutdownHookThread;
private java.util.logging.Logger FRAME_LOGGER;
private java.util.logging.Logger RAW_LOGGER;
@@ -79,7 +88,14 @@ public class Broker
public void shutdown()
{
- ApplicationRegistry.remove();
+ try
+ {
+ removeShutdownHook();
+ }
+ finally
+ {
+ ApplicationRegistry.remove();
+ }
}
public void startup() throws Exception
@@ -93,6 +109,7 @@ public class Broker
{
CurrentActor.set(new BrokerActor(new SystemOutMessageLogger()));
startupImpl(options);
+ addShutdownHook();
}
finally
{
@@ -185,32 +202,37 @@ public class Broker
bindAddr = serverConfig.getBind();
}
- InetAddress bindAddress = null;
+ InetAddress bindAddress;
if (bindAddr.equals(WILDCARD_ADDRESS))
{
- bindAddress = new InetSocketAddress(0).getAddress();
+ bindAddress = null;
}
else
{
- bindAddress = InetAddress.getByAddress(parseIP(bindAddr));
+ bindAddress = InetAddress.getByName(bindAddr);
}
- String hostName = bindAddress.getCanonicalHostName();
+
+ final AmqpProtocolVersion defaultSupportedProtocolReply = serverConfig.getDefaultSupportedProtocolReply();
if (!serverConfig.getSSLOnly())
{
for(int port : ports)
{
+ final InetSocketAddress inetSocketAddress = new InetSocketAddress(bindAddress, port);
+
final Set<AmqpProtocolVersion> supported =
- getSupportedVersions(port, exclude_0_10, exclude_0_9_1, exclude_0_9, exclude_0_8);
+ getSupportedVersions(port, exclude_0_10, exclude_0_9_1, exclude_0_9, exclude_0_8, serverConfig);
+
final NetworkTransportConfiguration settings =
- new ServerNetworkTransportConfiguration(serverConfig, port, bindAddress.getHostName(), Transport.TCP);
+ new ServerNetworkTransportConfiguration(serverConfig, inetSocketAddress, Transport.TCP);
final IncomingNetworkTransport transport = Transport.getIncomingTransportInstance();
final MultiVersionProtocolEngineFactory protocolEngineFactory =
- new MultiVersionProtocolEngineFactory(hostName, supported);
+ new MultiVersionProtocolEngineFactory(supported, defaultSupportedProtocolReply);
transport.accept(settings, protocolEngineFactory, null);
- ApplicationRegistry.getInstance().addAcceptor(new InetSocketAddress(bindAddress, port),
+
+ ApplicationRegistry.getInstance().addAcceptor(inetSocketAddress,
new QpidAcceptor(transport,"TCP"));
CurrentActor.get().message(BrokerMessages.LISTENING("TCP", port));
}
@@ -220,22 +242,25 @@ public class Broker
{
final String keystorePath = serverConfig.getConnectorKeyStorePath();
final String keystorePassword = serverConfig.getConnectorKeyStorePassword();
- final String certType = serverConfig.getConnectorCertType();
- final SSLContext sslContext = SSLContextFactory.buildServerContext(keystorePath, keystorePassword, certType);
+ final String keyManagerFactoryAlgorithm = serverConfig.getConnectorKeyManagerFactoryAlgorithm();
+ final SSLContext sslContext = SSLContextFactory.buildServerContext(keystorePath, keystorePassword, keyManagerFactoryAlgorithm);
for(int sslPort : sslPorts)
{
+ final InetSocketAddress inetSocketAddress = new InetSocketAddress(bindAddress, sslPort);
+
final Set<AmqpProtocolVersion> supported =
- getSupportedVersions(sslPort, exclude_0_10, exclude_0_9_1, exclude_0_9, exclude_0_8);
+ getSupportedVersions(sslPort, exclude_0_10, exclude_0_9_1, exclude_0_9, exclude_0_8, serverConfig);
final NetworkTransportConfiguration settings =
- new ServerNetworkTransportConfiguration(serverConfig, sslPort, bindAddress.getHostName(), Transport.TCP);
+ new ServerNetworkTransportConfiguration(serverConfig, inetSocketAddress, Transport.TCP);
final IncomingNetworkTransport transport = Transport.getIncomingTransportInstance();
final MultiVersionProtocolEngineFactory protocolEngineFactory =
- new MultiVersionProtocolEngineFactory(hostName, supported);
+ new MultiVersionProtocolEngineFactory(supported, defaultSupportedProtocolReply);
transport.accept(settings, protocolEngineFactory, sslContext);
- ApplicationRegistry.getInstance().addAcceptor(new InetSocketAddress(bindAddress, sslPort),
+
+ ApplicationRegistry.getInstance().addAcceptor(inetSocketAddress,
new QpidAcceptor(transport,"TCP"));
CurrentActor.get().message(BrokerMessages.LISTENING("TCP/SSL", sslPort));
}
@@ -252,23 +277,24 @@ public class Broker
private static Set<AmqpProtocolVersion> getSupportedVersions(final int port, final Set<Integer> exclude_0_10,
final Set<Integer> exclude_0_9_1, final Set<Integer> exclude_0_9,
- final Set<Integer> exclude_0_8)
+ final Set<Integer> exclude_0_8,
+ final ServerConfiguration serverConfig)
{
final EnumSet<AmqpProtocolVersion> supported = EnumSet.allOf(AmqpProtocolVersion.class);
- if(exclude_0_10.contains(port))
+ if(exclude_0_10.contains(port) || !serverConfig.isAmqp010enabled())
{
supported.remove(AmqpProtocolVersion.v0_10);
}
- if(exclude_0_9_1.contains(port))
+ if(exclude_0_9_1.contains(port) || !serverConfig.isAmqp091enabled())
{
supported.remove(AmqpProtocolVersion.v0_9_1);
}
- if(exclude_0_9.contains(port))
+ if(exclude_0_9.contains(port) || !serverConfig.isAmqp09enabled())
{
supported.remove(AmqpProtocolVersion.v0_9);
}
- if(exclude_0_8.contains(port))
+ if(exclude_0_8.contains(port) || !serverConfig.isAmqp08enabled())
{
supported.remove(AmqpProtocolVersion.v0_8);
}
@@ -354,34 +380,6 @@ public class Broker
}
}
- private byte[] parseIP(String address) throws Exception
- {
- char[] literalBuffer = address.toCharArray();
- int byteCount = 0;
- int currByte = 0;
- byte[] ip = new byte[IPV4_ADDRESS_LENGTH];
- for (int i = 0; i < literalBuffer.length; i++)
- {
- char currChar = literalBuffer[i];
- if ((currChar >= '0') && (currChar <= '9'))
- {
- currByte = (currByte * 10) + (Character.digit(currChar, 10) & 0xFF);
- }
-
- if (currChar == IPV4_LITERAL_SEPARATOR || (i + 1 == literalBuffer.length))
- {
- ip[byteCount++] = (byte) currByte;
- currByte = 0;
- }
- }
-
- if (byteCount != 4)
- {
- throw new Exception("Invalid IP address: " + address);
- }
- return ip;
- }
-
private void configureLogging(File logConfigFile, long logWatchTime) throws InitException, IOException
{
if (logConfigFile.exists() && logConfigFile.canRead())
@@ -447,7 +445,59 @@ public class Broker
blm.register();
}
- private java.util.logging.Logger updateLogger(final String logType, String logFileName) throws IOException
+ private void addShutdownHook()
+ {
+ Thread shutdownHookThread = new Thread(new ShutdownService());
+ shutdownHookThread.setName("QpidBrokerShutdownHook");
+
+ Runtime.getRuntime().addShutdownHook(shutdownHookThread);
+ _shutdownHookThread = shutdownHookThread;
+
+ LOGGER.debug("Added shutdown hook");
+ }
+
+ private void removeShutdownHook()
+ {
+ Thread shutdownThread = _shutdownHookThread;
+
+ //if there is a shutdown thread and we aren't it, we should remove it
+ if(shutdownThread != null && !(Thread.currentThread() == shutdownThread))
+ {
+ LOGGER.debug("Removing shutdown hook");
+
+ _shutdownHookThread = null;
+
+ boolean removed = false;
+ try
+ {
+ removed = Runtime.getRuntime().removeShutdownHook(shutdownThread);
+ }
+ catch(IllegalStateException ise)
+ {
+ //ignore, means the JVM is already shutting down
+ }
+
+ if(LOGGER.isDebugEnabled())
+ {
+ LOGGER.debug("Removed shutdown hook: " + removed);
+ }
+ }
+ else
+ {
+ LOGGER.debug("Skipping shutdown hook removal as there either isnt one, or we are it.");
+ }
+ }
+
+ private class ShutdownService implements Runnable
+ {
+ public void run()
+ {
+ LOGGER.debug("Shutdown hook running");
+ Broker.this.shutdown();
+ }
+ }
+
+ private java.util.logging.Logger updateLogger(final String logType, String logFileName) throws IOException
{
java.util.logging.Logger logger = java.util.logging.Logger.getLogger(logType);
logger.setLevel(Level.FINE);
@@ -479,4 +529,5 @@ public class Broker
logger.addHandler(handler);
return logger;
}
+
}
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/BrokerOptions.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/BrokerOptions.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/BrokerOptions.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/BrokerOptions.java Sat Mar 10 19:22:10 2012
@@ -20,23 +20,28 @@
*/
package org.apache.qpid.server;
+import org.osgi.framework.BundleContext;
+
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
-import org.osgi.framework.BundleContext;
-
public class BrokerOptions
{
- /** serialVersionUID */
- private static final long serialVersionUID = 8051825964945442234L;
-
public static final String DEFAULT_CONFIG_FILE = "etc/config.xml";
public static final String DEFAULT_LOG_CONFIG_FILE = "etc/log4j.xml";
public static final String QPID_HOME = "QPID_HOME";
+ public static final String PORTS = "p";
+ public static final String SSL_PORTS = "s";
+ public static final String BIND = "b";
+ public static final String MANAGEMENT = "m";
+ public static final String LOG_CONFIG = "l";
+ public static final String WATCH = "w";
+ public static final String CONFIG = "c";
+
private final Set<Integer> _ports = new HashSet<Integer>();
private final Set<Integer> _sslPorts = new HashSet<Integer>();
private final Map<ProtocolExclusion,Set<Integer>> _exclusionMap = new HashMap<ProtocolExclusion, Set<Integer>>();
@@ -50,7 +55,6 @@ public class BrokerOptions
private Integer _logWatchFrequency = 0;
-
public void addPort(final int port)
{
_ports.add(port);
@@ -110,7 +114,6 @@ public class BrokerOptions
{
_jmxPortConnectorServer = jmxPortConnectorServer;
}
-
public String getQpidHome()
{
return System.getProperty(QPID_HOME);
@@ -157,6 +160,7 @@ public class BrokerOptions
_logWatchFrequency = logWatchFrequency;
}
+
public BundleContext getBundleContext()
{
return _bundleContext ;
@@ -166,5 +170,4 @@ public class BrokerOptions
{
_bundleContext = bundleContext;
}
-
}
\ No newline at end of file
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=1299257&r1=1299256&r2=1299257&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 Sat Mar 10 19:22:10 2012
@@ -20,14 +20,15 @@
*/
package org.apache.qpid.server;
+import org.apache.log4j.Logger;
+
+import org.apache.qpid.AMQException;
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.txn.ServerTransaction;
import org.apache.qpid.server.txn.AutoCommitTransaction;
-import org.apache.qpid.AMQException;
-import org.apache.log4j.Logger;
+import org.apache.qpid.server.txn.ServerTransaction;
import java.util.Map;
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=1299257&r1=1299256&r2=1299257&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 Sat Mar 10 19:22:10 2012
@@ -28,6 +28,8 @@ 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.common.QpidProperties;
+import org.apache.qpid.framing.ProtocolVersion;
import org.apache.qpid.server.Broker.InitException;
import org.apache.qpid.server.registry.ApplicationRegistry;
@@ -124,7 +126,7 @@ public class Main
OPTIONS.addOption(OPTION_JMX_PORT_CONNECTOR_SERVER);
}
- private CommandLine commandLine;
+ protected CommandLine _commandLine;
public static void main(String[] args)
{
@@ -160,7 +162,7 @@ public class Main
{
try
{
- commandLine = new PosixParser().parse(OPTIONS, args);
+ _commandLine = new PosixParser().parse(OPTIONS, args);
return true;
}
@@ -176,66 +178,93 @@ public class Main
protected void execute() throws Exception
{
- BrokerOptions options = new BrokerOptions();
- String configFile = commandLine.getOptionValue(OPTION_CONFIG_FILE.getOpt());
- if(configFile != null)
+ if (_commandLine.hasOption(OPTION_HELP.getOpt()))
{
- options.setConfigFile(configFile);
+ final HelpFormatter formatter = new HelpFormatter();
+ formatter.printHelp("Qpid", OPTIONS, true);
}
-
- String logWatchConfig = commandLine.getOptionValue(OPTION_LOG_WATCH.getOpt());
- if(logWatchConfig != null)
+ else if (_commandLine.hasOption(OPTION_VERSION.getOpt()))
{
- options.setLogWatchFrequency(Integer.parseInt(logWatchConfig));
- }
+ final StringBuilder protocol = new StringBuilder("AMQP version(s) [major.minor]: ");
+ boolean first = true;
+ for (final ProtocolVersion pv : ProtocolVersion.getSupportedProtocolVersions())
+ {
+ if (first)
+ {
+ first = false;
+ }
+ else
+ {
+ protocol.append(", ");
+ }
- String logConfig = commandLine.getOptionValue(OPTION_LOG_CONFIG_FILE.getOpt());
- if(logConfig != null)
- {
- options.setLogConfigFile(logConfig);
+ protocol.append(pv.getMajorVersion()).append('-').append(pv.getMinorVersion());
+ }
+ System.out.println(QpidProperties.getVersionString() + " (" + protocol + ")");
}
-
- String jmxPortRegistryServer = commandLine.getOptionValue(OPTION_JMX_PORT_REGISTRY_SERVER.getOpt());
- if(jmxPortRegistryServer != null)
+ else
{
- options.setJmxPortRegistryServer(Integer.parseInt(jmxPortRegistryServer));
- }
+ BrokerOptions options = new BrokerOptions();
+ String configFile = _commandLine.getOptionValue(OPTION_CONFIG_FILE.getOpt());
+ if(configFile != null)
+ {
+ options.setConfigFile(configFile);
+ }
- String jmxPortConnectorServer = commandLine.getOptionValue(OPTION_JMX_PORT_CONNECTOR_SERVER.getLongOpt());
- if(jmxPortConnectorServer != null)
- {
- options.setJmxPortConnectorServer(Integer.parseInt(jmxPortConnectorServer));
- }
+ String logWatchConfig = _commandLine.getOptionValue(OPTION_LOG_WATCH.getOpt());
+ if(logWatchConfig != null)
+ {
+ options.setLogWatchFrequency(Integer.parseInt(logWatchConfig));
+ }
- String bindAddr = commandLine.getOptionValue(OPTION_BIND.getOpt());
- if (bindAddr != null)
- {
- options.setBind(bindAddr);
- }
+ String logConfig = _commandLine.getOptionValue(OPTION_LOG_CONFIG_FILE.getOpt());
+ if(logConfig != null)
+ {
+ options.setLogConfigFile(logConfig);
+ }
- String[] portStr = commandLine.getOptionValues(OPTION_PORT.getOpt());
- if(portStr != null)
- {
- parsePortArray(options, portStr, false);
- for(ProtocolExclusion pe : ProtocolExclusion.values())
+ String jmxPortRegistryServer = _commandLine.getOptionValue(OPTION_JMX_PORT_REGISTRY_SERVER.getOpt());
+ if(jmxPortRegistryServer != null)
{
- parsePortArray(options, commandLine.getOptionValues(pe.getExcludeName()), pe);
+ options.setJmxPortRegistryServer(Integer.parseInt(jmxPortRegistryServer));
}
- }
- String[] sslPortStr = commandLine.getOptionValues(OPTION_SSLPORT.getOpt());
- if(sslPortStr != null)
- {
- parsePortArray(options, sslPortStr, true);
- for(ProtocolExclusion pe : ProtocolExclusion.values())
+ String jmxPortConnectorServer = _commandLine.getOptionValue(OPTION_JMX_PORT_CONNECTOR_SERVER.getLongOpt());
+ if(jmxPortConnectorServer != null)
{
- parsePortArray(options, commandLine.getOptionValues(pe.getExcludeName()), pe);
+ options.setJmxPortConnectorServer(Integer.parseInt(jmxPortConnectorServer));
}
- }
-
- setExceptionHandler();
-
- startBroker(options);
+
+ String bindAddr = _commandLine.getOptionValue(OPTION_BIND.getOpt());
+ if (bindAddr != null)
+ {
+ options.setBind(bindAddr);
+ }
+
+ String[] portStr = _commandLine.getOptionValues(OPTION_PORT.getOpt());
+ if(portStr != null)
+ {
+ parsePortArray(options, portStr, false);
+ for(ProtocolExclusion pe : ProtocolExclusion.values())
+ {
+ parsePortArray(options, _commandLine.getOptionValues(pe.getExcludeName()), pe);
+ }
+ }
+
+ String[] sslPortStr = _commandLine.getOptionValues(OPTION_SSLPORT.getOpt());
+ if(sslPortStr != null)
+ {
+ parsePortArray(options, sslPortStr, true);
+ for(ProtocolExclusion pe : ProtocolExclusion.values())
+ {
+ parsePortArray(options, _commandLine.getOptionValues(pe.getExcludeName()), pe);
+ }
+ }
+
+ setExceptionHandler();
+
+ startBroker(options);
+ }
}
protected void setExceptionHandler()
@@ -273,6 +302,7 @@ public class Main
{
public void uncaughtException(final Thread t, final Throwable e)
{
+ boolean continueOnError = Boolean.getBoolean("qpid.broker.exceptionHandler.continue");
try
{
System.err.println("########################################################################");
@@ -282,17 +312,20 @@ public class Main
System.err.print(" in Thread ");
System.err.println(t.getName());
System.err.println("#");
- System.err.println("# Exiting");
+ System.err.println(continueOnError ? "# Forced to continue by JVM setting 'qpid.broker.exceptionHandler.continue'" : "# 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);
+ logger.error("Uncaught exception, " + (continueOnError ? "continuing." : "shutting down."), e);
}
finally
{
- Runtime.getRuntime().halt(1);
+ if (!continueOnError)
+ {
+ Runtime.getRuntime().halt(1);
+ }
}
}
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=1299257&r1=1299256&r2=1299257&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 Sat Mar 10 19:22:10 2012
@@ -20,13 +20,12 @@
*/
package org.apache.qpid.server.ack;
-import java.util.Collection;
-import java.util.Set;
-import java.util.Map;
-
import org.apache.qpid.AMQException;
import org.apache.qpid.server.queue.QueueEntry;
+import java.util.Collection;
+import java.util.Set;
+
public interface UnacknowledgedMessageMap
{
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=1299257&r1=1299256&r2=1299257&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 Sat Mar 10 19:22:10 2012
@@ -20,14 +20,14 @@
*/
package org.apache.qpid.server.ack;
+import org.apache.qpid.AMQException;
+import org.apache.qpid.server.queue.QueueEntry;
+
import java.util.Collection;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Set;
-import org.apache.qpid.AMQException;
-import org.apache.qpid.server.queue.QueueEntry;
-
public class UnacknowledgedMessageMapImpl implements UnacknowledgedMessageMap
{
private final Object _lock = new Object();
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=1299257&r1=1299256&r2=1299257&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 Sat Mar 10 19:22:10 2012
@@ -94,7 +94,7 @@ public class Binding
return true;
}
- if (o == null || !(o instanceof Binding))
+ if (!(o instanceof Binding))
{
return false;
}
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/binding/BindingFactory.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/BindingFactory.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/binding/BindingFactory.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/binding/BindingFactory.java Sat Mar 10 19:22:10 2012
@@ -20,10 +20,6 @@
*/
package org.apache.qpid.server.binding;
-import java.util.Collections;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-
import org.apache.qpid.AMQException;
import org.apache.qpid.AMQInternalException;
import org.apache.qpid.AMQSecurityException;
@@ -33,8 +29,6 @@ import org.apache.qpid.server.configurat
import org.apache.qpid.server.configuration.BindingConfigType;
import org.apache.qpid.server.configuration.ConfigStore;
import org.apache.qpid.server.configuration.ConfiguredObject;
-import org.apache.qpid.server.configuration.QueueConfiguration;
-import org.apache.qpid.server.configuration.plugins.ConfigurationPlugin;
import org.apache.qpid.server.exchange.Exchange;
import org.apache.qpid.server.logging.actors.CurrentActor;
import org.apache.qpid.server.logging.messages.BindingMessages;
@@ -43,6 +37,10 @@ import org.apache.qpid.server.queue.AMQQ
import org.apache.qpid.server.store.DurableConfigurationStore;
import org.apache.qpid.server.virtualhost.VirtualHost;
+import java.util.Collections;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
public class BindingFactory
{
private final VirtualHost _virtualHost;
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/BindingConfigType.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/BindingConfigType.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/BindingConfigType.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/BindingConfigType.java Sat Mar 10 19:22:10 2012
@@ -21,9 +21,11 @@
package org.apache.qpid.server.configuration;
-import org.apache.qpid.server.exchange.ExchangeType;
-
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
public final class BindingConfigType extends ConfigObjectType<BindingConfigType, BindingConfig>
{
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/BridgeConfigType.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/BridgeConfigType.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/BridgeConfigType.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/BridgeConfigType.java Sat Mar 10 19:22:10 2012
@@ -21,9 +21,10 @@
package org.apache.qpid.server.configuration;
-import org.apache.qpid.server.exchange.ExchangeType;
-
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
public final class BridgeConfigType extends ConfigObjectType<BridgeConfigType, BridgeConfig>
{
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/BrokerConfigType.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/BrokerConfigType.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/BrokerConfigType.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/BrokerConfigType.java Sat Mar 10 19:22:10 2012
@@ -21,7 +21,10 @@
package org.apache.qpid.server.configuration;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
public final class BrokerConfigType extends ConfigObjectType<BrokerConfigType, BrokerConfig>
{
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=1299257&r1=1299256&r2=1299257&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 Sat Mar 10 19:22:10 2012
@@ -21,9 +21,9 @@
package org.apache.qpid.server.configuration;
-import java.util.UUID;
import java.util.Collection;
import java.util.Collections;
+import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicLong;
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/ConfigurationManager.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/ConfigurationManager.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/ConfigurationManager.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/ConfigurationManager.java Sat Mar 10 19:22:10 2012
@@ -20,17 +20,18 @@
*/
package org.apache.qpid.server.configuration;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-import java.util.Map.Entry;
-
import org.apache.commons.configuration.Configuration;
import org.apache.commons.configuration.ConfigurationException;
+
import org.apache.qpid.server.configuration.plugins.ConfigurationPlugin;
import org.apache.qpid.server.configuration.plugins.ConfigurationPluginFactory;
import org.apache.qpid.server.registry.ApplicationRegistry;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+
public class ConfigurationManager
{
public List<ConfigurationPlugin> getConfigurationPlugins(String configurationElement, Configuration configuration) throws ConfigurationException
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/ConnectionConfigType.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/ConnectionConfigType.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/ConnectionConfigType.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/ConnectionConfigType.java Sat Mar 10 19:22:10 2012
@@ -21,9 +21,10 @@
package org.apache.qpid.server.configuration;
-import org.apache.qpid.server.exchange.ExchangeType;
-
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
public final class ConnectionConfigType extends ConfigObjectType<ConnectionConfigType, ConnectionConfig>
{
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/ExchangeConfigType.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/ExchangeConfigType.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/ExchangeConfigType.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/ExchangeConfigType.java Sat Mar 10 19:22:10 2012
@@ -21,9 +21,11 @@
package org.apache.qpid.server.configuration;
-import org.apache.qpid.server.exchange.ExchangeType;
-
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
public final class ExchangeConfigType extends ConfigObjectType<ExchangeConfigType, ExchangeConfig>
{
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=1299257&r1=1299256&r2=1299257&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 Sat Mar 10 19:22:10 2012
@@ -21,11 +21,6 @@
package org.apache.qpid.server.configuration;
-import org.apache.qpid.server.exchange.ExchangeType;
-
-import java.util.Map;
-
-
public interface LinkConfig extends ConfiguredObject<LinkConfigType, LinkConfig>
{
VirtualHostConfig getVirtualHost();
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/LinkConfigType.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/LinkConfigType.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/LinkConfigType.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/LinkConfigType.java Sat Mar 10 19:22:10 2012
@@ -21,9 +21,10 @@
package org.apache.qpid.server.configuration;
-import org.apache.qpid.server.exchange.ExchangeType;
-
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
public final class LinkConfigType extends ConfigObjectType<LinkConfigType, LinkConfig>
{
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/QueueConfig.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/QueueConfig.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/QueueConfig.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/QueueConfig.java Sat Mar 10 19:22:10 2012
@@ -21,10 +21,10 @@
package org.apache.qpid.server.configuration;
-import java.util.Map;
-
import org.apache.qpid.AMQException;
+import java.util.Map;
+
public interface QueueConfig extends ConfiguredObject<QueueConfigType, QueueConfig>
{
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/QueueConfigType.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/QueueConfigType.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/QueueConfigType.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/QueueConfigType.java Sat Mar 10 19:22:10 2012
@@ -21,9 +21,11 @@
package org.apache.qpid.server.configuration;
-import org.apache.qpid.server.exchange.ExchangeType;
-
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
public final class QueueConfigType extends ConfigObjectType<QueueConfigType, QueueConfig>
{
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/QueueConfiguration.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/QueueConfiguration.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/QueueConfiguration.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/QueueConfiguration.java Sat Mar 10 19:22:10 2012
@@ -20,13 +20,14 @@
*/
package org.apache.qpid.server.configuration;
-import java.util.List;
-
import org.apache.commons.configuration.CompositeConfiguration;
import org.apache.commons.configuration.ConfigurationException;
+
import org.apache.qpid.exchange.ExchangeDefaults;
import org.apache.qpid.server.configuration.plugins.ConfigurationPlugin;
+import java.util.List;
+
public class QueueConfiguration extends ConfigurationPlugin
{
private String _name;
@@ -211,7 +212,7 @@ public class QueueConfiguration extends
public void validateConfiguration() throws ConfigurationException
{
- if (_configuration.isEmpty())
+ if (getConfig().isEmpty())
{
throw new ConfigurationException("Queue section cannot be empty.");
}
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=1299257&r1=1299256&r2=1299257&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 Sat Mar 10 19:22:10 2012
@@ -20,17 +20,6 @@
package org.apache.qpid.server.configuration;
-import static org.apache.qpid.transport.ConnectionSettings.WILDCARD_ADDRESS;
-
-import java.io.File;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Locale;
-import java.util.Map;
-import java.util.Map.Entry;
-
import org.apache.commons.configuration.CompositeConfiguration;
import org.apache.commons.configuration.Configuration;
import org.apache.commons.configuration.ConfigurationException;
@@ -39,14 +28,29 @@ import org.apache.commons.configuration.
import org.apache.commons.configuration.SystemConfiguration;
import org.apache.commons.configuration.XMLConfiguration;
import org.apache.log4j.Logger;
+
import org.apache.qpid.server.configuration.plugins.ConfigurationPlugin;
import org.apache.qpid.server.exchange.DefaultExchangeFactory;
+import org.apache.qpid.server.protocol.AmqpProtocolVersion;
import org.apache.qpid.server.queue.AMQQueueFactory;
import org.apache.qpid.server.registry.ApplicationRegistry;
import org.apache.qpid.server.signal.SignalHandlerTask;
import org.apache.qpid.server.virtualhost.VirtualHost;
import org.apache.qpid.server.virtualhost.VirtualHostRegistry;
+import static org.apache.qpid.transport.ConnectionSettings.WILDCARD_ADDRESS;
+
+import java.io.File;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Map.Entry;
+
+import javax.net.ssl.KeyManagerFactory;
+
public class ServerConfiguration extends ConfigurationPlugin
{
protected static final Logger _logger = Logger.getLogger(ServerConfiguration.class);
@@ -84,10 +88,14 @@ public class ServerConfiguration extends
public static final String MGMT_JMXPORT_CONNECTORSERVER = "management.jmxport.connectorServer";
public static final String STATUS_UPDATES = "status-updates";
public static final String ADVANCED_LOCALE = "advanced.locale";
+ public static final String CONNECTOR_AMQP010ENABLED = "connector.amqp010enabled";
+ public static final String CONNECTOR_AMQP091ENABLED = "connector.amqp091enabled";
+ public static final String CONNECTOR_AMQP09ENABLED = "connector.amqp09enabled";
+ public static final String CONNECTOR_AMQP08ENABLED = "connector.amqp08enabled";
+ public static final String CONNECTOR_AMQP_SUPPORTED_REPLY = "connector.amqpDefaultSupportedProtocolReply";
{
envVarMap.put("QPID_PORT", "connector.port");
- envVarMap.put("QPID_ENABLEDIRECTBUFFERS", "advanced.enableDirectBuffers");
envVarMap.put("QPID_SSLPORT", "connector.ssl.port");
envVarMap.put("QPID_JMXPORT_REGISTRYSERVER", MGMT_JMXPORT_REGISTRYSERVER);
envVarMap.put("QPID_JMXPORT_CONNECTORSERVER", MGMT_JMXPORT_CONNECTORSERVER);
@@ -108,7 +116,6 @@ public class ServerConfiguration extends
envVarMap.put("QPID_SOCKETRECEIVEBUFFER", "connector.socketReceiveBuffer");
envVarMap.put("QPID_SOCKETWRITEBUFFER", "connector.socketWriteBuffer");
envVarMap.put("QPID_TCPNODELAY", "connector.tcpNoDelay");
- envVarMap.put("QPID_ENABLEPOOLEDALLOCATOR", "advanced.enablePooledAllocator");
envVarMap.put("QPID_STATUS-UPDATES", "status-updates");
}
@@ -177,7 +184,7 @@ public class ServerConfiguration extends
*/
public ServerConfiguration(Configuration conf)
{
- _configuration = conf;
+ setConfig(conf);
}
/**
@@ -197,8 +204,8 @@ public class ServerConfiguration extends
*/
public void initialise() throws ConfigurationException
{
- setConfiguration("", _configuration);
- setupVirtualHosts(_configuration);
+ setConfiguration("", getConfig());
+ setupVirtualHosts(getConfig());
}
public String[] getElementsProcessed()
@@ -256,6 +263,13 @@ public class ServerConfiguration extends
+ (_configFile == null ? "" : " Configuration file : " + _configFile));
}
}
+
+ // QPID-3739 certType was a misleading name.
+ if (contains("connector.ssl.certType"))
+ {
+ _logger.warn("Validation warning: connector/ssl/certType is deprecated and must be replaced by connector/ssl/keyManagerFactoryAlgorithm"
+ + (_configFile == null ? "" : " Configuration file : " + _configFile));
+ }
}
/*
@@ -303,7 +317,7 @@ public class ServerConfiguration extends
// save the default virtualhost name
String defaultVirtualHost = vhostConfiguration.getString("default");
- _configuration.setProperty("virtualhosts.default", defaultVirtualHost);
+ getConfig().setProperty("virtualhosts.default", defaultVirtualHost);
}
}
@@ -472,7 +486,7 @@ public class ServerConfiguration extends
{
VirtualHost vhost = vhostRegistry.getVirtualHost(hostName);
Configuration vhostConfig = newVhosts.subset("virtualhost." + hostName);
- vhost.getConfiguration().setConfiguration("virtualhosts.virtualhost", vhostConfig); // XXX
+ vhost.getConfiguration().setConfiguration("virtualhosts.virtualhost", vhostConfig);
vhost.getSecurityManager().configureGlobalPlugins(this);
vhost.getSecurityManager().configureHostPlugins(vhost.getConfiguration());
}
@@ -608,11 +622,6 @@ public class ServerConfiguration extends
return getDoubleValue("heartbeat.timeoutFactor", 2.0);
}
- public int getDeliveryPoolSize()
- {
- return getIntValue("delivery.poolsize");
- }
-
public long getMaximumMessageAge()
{
return getLongValue("maximumMessageAge");
@@ -698,11 +707,6 @@ public class ServerConfiguration extends
return getBooleanValue("connector.tcpNoDelay", true);
}
- public boolean getEnableExecutorPool()
- {
- return getBooleanValue("advanced.filterchain[@enableExecutorPool]");
- }
-
public boolean getEnableSSL()
{
return getBooleanValue("connector.ssl.enabled");
@@ -730,9 +734,12 @@ public class ServerConfiguration extends
return getStringValue("connector.ssl.keyStorePassword", fallback);
}
- public String getConnectorCertType()
+ public String getConnectorKeyManagerFactoryAlgorithm()
{
- return getStringValue("connector.ssl.certType", "SunX509");
+ final String systemFallback = KeyManagerFactory.getDefaultAlgorithm();
+ // deprecated, pre-0.17 brokers supported this name.
+ final String fallback = getStringValue("connector.ssl.certType", systemFallback);
+ return getStringValue("connector.ssl.keyManagerFactoryAlgorithm", fallback);
}
public String getDefaultVirtualHost()
@@ -836,4 +843,33 @@ public class ServerConfiguration extends
return getConfig().getString("deadLetterQueueSuffix", AMQQueueFactory.DEFAULT_DLQ_NAME_SUFFIX);
}
+ public boolean isAmqp010enabled()
+ {
+ return getConfig().getBoolean(CONNECTOR_AMQP010ENABLED, true);
+ }
+
+ public boolean isAmqp091enabled()
+ {
+ return getConfig().getBoolean(CONNECTOR_AMQP091ENABLED, true);
+ }
+
+ public boolean isAmqp09enabled()
+ {
+ return getConfig().getBoolean(CONNECTOR_AMQP09ENABLED, true);
+ }
+
+ public boolean isAmqp08enabled()
+ {
+ return getConfig().getBoolean(CONNECTOR_AMQP08ENABLED, true);
+ }
+
+ /**
+ * Returns the configured default reply to an unsupported AMQP protocol initiation, or null if there is none
+ */
+ public AmqpProtocolVersion getDefaultSupportedProtocolReply()
+ {
+ String reply = getConfig().getString(CONNECTOR_AMQP_SUPPORTED_REPLY, null);
+
+ return reply == null ? null : AmqpProtocolVersion.valueOf(reply);
+ }
}
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/ServerNetworkTransportConfiguration.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/ServerNetworkTransportConfiguration.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/ServerNetworkTransportConfiguration.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/ServerNetworkTransportConfiguration.java Sat Mar 10 19:22:10 2012
@@ -19,22 +19,21 @@
*/
package org.apache.qpid.server.configuration;
+import java.net.InetSocketAddress;
import org.apache.qpid.transport.NetworkTransportConfiguration;
public class ServerNetworkTransportConfiguration implements NetworkTransportConfiguration
{
private final ServerConfiguration _serverConfig;
- private final int _port;
- private final String _host;
private final String _transport;
+ private InetSocketAddress _address;
public ServerNetworkTransportConfiguration(final ServerConfiguration serverConfig,
- final int port, final String host,
+ final InetSocketAddress address,
final String transport)
{
_serverConfig = serverConfig;
- _port = port;
- _host = host;
+ _address = address;
_transport = transport;
}
@@ -55,12 +54,12 @@ public class ServerNetworkTransportConfi
public Integer getPort()
{
- return _port;
+ return _address.getPort();
}
public String getHost()
{
- return _host;
+ return _address.getHostName();
}
public String getTransport()
@@ -72,4 +71,9 @@ public class ServerNetworkTransportConfi
{
return _serverConfig.getConnectorProcessors();
}
+
+ public InetSocketAddress getAddress()
+ {
+ return _address;
+ }
}
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/SessionConfigType.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/SessionConfigType.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/SessionConfigType.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/SessionConfigType.java Sat Mar 10 19:22:10 2012
@@ -21,9 +21,10 @@
package org.apache.qpid.server.configuration;
-import org.apache.qpid.server.exchange.ExchangeType;
-
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
public final class SessionConfigType extends ConfigObjectType<SessionConfigType, SessionConfig>
{
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/SubscriptionConfigType.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/SubscriptionConfigType.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/SubscriptionConfigType.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/SubscriptionConfigType.java Sat Mar 10 19:22:10 2012
@@ -22,7 +22,11 @@
package org.apache.qpid.server.configuration;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
public final class SubscriptionConfigType extends ConfigObjectType<SubscriptionConfigType, SubscriptionConfig>
{
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/SystemConfigImpl.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/SystemConfigImpl.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/SystemConfigImpl.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/SystemConfigImpl.java Sat Mar 10 19:22:10 2012
@@ -21,11 +21,11 @@
package org.apache.qpid.server.configuration;
-import java.util.UUID;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
import java.net.InetAddress;
import java.net.UnknownHostException;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
public class SystemConfigImpl implements SystemConfig
{
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/SystemConfigType.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/SystemConfigType.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/SystemConfigType.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/SystemConfigType.java Sat Mar 10 19:22:10 2012
@@ -21,7 +21,11 @@
package org.apache.qpid.server.configuration;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.UUID;
public final class SystemConfigType extends ConfigObjectType<SystemConfigType, SystemConfig>
{
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/TopicConfig.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/TopicConfig.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/TopicConfig.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/TopicConfig.java Sat Mar 10 19:22:10 2012
@@ -22,13 +22,14 @@ package org.apache.qpid.server.configura
import org.apache.commons.configuration.ConfigurationException;
import org.apache.commons.configuration.PropertiesConfiguration;
+
import org.apache.qpid.server.configuration.plugins.ConfigurationPlugin;
public class TopicConfig extends ConfigurationPlugin
{
public TopicConfig()
{
- _configuration = new PropertiesConfiguration();
+ setConfig(new PropertiesConfiguration());
}
@Override
@@ -50,7 +51,7 @@ public class TopicConfig extends Configu
public void validateConfiguration() throws ConfigurationException
{
- if (_configuration.isEmpty())
+ if (getConfig().isEmpty())
{
throw new ConfigurationException("Topic section cannot be empty.");
}
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/TopicConfiguration.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/TopicConfiguration.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/TopicConfiguration.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/TopicConfiguration.java Sat Mar 10 19:22:10 2012
@@ -22,6 +22,7 @@ package org.apache.qpid.server.configura
import org.apache.commons.configuration.Configuration;
import org.apache.commons.configuration.ConfigurationException;
+
import org.apache.qpid.server.binding.Binding;
import org.apache.qpid.server.configuration.plugins.ConfigurationPlugin;
import org.apache.qpid.server.configuration.plugins.ConfigurationPluginFactory;
@@ -56,8 +57,8 @@ public class TopicConfiguration extends
}
}
- Map<String, TopicConfig> _topics = new HashMap<String, TopicConfig>();
- Map<String, Map<String, TopicConfig>> _subscriptions = new HashMap<String, Map<String, TopicConfig>>();
+ private Map<String, TopicConfig> _topics = new HashMap<String, TopicConfig>();
+ private Map<String, Map<String, TopicConfig>> _subscriptions = new HashMap<String, Map<String, TopicConfig>>();
public String[] getElementsProcessed()
{
@@ -67,17 +68,17 @@ public class TopicConfiguration extends
@Override
public void validateConfiguration() throws ConfigurationException
{
- if (_configuration.isEmpty())
+ if (getConfig().isEmpty())
{
throw new ConfigurationException("Topics section cannot be empty.");
}
- int topics = _configuration.getList("topic.name").size() +
- _configuration.getList("topic.subscriptionName").size();
+ int topics = getConfig().getList("topic.name").size() +
+ getConfig().getList("topic.subscriptionName").size();
for (int index = 0; index < topics; index++)
{
- Configuration topicSubset = _configuration.subset("topic(" + index + ")");
+ Configuration topicSubset = getConfig().subset("topic(" + index + ")");
// This will occur when we have a subscriptionName that is bound to a
// topic.
@@ -90,8 +91,8 @@ public class TopicConfiguration extends
topic.setConfiguration(VIRTUALHOSTS_VIRTUALHOST_TOPICS + ".topic", topicSubset );
- String name = _configuration.getString("topic(" + index + ").name");
- String subscriptionName = _configuration.getString("topic(" + index + ").subscriptionName");
+ String name = getConfig().getString("topic(" + index + ").name");
+ String subscriptionName = getConfig().getString("topic(" + index + ").subscriptionName");
// Record config if subscriptionName is there
if (subscriptionName != null)
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/VirtualHostConfigType.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/VirtualHostConfigType.java?rev=1299257&r1=1299256&r2=1299257&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/VirtualHostConfigType.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/VirtualHostConfigType.java Sat Mar 10 19:22:10 2012
@@ -21,7 +21,10 @@
package org.apache.qpid.server.configuration;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
public class VirtualHostConfigType extends ConfigObjectType<VirtualHostConfigType, VirtualHostConfig>
{
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org