You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by or...@apache.org on 2013/02/19 10:35:33 UTC
svn commit: r1447646 [8/16] - in /qpid/trunk/qpid/java: ./ bdbstore/
bdbstore/jmx/src/main/java/org/apache/qpid/server/store/berkeleydb/jmx/
bdbstore/jmx/src/test/java/org/apache/qpid/server/store/berkeleydb/
bdbstore/src/main/java/org/apache/qpid/serv...
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/model/adapter/SessionAdapter.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/model/adapter/SessionAdapter.java?rev=1447646&r1=1447645&r2=1447646&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/model/adapter/SessionAdapter.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/model/adapter/SessionAdapter.java Tue Feb 19 09:35:28 2013
@@ -34,6 +34,7 @@ import org.apache.qpid.server.model.Stat
import org.apache.qpid.server.model.Statistics;
import org.apache.qpid.server.model.Consumer;
import org.apache.qpid.server.model.UUIDGenerator;
+import org.apache.qpid.server.configuration.updater.TaskExecutor;
import org.apache.qpid.server.protocol.AMQSessionModel;
final class SessionAdapter extends AbstractAdapter implements Session
@@ -44,9 +45,9 @@ final class SessionAdapter extends Abstr
private AMQSessionModel _session;
private SessionStatistics _statistics;
- public SessionAdapter(final AMQSessionModel session)
+ public SessionAdapter(final AMQSessionModel session, TaskExecutor taskExecutor)
{
- super(UUIDGenerator.generateRandomUUID());
+ super(UUIDGenerator.generateRandomUUID(), taskExecutor);
_session = session;
_statistics = new SessionStatistics();
}
@@ -141,13 +142,6 @@ final class SessionAdapter extends Abstr
return super.getAttribute(name); //TODO - Implement
}
- @Override
- public Object setAttribute(String name, Object expected, Object desired)
- throws IllegalStateException, AccessControlException, IllegalArgumentException
- {
- return super.setAttribute(name, expected, desired); //TODO - Implement
- }
-
public Statistics getStatistics()
{
return _statistics;
@@ -237,4 +231,11 @@ final class SessionAdapter extends Abstr
return null; // TODO - Implement
}
}
+
+ @Override
+ protected boolean setState(State currentState, State desiredState)
+ {
+ // TODO : add state management
+ return false;
+ }
}
Copied: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/model/adapter/TrustStoreAdapter.java (from r1447519, qpid/trunk/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/actors/BaseConnectionActorTestCase.java)
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/model/adapter/TrustStoreAdapter.java?p2=qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/model/adapter/TrustStoreAdapter.java&p1=qpid/trunk/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/actors/BaseConnectionActorTestCase.java&r1=1447519&r2=1447646&rev=1447646&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/actors/BaseConnectionActorTestCase.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/model/adapter/TrustStoreAdapter.java Tue Feb 19 09:35:28 2013
@@ -18,16 +18,26 @@
* under the License.
*
*/
-package org.apache.qpid.server.logging.actors;
+package org.apache.qpid.server.model.adapter;
-public class BaseConnectionActorTestCase extends BaseActorTestCase
+import java.util.Collection;
+import java.util.Map;
+import java.util.UUID;
+
+import org.apache.qpid.server.model.Broker;
+import org.apache.qpid.server.model.TrustStore;
+
+public class TrustStoreAdapter extends AbstractKeyStoreAdapter implements TrustStore
{
+ public TrustStoreAdapter(UUID id, Broker broker, Map<String, Object> attributes)
+ {
+ super(id, broker, attributes);
+ }
@Override
- public void createBroker() throws Exception
+ public Collection<String> getAttributeNames()
{
- super.createBroker();
-
- _amqpActor = new AMQPConnectionActor(getSession(), _rootLogger);
+ return AVAILABLE_ATTRIBUTES;
}
+
}
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/model/adapter/VirtualHostAdapter.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/model/adapter/VirtualHostAdapter.java?rev=1447646&r1=1447645&r2=1447646&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/model/adapter/VirtualHostAdapter.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/model/adapter/VirtualHostAdapter.java Tue Feb 19 09:35:28 2013
@@ -20,6 +20,7 @@
*/
package org.apache.qpid.server.model.adapter;
+import java.io.File;
import java.security.AccessControlException;
import java.security.Principal;
import java.util.ArrayList;
@@ -31,16 +32,27 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.UUID;
+
+import org.apache.commons.configuration.CompositeConfiguration;
+import org.apache.commons.configuration.ConfigurationException;
+import org.apache.commons.configuration.PropertiesConfiguration;
+import org.apache.commons.configuration.SystemConfiguration;
import org.apache.qpid.AMQException;
import org.apache.qpid.framing.FieldTable;
+import org.apache.qpid.server.configuration.IllegalConfigurationException;
+import org.apache.qpid.server.configuration.VirtualHostConfiguration;
+import org.apache.qpid.server.configuration.XmlConfigurationUtilities.MyConfiguration;
import org.apache.qpid.server.connection.IConnectionRegistry;
import org.apache.qpid.server.exchange.ExchangeRegistry;
import org.apache.qpid.server.message.ServerMessage;
+import org.apache.qpid.server.model.Broker;
import org.apache.qpid.server.model.ConfiguredObject;
import org.apache.qpid.server.model.Connection;
import org.apache.qpid.server.model.Exchange;
import org.apache.qpid.server.model.LifetimePolicy;
import org.apache.qpid.server.model.Port;
+import org.apache.qpid.server.model.Protocol;
import org.apache.qpid.server.model.Queue;
import org.apache.qpid.server.model.QueueType;
import org.apache.qpid.server.model.State;
@@ -48,6 +60,7 @@ import org.apache.qpid.server.model.Stat
import org.apache.qpid.server.model.UUIDGenerator;
import org.apache.qpid.server.model.VirtualHost;
import org.apache.qpid.server.model.VirtualHostAlias;
+import org.apache.qpid.server.configuration.updater.TaskExecutor;
import org.apache.qpid.server.plugin.ExchangeType;
import org.apache.qpid.server.protocol.AMQConnectionModel;
import org.apache.qpid.server.queue.AMQQueue;
@@ -56,16 +69,28 @@ import org.apache.qpid.server.queue.Queu
import org.apache.qpid.server.queue.QueueRegistry;
import org.apache.qpid.server.security.SecurityManager;
import org.apache.qpid.server.security.auth.AuthenticatedPrincipal;
+import org.apache.qpid.server.stats.StatisticsGatherer;
import org.apache.qpid.server.store.MessageStore;
import org.apache.qpid.server.txn.LocalTransaction;
import org.apache.qpid.server.txn.ServerTransaction;
+import org.apache.qpid.server.util.MapValueConverter;
+import org.apache.qpid.server.virtualhost.VirtualHostImpl;
+import org.apache.qpid.server.virtualhost.VirtualHostRegistry;
-final class VirtualHostAdapter extends AbstractAdapter implements VirtualHost, ExchangeRegistry.RegistryChangeListener,
+public final class VirtualHostAdapter extends AbstractAdapter implements VirtualHost, ExchangeRegistry.RegistryChangeListener,
QueueRegistry.RegistryChangeListener,
IConnectionRegistry.RegistryChangeListener
{
- private final org.apache.qpid.server.virtualhost.VirtualHost _virtualHost;
+ @SuppressWarnings("serial")
+ public static final Map<String, Class<?>> ATTRIBUTE_TYPES = Collections.unmodifiableMap(new HashMap<String, Class<?>>(){{
+ put(NAME, String.class);
+ put(STORE_PATH, String.class);
+ put(STORE_TYPE, String.class);
+ put(CONFIG_PATH, String.class);
+ }});
+
+ private org.apache.qpid.server.virtualhost.VirtualHost _virtualHost;
private final Map<AMQConnectionModel, ConnectionAdapter> _connectionAdapters =
new HashMap<AMQConnectionModel, ConnectionAdapter>();
@@ -75,37 +100,52 @@ final class VirtualHostAdapter extends A
private final Map<org.apache.qpid.server.exchange.Exchange, ExchangeAdapter> _exchangeAdapters =
new HashMap<org.apache.qpid.server.exchange.Exchange, ExchangeAdapter>();
-
- private final StatisticsAdapter _statistics;
-
- private final BrokerAdapter _broker;
-
+ private StatisticsAdapter _statistics;
+ private final Broker _broker;
private final List<VirtualHostAlias> _aliases = new ArrayList<VirtualHostAlias>();
+ private StatisticsGatherer _brokerStatisticsGatherer;
-
- VirtualHostAdapter(BrokerAdapter brokerAdapter,
- final org.apache.qpid.server.virtualhost.VirtualHost virtualHost)
+ public VirtualHostAdapter(UUID id, Map<String, Object> attributes, Broker broker, StatisticsGatherer brokerStatisticsGatherer, TaskExecutor taskExecutor)
{
- super(virtualHost.getId());
- _broker = brokerAdapter;
- _virtualHost = virtualHost;
- _statistics = new VirtualHostStatisticsAdapter(virtualHost);
- virtualHost.getQueueRegistry().addRegistryChangeListener(this);
- populateQueues();
- virtualHost.getExchangeRegistry().addRegistryChangeListener(this);
- populateExchanges();
- virtualHost.getConnectionRegistry().addRegistryChangeListener(this);
- populateConnections();
-
+ super(id, null, MapValueConverter.convert(attributes, ATTRIBUTE_TYPES), taskExecutor);
+ validateAttributes();
+ _broker = broker;
+ _brokerStatisticsGatherer = brokerStatisticsGatherer;
+ addParent(Broker.class, broker);
+ }
+ private void validateAttributes()
+ {
+ String name = getName();
+ if (name == null || "".equals(name.trim()))
+ {
+ throw new IllegalConfigurationException("Virtual host name must be specified");
+ }
- for(Port port :_broker.getPorts())
+ String configurationFile = (String) getAttribute(CONFIG_PATH);
+ String storePath = (String) getAttribute(STORE_PATH);
+ String storeType = (String) getAttribute(STORE_TYPE);
+ boolean invalidAttributes = false;
+ if (configurationFile == null)
{
- _aliases.add(new VirtualHostAliasAdapter(this, port));
+ if (storePath == null || storeType == null)
+ {
+ invalidAttributes = true;
+ }
+ }
+ else
+ {
+ if (storePath != null || storeType != null)
+ {
+ invalidAttributes = true;
+ }
+ }
+ if (invalidAttributes)
+ {
+ throw new IllegalConfigurationException("Please specify either the 'configPath' attribute or both 'storePath' and 'storeType' attributes");
}
}
-
private void populateExchanges()
{
Collection<org.apache.qpid.server.exchange.Exchange> actualExchanges =
@@ -127,37 +167,22 @@ final class VirtualHostAdapter extends A
private void populateQueues()
{
Collection<AMQQueue> actualQueues = _virtualHost.getQueueRegistry().getQueues();
-
- synchronized(_queueAdapters)
- {
- for(AMQQueue queue : actualQueues)
- {
- if(!_queueAdapters.containsKey(queue))
- {
- _queueAdapters.put(queue, new QueueAdapter(this,queue));
- }
- }
- }
- }
-
- private void populateConnections()
- {
-
- List<AMQConnectionModel> actualConnections = _virtualHost.getConnectionRegistry().getConnections();
-
- synchronized(_connectionAdapters)
+ if ( actualQueues != null )
{
- for(AMQConnectionModel conn : actualConnections)
+ synchronized(_queueAdapters)
{
- if(!_connectionAdapters.containsKey(conn))
+ for(AMQQueue queue : actualQueues)
{
- _connectionAdapters.put(conn, new ConnectionAdapter(conn));
+ if(!_queueAdapters.containsKey(queue))
+ {
+ _queueAdapters.put(queue, new QueueAdapter(this, queue));
+ }
}
}
}
-
}
+ @Override
public String getReplicationGroupName()
{
return null; //TODO
@@ -199,12 +224,12 @@ final class VirtualHostAdapter extends A
{
attributes = new HashMap<String, Object>(attributes);
- String name = getStringAttribute(Exchange.NAME, attributes, null);
- State state = getEnumAttribute(State.class, Exchange.STATE, attributes, State.ACTIVE);
- boolean durable = getBooleanAttribute(Exchange.DURABLE, attributes, false);
- LifetimePolicy lifetime = getEnumAttribute(LifetimePolicy.class, Exchange.LIFETIME_POLICY, attributes, LifetimePolicy.PERMANENT);
- String type = getStringAttribute(Exchange.TYPE, attributes, null);
- long ttl = getLongAttribute(Exchange.TIME_TO_LIVE, attributes, 0l);
+ String name = MapValueConverter.getStringAttribute(Exchange.NAME, attributes, null);
+ State state = MapValueConverter.getEnumAttribute(State.class, Exchange.STATE, attributes, State.ACTIVE);
+ boolean durable = MapValueConverter.getBooleanAttribute(Exchange.DURABLE, attributes, false);
+ LifetimePolicy lifetime = MapValueConverter.getEnumAttribute(LifetimePolicy.class, Exchange.LIFETIME_POLICY, attributes, LifetimePolicy.PERMANENT);
+ String type = MapValueConverter.getStringAttribute(Exchange.TYPE, attributes, null);
+ long ttl = MapValueConverter.getLongAttribute(Exchange.TIME_TO_LIVE, attributes, 0l);
attributes.remove(Exchange.NAME);
attributes.remove(Exchange.STATE);
@@ -267,7 +292,7 @@ final class VirtualHostAdapter extends A
if (attributes.containsKey(Queue.TYPE))
{
- String typeAttribute = getStringAttribute(Queue.TYPE, attributes, null);
+ String typeAttribute = MapValueConverter.getStringAttribute(Queue.TYPE, attributes, null);
QueueType queueType = null;
try
{
@@ -290,12 +315,12 @@ final class VirtualHostAdapter extends A
throw new IllegalArgumentException("Sort key is not specified for sorted queue");
}
}
- String name = getStringAttribute(Queue.NAME, attributes, null);
- State state = getEnumAttribute(State.class, Queue.STATE, attributes, State.ACTIVE);
- boolean durable = getBooleanAttribute(Queue.DURABLE, attributes, false);
- LifetimePolicy lifetime = getEnumAttribute(LifetimePolicy.class, Queue.LIFETIME_POLICY, attributes, LifetimePolicy.PERMANENT);
- long ttl = getLongAttribute(Queue.TIME_TO_LIVE, attributes, 0l);
- boolean exclusive= getBooleanAttribute(Queue.EXCLUSIVE, attributes, false);
+ String name = MapValueConverter.getStringAttribute(Queue.NAME, attributes, null);
+ State state = MapValueConverter.getEnumAttribute(State.class, Queue.STATE, attributes, State.ACTIVE);
+ boolean durable = MapValueConverter.getBooleanAttribute(Queue.DURABLE, attributes, false);
+ LifetimePolicy lifetime = MapValueConverter.getEnumAttribute(LifetimePolicy.class, Queue.LIFETIME_POLICY, attributes, LifetimePolicy.PERMANENT);
+ long ttl = MapValueConverter.getLongAttribute(Queue.TIME_TO_LIVE, attributes, 0l);
+ boolean exclusive= MapValueConverter.getBooleanAttribute(Queue.EXCLUSIVE, attributes, false);
attributes.remove(Queue.NAME);
attributes.remove(Queue.STATE);
@@ -370,7 +395,7 @@ final class VirtualHostAdapter extends A
public String getName()
{
- return _virtualHost.getName();
+ return (String)getAttribute(NAME);
}
public String setName(final String currentName, final String desiredName)
@@ -381,7 +406,28 @@ final class VirtualHostAdapter extends A
public State getActualState()
{
- return getDesiredState();
+ if (_virtualHost == null)
+ {
+ return State.INITIALISING;
+ }
+ else
+ {
+ org.apache.qpid.server.virtualhost.State implementationState = _virtualHost.getState();
+ switch(implementationState)
+ {
+ case INITIALISING:
+ return State.INITIALISING;
+ case ACTIVE:
+ return State.ACTIVE;
+ case PASSIVE:
+ return State.QUIESCED;
+ case STOPPED:
+ return State.STOPPED;
+ default:
+ // unexpected state
+ return null;
+ }
+ }
}
public boolean isDurable()
@@ -448,7 +494,7 @@ final class VirtualHostAdapter extends A
}
@Override
- public <C extends ConfiguredObject> C createChild(Class<C> childClass, Map<String, Object> attributes, ConfiguredObject... otherParents)
+ public <C extends ConfiguredObject> C addChild(Class<C> childClass, Map<String, Object> attributes, ConfiguredObject... otherParents)
{
if(childClass == Exchange.class)
{
@@ -548,7 +594,7 @@ final class VirtualHostAdapter extends A
{
if(!_connectionAdapters.containsKey(connection))
{
- adapter = new ConnectionAdapter(connection);
+ adapter = new ConnectionAdapter(connection, getTaskExecutor());
_connectionAdapters.put(connection, adapter);
}
@@ -709,10 +755,6 @@ final class VirtualHostAdapter extends A
{
return getId();
}
- else if(NAME.equals(name))
- {
- return getName();
- }
else if(STATE.equals(name))
{
return State.ACTIVE;
@@ -737,10 +779,19 @@ final class VirtualHostAdapter extends A
{
// TODO
}
- else if(SUPPORTED_EXCHANGE_TYPES.equals(name))
+ else if (_virtualHost != null)
+ {
+ return getAttributeFromVirtualHostImplementation(name);
+ }
+ return super.getAttribute(name);
+ }
+
+ private Object getAttributeFromVirtualHostImplementation(String name)
+ {
+ if(SUPPORTED_EXCHANGE_TYPES.equals(name))
{
List<String> types = new ArrayList<String>();
- for(ExchangeType type : _virtualHost.getExchangeFactory().getRegisteredTypes())
+ for(@SuppressWarnings("rawtypes") ExchangeType type : _virtualHost.getExchangeFactory().getRegisteredTypes())
{
types.add(type.getName().asString());
}
@@ -774,9 +825,9 @@ final class VirtualHostAdapter extends A
{
return _virtualHost.getMessageStore().getStoreType();
}
- else if(STORE_CONFIGURATION.equals(name))
+ else if(STORE_PATH.equals(name))
{
- // TODO
+ return _virtualHost.getMessageStore().getStoreLocation();
}
else if(STORE_TRANSACTION_IDLE_TIMEOUT_CLOSE.equals(name))
{
@@ -818,13 +869,6 @@ final class VirtualHostAdapter extends A
}
@Override
- public Object setAttribute(String name, Object expected, Object desired)
- throws IllegalStateException, AccessControlException, IllegalArgumentException
- {
- return super.setAttribute(name, expected, desired); //TODO - Implement
- }
-
- @Override
public Collection<String> getAttributeNames()
{
return AVAILABLE_ATTRIBUTES;
@@ -885,4 +929,111 @@ final class VirtualHostAdapter extends A
}
}
+
+ @Override
+ protected boolean setState(State currentState, State desiredState)
+ {
+ if (desiredState == State.ACTIVE)
+ {
+ activate();
+ return true;
+ }
+ else if (desiredState == State.STOPPED)
+ {
+ if (_virtualHost != null)
+ {
+ try
+ {
+ _virtualHost.close();
+ }
+ finally
+ {
+ _broker.getVirtualHostRegistry().unregisterVirtualHost(_virtualHost);
+ }
+ }
+ return true;
+ }
+ else if (desiredState == State.DELETED)
+ {
+ //TODO: add ACL check to authorize the operation
+ if (_virtualHost != null && _virtualHost.getState() == org.apache.qpid.server.virtualhost.State.ACTIVE)
+ {
+ setDesiredState(currentState, State.STOPPED);
+ }
+ return true;
+ }
+ return false;
+ }
+
+ private void activate()
+ {
+ VirtualHostRegistry virtualHostRegistry = _broker.getVirtualHostRegistry();
+ String virtualHostName = getName();
+ try
+ {
+ VirtualHostConfiguration configuration = createVirtualHostConfiguration(virtualHostName);
+ _virtualHost = new VirtualHostImpl(_broker.getVirtualHostRegistry(), _brokerStatisticsGatherer, _broker.getSecurityManager(), configuration);
+ }
+ catch (Exception e)
+ {
+ throw new RuntimeException("Failed to create virtual host " + virtualHostName, e);
+ }
+
+ virtualHostRegistry.registerVirtualHost(_virtualHost);
+
+ _statistics = new VirtualHostStatisticsAdapter(_virtualHost);
+ _virtualHost.getQueueRegistry().addRegistryChangeListener(this);
+ populateQueues();
+ _virtualHost.getExchangeRegistry().addRegistryChangeListener(this);
+ populateExchanges();
+ _virtualHost.getConnectionRegistry().addRegistryChangeListener(this);
+
+ synchronized(_aliases)
+ {
+ for(Port port :_broker.getPorts())
+ {
+ if (Protocol.hasAmqpProtocol(port.getProtocols()))
+ {
+ _aliases.add(new VirtualHostAliasAdapter(this, port));
+ }
+ }
+ }
+ }
+
+ private VirtualHostConfiguration createVirtualHostConfiguration(String virtualHostName) throws ConfigurationException
+ {
+ VirtualHostConfiguration configuration;
+ String configurationFile = (String)getAttribute(CONFIG_PATH);
+ if (configurationFile == null)
+ {
+ final MyConfiguration basicConfiguration = new MyConfiguration();
+ PropertiesConfiguration config = new PropertiesConfiguration();
+ config.addProperty("store.type", (String)getAttribute(STORE_TYPE));
+ config.addProperty("store.environment-path", (String)getAttribute(STORE_PATH));
+ basicConfiguration.addConfiguration(config);
+
+ CompositeConfiguration compositeConfiguration = new CompositeConfiguration();
+ compositeConfiguration.addConfiguration(new SystemConfiguration());
+ compositeConfiguration.addConfiguration(basicConfiguration);
+ configuration = new VirtualHostConfiguration(virtualHostName, compositeConfiguration , _broker);
+ }
+ else
+ {
+ configuration = new VirtualHostConfiguration(virtualHostName, new File(configurationFile) , _broker);
+ }
+ return configuration;
+ }
+
+ @Override
+ public SecurityManager getSecurityManager()
+ {
+ return _virtualHost.getSecurityManager();
+ }
+
+ @Override
+ public MessageStore getMessageStore()
+ {
+ return _virtualHost.getMessageStore();
+ }
+
}
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/model/adapter/VirtualHostAliasAdapter.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/model/adapter/VirtualHostAliasAdapter.java?rev=1447646&r1=1447645&r2=1447646&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/model/adapter/VirtualHostAliasAdapter.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/model/adapter/VirtualHostAliasAdapter.java Tue Feb 19 09:35:28 2013
@@ -43,7 +43,7 @@ public class VirtualHostAliasAdapter ex
public VirtualHostAliasAdapter(VirtualHostAdapter virtualHostAdapter, Port port)
{
- super(UUIDGenerator.generateVhostAliasUUID(virtualHostAdapter.getName(), port.getName()));
+ super(UUIDGenerator.generateVhostAliasUUID(virtualHostAdapter.getName(), port.getName()), virtualHostAdapter.getTaskExecutor());
_vhost = virtualHostAdapter;
_port = port;
}
@@ -140,4 +140,11 @@ public class VirtualHostAliasAdapter ex
{
throw new UnsupportedOperationException();
}
+
+ @Override
+ protected boolean setState(State currentState, State desiredState)
+ {
+ // TODO: state is not supported at the moment
+ return false;
+ }
}
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/plugin/AccessControlFactory.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/plugin/AccessControlFactory.java?rev=1447646&r1=1447645&r2=1447646&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/plugin/AccessControlFactory.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/plugin/AccessControlFactory.java Tue Feb 19 09:35:28 2013
@@ -18,10 +18,11 @@
*/
package org.apache.qpid.server.plugin;
-import org.apache.commons.configuration.Configuration;
+import java.util.Map;
+
import org.apache.qpid.server.security.AccessControl;
public interface AccessControlFactory
{
- AccessControl createInstance(Configuration securityConfig);
+ AccessControl createInstance(Map<String, Object> attributes);
}
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/plugin/AuthenticationManagerFactory.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/plugin/AuthenticationManagerFactory.java?rev=1447646&r1=1447645&r2=1447646&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/plugin/AuthenticationManagerFactory.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/plugin/AuthenticationManagerFactory.java Tue Feb 19 09:35:28 2013
@@ -18,11 +18,14 @@
*/
package org.apache.qpid.server.plugin;
-import org.apache.commons.configuration.Configuration;
+import java.util.Map;
+
import org.apache.qpid.server.security.auth.manager.AuthenticationManager;
public interface AuthenticationManagerFactory
{
- AuthenticationManager createInstance(Configuration securityConfiguration);
+ public static final String ATTRIBUTE_TYPE = "authenticationProviderType";
+
+ AuthenticationManager createInstance(Map<String, Object> attributes);
}
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/plugin/GroupManagerFactory.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/plugin/GroupManagerFactory.java?rev=1447646&r1=1447645&r2=1447646&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/plugin/GroupManagerFactory.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/plugin/GroupManagerFactory.java Tue Feb 19 09:35:28 2013
@@ -18,10 +18,11 @@
*/
package org.apache.qpid.server.plugin;
-import org.apache.commons.configuration.Configuration;
+import java.util.Map;
+
import org.apache.qpid.server.security.group.GroupManager;
public interface GroupManagerFactory
{
- GroupManager createInstance(Configuration securityConfiguration);
+ GroupManager createInstance(Map<String, Object> attributes);
}
Copied: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/plugin/PluginFactory.java (from r1447519, qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/plugin/ManagementFactory.java)
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/plugin/PluginFactory.java?p2=qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/plugin/PluginFactory.java&p1=qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/plugin/ManagementFactory.java&r1=1447519&r2=1447646&rev=1447646&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/plugin/ManagementFactory.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/plugin/PluginFactory.java Tue Feb 19 09:35:28 2013
@@ -18,11 +18,15 @@
*/
package org.apache.qpid.server.plugin;
-import org.apache.qpid.server.configuration.ServerConfiguration;
-import org.apache.qpid.server.management.plugin.ManagementPlugin;
+import java.util.Map;
+import java.util.UUID;
+
import org.apache.qpid.server.model.Broker;
+import org.apache.qpid.server.model.Plugin;
-public interface ManagementFactory
+public interface PluginFactory
{
- ManagementPlugin createInstance(ServerConfiguration configuration, Broker broker);
+ static final String PLUGIN_TYPE = "pluginType";
+
+ Plugin createInstance(UUID id, Map<String, Object> attributes, Broker broker);
}
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolEngine.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolEngine.java?rev=1447646&r1=1447645&r2=1447646&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolEngine.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolEngine.java Tue Feb 19 09:35:28 2013
@@ -52,6 +52,7 @@ import org.apache.qpid.protocol.AMQMetho
import org.apache.qpid.protocol.AMQMethodListener;
import org.apache.qpid.protocol.ServerProtocolEngine;
import org.apache.qpid.server.AMQChannel;
+import org.apache.qpid.server.configuration.BrokerProperties;
import org.apache.qpid.server.handler.ServerMethodDispatcherImpl;
import org.apache.qpid.server.logging.LogActor;
import org.apache.qpid.server.logging.LogSubject;
@@ -60,10 +61,10 @@ import org.apache.qpid.server.logging.ac
import org.apache.qpid.server.logging.actors.ManagementActor;
import org.apache.qpid.server.logging.messages.ConnectionMessages;
import org.apache.qpid.server.logging.subjects.ConnectionLogSubject;
+import org.apache.qpid.server.model.Broker;
import org.apache.qpid.server.output.ProtocolOutputConverter;
import org.apache.qpid.server.output.ProtocolOutputConverterRegistry;
import org.apache.qpid.server.queue.QueueEntry;
-import org.apache.qpid.server.registry.ApplicationRegistry;
import org.apache.qpid.server.security.auth.AuthenticatedPrincipal;
import org.apache.qpid.server.state.AMQState;
import org.apache.qpid.server.state.AMQStateManager;
@@ -72,7 +73,6 @@ import org.apache.qpid.server.subscripti
import org.apache.qpid.server.subscription.Subscription;
import org.apache.qpid.server.subscription.SubscriptionImpl;
import org.apache.qpid.server.virtualhost.VirtualHost;
-import org.apache.qpid.server.virtualhost.VirtualHostRegistry;
import org.apache.qpid.transport.Sender;
import org.apache.qpid.transport.TransportException;
import org.apache.qpid.transport.network.NetworkConnection;
@@ -112,7 +112,7 @@ public class AMQProtocolEngine implement
private volatile boolean _closed;
// maximum number of channels this session should have
- private long _maxNoOfChannels = ApplicationRegistry.getInstance().getConfiguration().getMaxChannelCount();
+ private long _maxNoOfChannels;
/* AMQP Version for this session */
private ProtocolVersion _protocolVersion = ProtocolVersion.getLatestSupportedVersion();
@@ -152,18 +152,21 @@ public class AMQProtocolEngine implement
private final Lock _receivedLock;
private AtomicLong _lastWriteTime = new AtomicLong(System.currentTimeMillis());
+ private final Broker _broker;
- public AMQProtocolEngine(VirtualHostRegistry virtualHostRegistry, NetworkConnection network, final long connectionId)
+ public AMQProtocolEngine(Broker broker, NetworkConnection network, final long connectionId)
{
+ _broker = broker;
+ _maxNoOfChannels = (Integer)broker.getAttribute(Broker.SESSION_COUNT_LIMIT);
_receivedLock = new ReentrantLock();
- _stateManager = new AMQStateManager(virtualHostRegistry, this);
+ _stateManager = new AMQStateManager(broker, this);
_codecFactory = new AMQCodecFactory(true, this);
setNetworkConnection(network);
_connectionID = connectionId;
- _actor = new AMQPConnectionActor(this, virtualHostRegistry.getApplicationRegistry().getRootMessageLogger());
+ _actor = new AMQPConnectionActor(this, _broker.getRootMessageLogger());
_logSubject = new ConnectionLogSubject(this);
@@ -370,7 +373,7 @@ public class AMQProtocolEngine implement
// This sets the protocol version (and hence framing classes) for this session.
setProtocolVersion(pv);
- String mechanisms = ApplicationRegistry.getInstance().getSubjectCreator(getLocalAddress()).getMechanisms();
+ String mechanisms = _broker.getSubjectCreator(getLocalAddress()).getMechanisms();
String locales = "en_US";
@@ -761,7 +764,7 @@ public class AMQProtocolEngine implement
if (delay > 0)
{
_network.setMaxWriteIdle(delay);
- _network.setMaxReadIdle((int) (ApplicationRegistry.getInstance().getConfiguration().getHeartBeatTimeout() * delay));
+ _network.setMaxReadIdle(BrokerProperties.DEFAULT_HEART_BEAT_TIMEOUT_FACTOR * delay);
}
}
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/MultiVersionProtocolEngine.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/MultiVersionProtocolEngine.java?rev=1447646&r1=1447645&r2=1447646&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/MultiVersionProtocolEngine.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/MultiVersionProtocolEngine.java Tue Feb 19 09:35:28 2013
@@ -28,7 +28,7 @@ import java.util.Set;
import org.apache.log4j.Logger;
import org.apache.qpid.protocol.ServerProtocolEngine;
-import org.apache.qpid.server.registry.IApplicationRegistry;
+import org.apache.qpid.server.model.Broker;
import org.apache.qpid.server.transport.ServerConnection;
import org.apache.qpid.transport.ConnectionDelegate;
import org.apache.qpid.transport.Sender;
@@ -42,24 +42,24 @@ public class MultiVersionProtocolEngine
private Set<AmqpProtocolVersion> _supported;
private String _fqdn;
- private IApplicationRegistry _appRegistry;
+ private final Broker _broker;
private NetworkConnection _network;
private Sender<ByteBuffer> _sender;
private final AmqpProtocolVersion _defaultSupportedReply;
private volatile ServerProtocolEngine _delegate = new SelfDelegateProtocolEngine();
- public MultiVersionProtocolEngine(final IApplicationRegistry appRegistry,
+ public MultiVersionProtocolEngine(final Broker broker,
final Set<AmqpProtocolVersion> supported,
final AmqpProtocolVersion defaultSupportedReply,
final long id,
final NetworkConnection network)
{
- this(appRegistry, supported, defaultSupportedReply, id);
+ this(broker, supported, defaultSupportedReply, id);
setNetworkConnection(network);
}
- public MultiVersionProtocolEngine(final IApplicationRegistry appRegistry,
+ public MultiVersionProtocolEngine(final Broker broker,
final Set<AmqpProtocolVersion> supported,
final AmqpProtocolVersion defaultSupportedReply,
final long id)
@@ -71,7 +71,7 @@ public class MultiVersionProtocolEngine
}
_id = id;
- _appRegistry = appRegistry;
+ _broker = broker;
_supported = supported;
_defaultSupportedReply = defaultSupportedReply;
}
@@ -252,7 +252,7 @@ public class MultiVersionProtocolEngine
public ServerProtocolEngine getProtocolEngine()
{
- return new AMQProtocolEngine(_appRegistry.getVirtualHostRegistry(), _network, _id);
+ return new AMQProtocolEngine(_broker, _network, _id);
}
};
@@ -272,7 +272,7 @@ public class MultiVersionProtocolEngine
public ServerProtocolEngine getProtocolEngine()
{
- return new AMQProtocolEngine(_appRegistry.getVirtualHostRegistry(), _network, _id);
+ return new AMQProtocolEngine(_broker, _network, _id);
}
};
@@ -292,7 +292,7 @@ public class MultiVersionProtocolEngine
public ServerProtocolEngine getProtocolEngine()
{
- return new AMQProtocolEngine(_appRegistry.getVirtualHostRegistry(), _network, _id);
+ return new AMQProtocolEngine(_broker, _network, _id);
}
};
@@ -313,15 +313,15 @@ public class MultiVersionProtocolEngine
public ServerProtocolEngine getProtocolEngine()
{
- final ConnectionDelegate connDelegate =
- new org.apache.qpid.server.transport.ServerConnectionDelegate(_appRegistry, _fqdn, _appRegistry.getSubjectCreator(getLocalAddress()));
+ final ConnectionDelegate connDelegate = new org.apache.qpid.server.transport.ServerConnectionDelegate(_broker,
+ _fqdn, _broker.getSubjectCreator(getLocalAddress()));
ServerConnection conn = new ServerConnection(_id);
conn.setConnectionDelegate(connDelegate);
conn.setRemoteAddress(_network.getRemoteAddress());
conn.setLocalAddress(_network.getLocalAddress());
- return new ProtocolEngine_0_10( conn, _network, _appRegistry);
+ return new ProtocolEngine_0_10( conn, _network);
}
};
@@ -341,7 +341,7 @@ public class MultiVersionProtocolEngine
public ServerProtocolEngine getProtocolEngine()
{
- return new ProtocolEngine_1_0_0(_network, _appRegistry,_id);
+ return new ProtocolEngine_1_0_0(_network, _broker, _id);
}
};
@@ -361,7 +361,7 @@ public class MultiVersionProtocolEngine
public ServerProtocolEngine getProtocolEngine()
{
- return new ProtocolEngine_1_0_0_SASL(_network, _appRegistry, _id);
+ return new ProtocolEngine_1_0_0_SASL(_network, _broker, _id);
}
};
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/MultiVersionProtocolEngineFactory.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/MultiVersionProtocolEngineFactory.java?rev=1447646&r1=1447645&r2=1447646&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/MultiVersionProtocolEngineFactory.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/MultiVersionProtocolEngineFactory.java Tue Feb 19 09:35:28 2013
@@ -22,8 +22,7 @@ package org.apache.qpid.server.protocol;
import org.apache.qpid.protocol.ProtocolEngineFactory;
import org.apache.qpid.protocol.ServerProtocolEngine;
-import org.apache.qpid.server.registry.ApplicationRegistry;
-import org.apache.qpid.server.registry.IApplicationRegistry;
+import org.apache.qpid.server.model.Broker;
import java.util.Set;
import java.util.concurrent.atomic.AtomicLong;
@@ -32,11 +31,12 @@ public class MultiVersionProtocolEngineF
{
private static final AtomicLong ID_GENERATOR = new AtomicLong(0);
- private final IApplicationRegistry _appRegistry;
+ private final Broker _broker;
private final Set<AmqpProtocolVersion> _supported;
private final AmqpProtocolVersion _defaultSupportedReply;
- public MultiVersionProtocolEngineFactory(final Set<AmqpProtocolVersion> supportedVersions, final AmqpProtocolVersion defaultSupportedReply)
+ public MultiVersionProtocolEngineFactory(Broker broker,
+ final Set<AmqpProtocolVersion> supportedVersions, final AmqpProtocolVersion defaultSupportedReply)
{
if(defaultSupportedReply != null && !supportedVersions.contains(defaultSupportedReply))
{
@@ -44,14 +44,14 @@ public class MultiVersionProtocolEngineF
+ ") to an unsupported protocol version initiation is itself not supported!");
}
- _appRegistry = ApplicationRegistry.getInstance();
+ _broker = broker;
_supported = supportedVersions;
_defaultSupportedReply = defaultSupportedReply;
}
public ServerProtocolEngine newProtocolEngine()
{
- return new MultiVersionProtocolEngine(_appRegistry, _supported, _defaultSupportedReply, ID_GENERATOR.getAndIncrement());
+ return new MultiVersionProtocolEngine(_broker, _supported, _defaultSupportedReply, ID_GENERATOR.getAndIncrement());
}
}
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_0_10.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_0_10.java?rev=1447646&r1=1447645&r2=1447646&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_0_10.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_0_10.java Tue Feb 19 09:35:28 2013
@@ -22,7 +22,6 @@ package org.apache.qpid.server.protocol;
import org.apache.qpid.protocol.ServerProtocolEngine;
import org.apache.qpid.server.logging.messages.ConnectionMessages;
-import org.apache.qpid.server.registry.IApplicationRegistry;
import org.apache.qpid.server.transport.ServerConnection;
import org.apache.qpid.transport.Sender;
import org.apache.qpid.transport.network.Assembler;
@@ -32,7 +31,7 @@ import org.apache.qpid.transport.network
import java.net.SocketAddress;
import java.nio.ByteBuffer;
-import java.util.UUID;
+
public class ProtocolEngine_0_10 extends InputHandler implements ServerProtocolEngine
{
@@ -42,19 +41,17 @@ public class ProtocolEngine_0_10 extend
private long _readBytes;
private long _writtenBytes;
private ServerConnection _connection;
- private final IApplicationRegistry _appRegistry;
+
private long _createTime = System.currentTimeMillis();
private long _lastReadTime;
private long _lastWriteTime;
public ProtocolEngine_0_10(ServerConnection conn,
- NetworkConnection network,
- final IApplicationRegistry appRegistry)
+ NetworkConnection network)
{
super(new Assembler(conn));
_connection = conn;
- _appRegistry = appRegistry;
if(network != null)
{
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_1_0_0.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_1_0_0.java?rev=1447646&r1=1447645&r2=1447646&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_1_0_0.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_1_0_0.java Tue Feb 19 09:35:28 2013
@@ -22,7 +22,6 @@ package org.apache.qpid.server.protocol;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
-import java.util.UUID;
import java.util.concurrent.atomic.AtomicLong;
import java.util.logging.Level;
import java.util.logging.Logger;
@@ -39,10 +38,10 @@ import org.apache.qpid.amqp_1_0.transpor
import org.apache.qpid.amqp_1_0.type.Binary;
import org.apache.qpid.amqp_1_0.type.FrameBody;
import org.apache.qpid.protocol.ServerProtocolEngine;
+import org.apache.qpid.server.model.Broker;
import org.apache.qpid.server.protocol.v1_0.Connection_1_0;
-import org.apache.qpid.server.registry.IApplicationRegistry;
import org.apache.qpid.server.security.SubjectCreator;
-import org.apache.qpid.server.security.auth.manager.AuthenticationManager;
+import org.apache.qpid.server.virtualhost.VirtualHost;
import org.apache.qpid.transport.Sender;
import org.apache.qpid.transport.network.NetworkConnection;
@@ -55,7 +54,7 @@ public class ProtocolEngine_1_0_0 implem
private long _writtenBytes;
private long _lastReadTime;
private long _lastWriteTime;
- private final IApplicationRegistry _appRegistry;
+ private final Broker _broker;
private long _createTime = System.currentTimeMillis();
private ConnectionEndpoint _conn;
private final long _connectionId;
@@ -99,9 +98,9 @@ public class ProtocolEngine_1_0_0 implem
- public ProtocolEngine_1_0_0(final NetworkConnection networkDriver, final IApplicationRegistry appRegistry, long id)
+ public ProtocolEngine_1_0_0(final NetworkConnection networkDriver, final Broker broker, long id)
{
- _appRegistry = appRegistry;
+ _broker = broker;
_connectionId = id;
if(networkDriver != null)
{
@@ -145,12 +144,14 @@ public class ProtocolEngine_1_0_0 implem
_network = network;
_sender = sender;
- Container container = new Container(_appRegistry.getBrokerId().toString());
+ Container container = new Container(_broker.getId().toString());
- _conn = new ConnectionEndpoint(container, asSaslServerProvider(_appRegistry.getSubjectCreator(
+ VirtualHost virtualHost = _broker.getVirtualHostRegistry().getVirtualHost((String)_broker.getAttribute(Broker.DEFAULT_VIRTUAL_HOST));
+
+ _conn = new ConnectionEndpoint(container, asSaslServerProvider(_broker.getSubjectCreator(
getLocalAddress())));
_conn.setRemoteAddress(_network.getRemoteAddress());
- _conn.setConnectionEventListener(new Connection_1_0(_appRegistry, _conn, _connectionId));
+ _conn.setConnectionEventListener(new Connection_1_0(virtualHost, _conn, _connectionId));
_conn.setFrameOutputHandler(this);
_frameWriter = new FrameWriter(_conn.getDescribedTypeRegistry());
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_1_0_0_SASL.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_1_0_0_SASL.java?rev=1447646&r1=1447645&r2=1447646&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_1_0_0_SASL.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_1_0_0_SASL.java Tue Feb 19 09:35:28 2013
@@ -23,7 +23,6 @@ package org.apache.qpid.server.protocol;
import java.io.PrintWriter;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
-import java.util.UUID;
import java.util.logging.Level;
import java.util.logging.Logger;
import javax.security.sasl.SaslException;
@@ -40,10 +39,10 @@ import org.apache.qpid.amqp_1_0.transpor
import org.apache.qpid.amqp_1_0.type.Binary;
import org.apache.qpid.amqp_1_0.type.FrameBody;
import org.apache.qpid.protocol.ServerProtocolEngine;
+import org.apache.qpid.server.model.Broker;
import org.apache.qpid.server.protocol.v1_0.Connection_1_0;
-import org.apache.qpid.server.registry.ApplicationRegistry;
-import org.apache.qpid.server.registry.IApplicationRegistry;
import org.apache.qpid.server.security.SubjectCreator;
+import org.apache.qpid.server.virtualhost.VirtualHost;
import org.apache.qpid.transport.Sender;
import org.apache.qpid.transport.network.NetworkConnection;
@@ -54,7 +53,7 @@ public class ProtocolEngine_1_0_0_SASL i
private long _lastReadTime;
private long _lastWriteTime;
- private final IApplicationRegistry _appRegistry;
+ private final Broker _broker;
private long _createTime = System.currentTimeMillis();
private ConnectionEndpoint _conn;
private long _connectionId;
@@ -113,12 +112,11 @@ public class ProtocolEngine_1_0_0_SASL i
private State _state = State.A;
- public ProtocolEngine_1_0_0_SASL(final NetworkConnection networkDriver, final IApplicationRegistry appRegistry,
+ public ProtocolEngine_1_0_0_SASL(final NetworkConnection networkDriver, final Broker broker,
long id)
{
_connectionId = id;
- _appRegistry = appRegistry;
-
+ _broker = broker;
if(networkDriver != null)
{
setNetworkConnection(networkDriver, networkDriver.getSender());
@@ -161,15 +159,12 @@ public class ProtocolEngine_1_0_0_SASL i
_network = network;
_sender = sender;
- Container container = new Container(_appRegistry.getBrokerId().toString());
+ Container container = new Container(_broker.getId().toString());
- _conn = new ConnectionEndpoint(container, asSaslServerProvider(ApplicationRegistry.getInstance()
- .getSubjectCreator(getLocalAddress())));
+ VirtualHost virtualHost = _broker.getVirtualHostRegistry().getVirtualHost((String)_broker.getAttribute(Broker.DEFAULT_VIRTUAL_HOST));
+ _conn = new ConnectionEndpoint(container, asSaslServerProvider(_broker.getSubjectCreator(getLocalAddress())));
_conn.setRemoteAddress(getRemoteAddress());
- _conn.setConnectionEventListener(new Connection_1_0(_appRegistry, _conn, _connectionId));
-
-
-
+ _conn.setConnectionEventListener(new Connection_1_0(virtualHost, _conn, _connectionId));
_conn.setFrameOutputHandler(this);
_conn.setSaslFrameOutput(this);
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/v1_0/Connection_1_0.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/v1_0/Connection_1_0.java?rev=1447646&r1=1447645&r2=1447646&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/v1_0/Connection_1_0.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/v1_0/Connection_1_0.java Tue Feb 19 09:35:28 2013
@@ -31,7 +31,6 @@ import org.apache.qpid.protocol.AMQConst
import org.apache.qpid.server.logging.LogSubject;
import org.apache.qpid.server.protocol.AMQConnectionModel;
import org.apache.qpid.server.protocol.AMQSessionModel;
-import org.apache.qpid.server.registry.IApplicationRegistry;
import org.apache.qpid.server.stats.StatisticsCounter;
import org.apache.qpid.server.virtualhost.VirtualHost;
@@ -44,7 +43,6 @@ import static org.apache.qpid.server.log
public class Connection_1_0 implements ConnectionEventListener
{
- private IApplicationRegistry _appRegistry;
private VirtualHost _vhost;
private final ConnectionEndpoint _conn;
private final long _connectionId;
@@ -62,10 +60,9 @@ public class Connection_1_0 implements C
- public Connection_1_0(IApplicationRegistry appRegistry, ConnectionEndpoint conn, long connectionId)
+ public Connection_1_0(VirtualHost virtualHost, ConnectionEndpoint conn, long connectionId)
{
- _appRegistry = appRegistry;
- _vhost = _appRegistry.getVirtualHostRegistry().getDefaultVirtualHost();
+ _vhost = virtualHost;
_conn = conn;
_connectionId = connectionId;
_vhost.getConnectionRegistry().registerConnection(_model);
@@ -74,7 +71,7 @@ public class Connection_1_0 implements C
public void remoteSessionCreation(SessionEndpoint endpoint)
{
- Session_1_0 session = new Session_1_0(_vhost, _appRegistry, this);
+ Session_1_0 session = new Session_1_0(_vhost, this);
_sessions.add(session);
endpoint.setSessionEventListener(session);
}
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java?rev=1447646&r1=1447645&r2=1447646&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java Tue Feb 19 09:35:28 2013
@@ -36,7 +36,6 @@ import org.apache.qpid.amqp_1_0.type.tra
import org.apache.qpid.amqp_1_0.type.transport.Error;
import org.apache.qpid.AMQException;
import org.apache.qpid.AMQSecurityException;
-import org.apache.qpid.protocol.ProtocolEngine;
import org.apache.qpid.server.exchange.Exchange;
import org.apache.qpid.server.logging.LogSubject;
import org.apache.qpid.server.message.InboundMessage;
@@ -45,8 +44,6 @@ import org.apache.qpid.server.protocol.A
import org.apache.qpid.server.protocol.AMQSessionModel;
import org.apache.qpid.server.queue.AMQQueue;
import org.apache.qpid.server.queue.AMQQueueFactory;
-import org.apache.qpid.server.registry.IApplicationRegistry;
-import org.apache.qpid.server.transport.ServerConnection;
import org.apache.qpid.server.txn.AutoCommitTransaction;
import org.apache.qpid.server.txn.ServerTransaction;
import org.apache.qpid.server.virtualhost.VirtualHost;
@@ -58,7 +55,6 @@ import static org.apache.qpid.server.log
public class Session_1_0 implements SessionEventListener, AMQSessionModel, LogSubject
{
private static final Symbol LIFETIME_POLICY = Symbol.valueOf("lifetime-policy");
- private IApplicationRegistry _appRegistry;
private VirtualHost _vhost;
private AutoCommitTransaction _transaction;
@@ -68,9 +64,8 @@ public class Session_1_0 implements Sess
private UUID _id = UUID.randomUUID();
- public Session_1_0(VirtualHost vhost, IApplicationRegistry appRegistry, final Connection_1_0 connection)
+ public Session_1_0(VirtualHost vhost, final Connection_1_0 connection)
{
- _appRegistry = appRegistry;
_vhost = vhost;
_transaction = new AutoCommitTransaction(vhost.getMessageStore());
_connection = connection;
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueueFactory.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueueFactory.java?rev=1447646&r1=1447645&r2=1447646&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueueFactory.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueueFactory.java Tue Feb 19 09:35:28 2013
@@ -30,13 +30,13 @@ import org.apache.qpid.AMQSecurityExcept
import org.apache.qpid.exchange.ExchangeDefaults;
import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.framing.FieldTable;
+import org.apache.qpid.server.configuration.BrokerProperties;
import org.apache.qpid.server.configuration.QueueConfiguration;
-import org.apache.qpid.server.configuration.ServerConfiguration;
+import org.apache.qpid.server.exchange.DefaultExchangeFactory;
import org.apache.qpid.server.exchange.Exchange;
import org.apache.qpid.server.exchange.ExchangeFactory;
import org.apache.qpid.server.exchange.ExchangeRegistry;
import org.apache.qpid.server.model.UUIDGenerator;
-import org.apache.qpid.server.registry.ApplicationRegistry;
import org.apache.qpid.server.virtualhost.VirtualHost;
public class AMQQueueFactory
@@ -426,9 +426,7 @@ public class AMQQueueFactory
*/
protected static String getDeadLetterQueueName(String name)
{
- ServerConfiguration serverConfig = ApplicationRegistry.getInstance().getConfiguration();
- String dlQueueName = name + serverConfig.getDeadLetterQueueSuffix();
- return dlQueueName;
+ return name + System.getProperty(BrokerProperties.PROPERTY_DEAD_LETTER_QUEUE_SUFFIX, DEFAULT_DLQ_NAME_SUFFIX);
}
/**
@@ -440,9 +438,7 @@ public class AMQQueueFactory
*/
protected static String getDeadLetterExchangeName(String name)
{
- ServerConfiguration serverConfig = ApplicationRegistry.getInstance().getConfiguration();
- String dlExchangeName = name + serverConfig.getDeadLetterExchangeSuffix();
- return dlExchangeName;
+ return name + System.getProperty(BrokerProperties.PROPERTY_DEAD_LETTER_EXCHANGE_SUFFIX, DefaultExchangeFactory.DEFAULT_DLE_NAME_SUFFIX);
}
private static Map<String, Object> createQueueArgumentsFromConfig(QueueConfiguration config)
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/IncomingMessage.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/IncomingMessage.java?rev=1447646&r1=1447645&r2=1447646&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/IncomingMessage.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/IncomingMessage.java Tue Feb 19 09:35:28 2013
@@ -34,7 +34,6 @@ import org.apache.qpid.server.message.En
import org.apache.qpid.server.message.InboundMessage;
import org.apache.qpid.server.message.MessageContentSource;
import org.apache.qpid.server.message.MessageMetaData;
-import org.apache.qpid.server.registry.ApplicationRegistry;
import org.apache.qpid.server.store.StoredMessage;
import java.nio.ByteBuffer;
@@ -47,9 +46,6 @@ public class IncomingMessage implements
/** Used for debugging purposes. */
private static final Logger _logger = Logger.getLogger(IncomingMessage.class);
- private static final boolean SYNCHED_CLOCKS =
- ApplicationRegistry.getInstance().getConfiguration().getSynchedClocks();
-
private final MessagePublishInfo _messagePublishInfo;
private ContentHeaderBody _contentHeaderBody;
@@ -101,33 +97,7 @@ public class IncomingMessage implements
public void setExpiration()
{
- long expiration =
- ((BasicContentHeaderProperties) _contentHeaderBody.getProperties()).getExpiration();
- long timestamp =
- ((BasicContentHeaderProperties) _contentHeaderBody.getProperties()).getTimestamp();
-
- if (SYNCHED_CLOCKS)
- {
- _expiration = expiration;
- }
- else
- {
- // Update TTL to be in broker time.
- if (expiration != 0L)
- {
- if (timestamp != 0L)
- {
- // todo perhaps use arrival time
- long diff = (System.currentTimeMillis() - timestamp);
-
- if ((diff > 1000L) || (diff < 1000L))
- {
- _expiration = expiration + diff;
- }
- }
- }
- }
-
+ _expiration = ((BasicContentHeaderProperties) _contentHeaderBody.getProperties()).getExpiration();
}
public MessageMetaData headersReceived(long currentTime)
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java?rev=1447646&r1=1447645&r2=1447646&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java Tue Feb 19 09:35:28 2013
@@ -52,7 +52,6 @@ import org.apache.qpid.server.logging.me
import org.apache.qpid.server.logging.subjects.QueueLogSubject;
import org.apache.qpid.server.message.ServerMessage;
import org.apache.qpid.server.protocol.AMQSessionModel;
-import org.apache.qpid.server.registry.ApplicationRegistry;
import org.apache.qpid.server.security.AuthorizationHolder;
import org.apache.qpid.server.subscription.AssignedSubscriptionMessageGroupManager;
import org.apache.qpid.server.subscription.DefinedGroupMessageGroupManager;
@@ -132,23 +131,23 @@ public class SimpleAMQQueue implements A
private final AtomicInteger _bindingCountHigh = new AtomicInteger();
/** max allowed size(KB) of a single message */
- private long _maximumMessageSize = ApplicationRegistry.getInstance().getConfiguration().getMaximumMessageSize();
+ private long _maximumMessageSize;
/** max allowed number of messages on a queue. */
- private long _maximumMessageCount = ApplicationRegistry.getInstance().getConfiguration().getMaximumMessageCount();
+ private long _maximumMessageCount;
/** max queue depth for the queue */
- private long _maximumQueueDepth = ApplicationRegistry.getInstance().getConfiguration().getMaximumQueueDepth();
+ private long _maximumQueueDepth;
/** maximum message age before alerts occur */
- private long _maximumMessageAge = ApplicationRegistry.getInstance().getConfiguration().getMaximumMessageAge();
+ private long _maximumMessageAge;
/** the minimum interval between sending out consecutive alerts of the same type */
- private long _minimumAlertRepeatGap = ApplicationRegistry.getInstance().getConfiguration().getMinimumAlertRepeatGap();
+ private long _minimumAlertRepeatGap;
- private long _capacity = ApplicationRegistry.getInstance().getConfiguration().getCapacity();
+ private long _capacity;
- private long _flowResumeCapacity = ApplicationRegistry.getInstance().getConfiguration().getFlowResumeCapacity();
+ private long _flowResumeCapacity;
private final Set<NotificationCheck> _notificationChecks = EnumSet.noneOf(NotificationCheck.class);
@@ -185,7 +184,7 @@ public class SimpleAMQQueue implements A
private AbstractConfiguration _queueConfiguration;
/** the maximum delivery count for each message on this queue or 0 if maximum delivery count is not to be enforced. */
- private int _maximumDeliveryCount = ApplicationRegistry.getInstance().getConfiguration().getMaxDeliveryCount();
+ private int _maximumDeliveryCount;
private final MessageGroupManager _messageGroupManager;
private final Collection<SubscriptionRegistrationListener> _subscriptionListeners =
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org