You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ro...@apache.org on 2012/04/17 12:29:16 UTC
svn commit: r1327029 [3/5] - in
/qpid/branches/java-config-and-management/qpid/java: ./
bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/
bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/
broker-plugins/extras/src/main/java/...
Modified: qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/binding/BindingFactory.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/binding/BindingFactory.java?rev=1327029&r1=1327028&r2=1327029&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/binding/BindingFactory.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/binding/BindingFactory.java Tue Apr 17 10:29:11 2012
@@ -34,7 +34,6 @@ import org.apache.qpid.server.logging.ac
import org.apache.qpid.server.logging.messages.BindingMessages;
import org.apache.qpid.server.logging.subjects.BindingLogSubject;
import org.apache.qpid.server.queue.AMQQueue;
-import org.apache.qpid.server.store.DurableConfigurationStore;
import org.apache.qpid.server.virtualhost.VirtualHost;
import java.util.Collections;
@@ -44,38 +43,14 @@ import java.util.concurrent.ConcurrentHa
public class BindingFactory
{
private final VirtualHost _virtualHost;
- private final DurableConfigurationStore.Source _configSource;
- private final Exchange _defaultExchange;
private final ConcurrentHashMap<BindingImpl, BindingImpl> _bindings = new ConcurrentHashMap<BindingImpl, BindingImpl>();
-
public BindingFactory(final VirtualHost vhost)
{
- this(vhost, vhost.getExchangeRegistry().getDefaultExchange());
- }
-
- public BindingFactory(final DurableConfigurationStore.Source configSource, final Exchange defaultExchange)
- {
- _configSource = configSource;
- _defaultExchange = defaultExchange;
- if (configSource instanceof VirtualHost)
- {
- _virtualHost = (VirtualHost) configSource;
- }
- else
- {
- _virtualHost = null;
- }
- }
-
- public VirtualHost getVirtualHost()
- {
- return _virtualHost;
+ _virtualHost = vhost;
}
-
-
private final class BindingImpl extends Binding implements AMQQueue.Task, Exchange.Task, BindingConfig
{
private final BindingLogSubject _logSubject;
@@ -156,30 +131,38 @@ public class BindingFactory
private boolean makeBinding(String bindingKey, AMQQueue queue, Exchange exchange, Map<String, Object> arguments, boolean restore, boolean force) throws AMQSecurityException, AMQInternalException
{
assert queue != null;
+ final Exchange defaultExchange = _virtualHost.getExchangeRegistry().getDefaultExchange();
+
if (bindingKey == null)
{
bindingKey = "";
}
if (exchange == null)
{
- exchange = _defaultExchange;
+ exchange = defaultExchange;
}
if (arguments == null)
{
arguments = Collections.emptyMap();
}
+ if (exchange == null)
+ {
+ throw new IllegalArgumentException("exchange cannot be null");
+ }
+
// The default exchange bindings must reflect the existence of queues, allow
// all operations on it to succeed. It is up to the broker to prevent illegal
// attempts at binding to this exchange, not the ACLs.
- if(exchange != _defaultExchange)
+ if(exchange != defaultExchange)
{
//Perform ACLs
- if (!getVirtualHost().getSecurityManager().authoriseBind(exchange, queue, new AMQShortString(bindingKey)))
+ if (!_virtualHost.getSecurityManager().authoriseBind(exchange, queue, new AMQShortString(bindingKey)))
{
throw new AMQSecurityException("Permission denied: binding " + bindingKey);
}
}
+
BindingImpl b = new BindingImpl(bindingKey,queue,exchange,arguments);
BindingImpl existingMapping = _bindings.putIfAbsent(b,b);
@@ -192,7 +175,7 @@ public class BindingFactory
if (b.isDurable() && !restore)
{
- _configSource.getMessageStore().bindQueue(exchange,new AMQShortString(bindingKey),queue,FieldTable.convertToFieldTable(arguments));
+ _virtualHost.getMessageStore().bindQueue(exchange,new AMQShortString(bindingKey),queue,FieldTable.convertToFieldTable(arguments));
}
queue.addQueueDeleteTask(b);
@@ -212,7 +195,7 @@ public class BindingFactory
private ConfigStore getConfigStore()
{
- return getVirtualHost().getConfigStore();
+ return _virtualHost.getConfigStore();
}
public void restoreBinding(final String bindingKey, final AMQQueue queue, final Exchange exchange, final Map<String, Object> argumentMap) throws AMQSecurityException, AMQInternalException
@@ -229,13 +212,15 @@ public class BindingFactory
public Binding removeBinding(String bindingKey, AMQQueue queue, Exchange exchange, Map<String, Object> arguments) throws AMQSecurityException, AMQInternalException
{
assert queue != null;
+ final Exchange defaultExchange = _virtualHost.getExchangeRegistry().getDefaultExchange();
+
if (bindingKey == null)
{
bindingKey = "";
}
if (exchange == null)
{
- exchange = _defaultExchange;
+ exchange = defaultExchange;
}
if (arguments == null)
{
@@ -245,10 +230,10 @@ public class BindingFactory
// The default exchange bindings must reflect the existence of queues, allow
// all operations on it to succeed. It is up to the broker to prevent illegal
// attempts at binding to this exchange, not the ACLs.
- if(exchange != _defaultExchange)
+ if(exchange != defaultExchange)
{
// Check access
- if (!getVirtualHost().getSecurityManager().authoriseUnbind(exchange, new AMQShortString(bindingKey), queue))
+ if (!_virtualHost.getSecurityManager().authoriseUnbind(exchange, new AMQShortString(bindingKey), queue))
{
throw new AMQSecurityException("Permission denied: unbinding " + bindingKey);
}
@@ -265,7 +250,7 @@ public class BindingFactory
if (b.isDurable())
{
- _configSource.getMessageStore().unbindQueue(exchange,
+ _virtualHost.getMessageStore().unbindQueue(exchange,
new AMQShortString(bindingKey),
queue,
FieldTable.convertToFieldTable(arguments));
@@ -280,13 +265,15 @@ public class BindingFactory
public Binding getBinding(String bindingKey, AMQQueue queue, Exchange exchange, Map<String, Object> arguments)
{
assert queue != null;
+ final Exchange defaultExchange = _virtualHost.getExchangeRegistry().getDefaultExchange();
+
if(bindingKey == null)
{
bindingKey = "";
}
if(exchange == null)
{
- exchange = _defaultExchange;
+ exchange = defaultExchange;
}
if(arguments == null)
{
Modified: qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/VirtualHostConfiguration.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/VirtualHostConfiguration.java?rev=1327029&r1=1327028&r2=1327029&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/VirtualHostConfiguration.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/VirtualHostConfiguration.java Tue Apr 17 10:29:11 2012
@@ -32,6 +32,7 @@ import org.apache.qpid.server.configurat
import org.apache.qpid.server.queue.AMQQueue;
import org.apache.qpid.server.registry.ApplicationRegistry;
import org.apache.qpid.server.store.MemoryMessageStore;
+import org.apache.qpid.server.store.MemoryMessageStoreFactory;
import java.util.ArrayList;
import java.util.Arrays;
@@ -102,14 +103,14 @@ public class VirtualHostConfiguration ex
return getConfig().subset("store");
}
- public String getMessageStoreClass()
+ public String getMessageStoreFactoryClass()
{
- return getStringValue("store.class", MemoryMessageStore.class.getName());
+ return getStringValue("store.factoryclass", MemoryMessageStoreFactory.class.getName());
}
- public void setMessageStoreClass(String storeClass)
+ public void setMessageStoreFactoryClass(String storeFactoryClass)
{
- getConfig().setProperty("store.class", storeClass);
+ getConfig().setProperty("store.factoryclass", storeFactoryClass);
}
public List getExchanges()
Modified: qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/connection/ConnectionRegistry.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/connection/ConnectionRegistry.java?rev=1327029&r1=1327028&r2=1327029&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/connection/ConnectionRegistry.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/connection/ConnectionRegistry.java Tue Apr 17 10:29:11 2012
@@ -49,11 +49,19 @@ public class ConnectionRegistry implemen
/** Close all of the currently open connections. */
public void close()
{
- _logger.debug("Closing connection registry :" + _registry.size() + " connections.");
+ close(IConnectionRegistry.BROKER_SHUTDOWN_REPLY_TEXT);
+ }
+
+ public void close(final String replyText)
+ {
+ if (_logger.isDebugEnabled())
+ {
+ _logger.debug("Closing connection registry :" + _registry.size() + " connections.");
+ }
while (!_registry.isEmpty())
{
AMQConnectionModel connection = _registry.get(0);
- closeConnection(connection, AMQConstant.CONNECTION_FORCED, "Broker is shutting down");
+ closeConnection(connection, AMQConstant.CONNECTION_FORCED, replyText);
}
}
Modified: qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/connection/IConnectionRegistry.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/connection/IConnectionRegistry.java?rev=1327029&r1=1327028&r2=1327029&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/connection/IConnectionRegistry.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/connection/IConnectionRegistry.java Tue Apr 17 10:29:11 2012
@@ -28,12 +28,17 @@ import java.util.List;
public interface IConnectionRegistry
{
+ public static final String BROKER_SHUTDOWN_REPLY_TEXT = "Broker is shutting down";
+ public static final String VHOST_PASSIVATE_REPLY_TEXT = "Virtual host is being passivated";
+
public void initialise();
public void close() throws AMQException;
-
+
+ public void close(String replyText) throws AMQException;
+
public void closeConnection(AMQConnectionModel connection, AMQConstant cause, String message);
-
+
public List<AMQConnectionModel> getConnections();
public void registerConnection(AMQConnectionModel connnection);
Modified: qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/AbstractExchange.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/AbstractExchange.java?rev=1327029&r1=1327028&r2=1327029&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/AbstractExchange.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/AbstractExchange.java Tue Apr 17 10:29:11 2012
@@ -20,8 +20,6 @@
*/
package org.apache.qpid.server.exchange;
-import org.apache.log4j.Logger;
-
import org.apache.qpid.AMQException;
import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.server.binding.Binding;
@@ -128,10 +126,18 @@ public abstract class AbstractExchange i
_autoDelete = autoDelete;
_ticket = ticket;
- // TODO - fix
_id = getConfigStore().createId();
getConfigStore().addConfiguredObject(this);
+ createAndRegisterMBean();
+ _logSubject = new ExchangeLogSubject(this, this.getVirtualHost());
+
+ // Log Exchange creation
+ CurrentActor.get().message(ExchangeMessages.CREATED(String.valueOf(getTypeShortString()), String.valueOf(name), durable));
+ }
+
+ private void createAndRegisterMBean()
+ {
try
{
_exchangeMbean = createMBean();
@@ -139,12 +145,8 @@ public abstract class AbstractExchange i
}
catch (JMException e)
{
- getLogger().error(e);
+ throw new RuntimeException("Failed to register mbean",e);
}
- _logSubject = new ExchangeLogSubject(this, this.getVirtualHost());
-
- // Log Exchange creation
- CurrentActor.get().message(ExchangeMessages.CREATED(String.valueOf(getTypeShortString()), String.valueOf(name), durable));
}
public ConfigStore getConfigStore()
@@ -152,8 +154,6 @@ public abstract class AbstractExchange i
return getVirtualHost().getConfigStore();
}
- public abstract Logger getLogger();
-
public boolean isDurable()
{
return _durable;
@@ -327,8 +327,7 @@ public abstract class AbstractExchange i
public Map<String, Object> getArguments()
{
- // TODO - Fix
- return Collections.EMPTY_MAP;
+ return Collections.emptyMap();
}
public UUID getId()
Modified: qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/DefaultExchangeRegistry.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/DefaultExchangeRegistry.java?rev=1327029&r1=1327028&r2=1327029&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/DefaultExchangeRegistry.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/DefaultExchangeRegistry.java Tue Apr 17 10:29:11 2012
@@ -21,7 +21,6 @@
package org.apache.qpid.server.exchange;
import org.apache.log4j.Logger;
-
import org.apache.qpid.AMQException;
import org.apache.qpid.AMQSecurityException;
import org.apache.qpid.exchange.ExchangeDefaults;
@@ -38,7 +37,7 @@ import java.util.concurrent.ConcurrentMa
public class DefaultExchangeRegistry implements ExchangeRegistry
{
- private static final Logger _log = Logger.getLogger(DefaultExchangeRegistry.class);
+ private static final Logger LOGGER = Logger.getLogger(DefaultExchangeRegistry.class);
/**
* Maps from exchange name to exchange instance
@@ -63,8 +62,6 @@ public class DefaultExchangeRegistry imp
new ExchangeInitialiser().initialise(_host.getExchangeFactory(), this, getDurableConfigurationStore());
}
-
-
public DurableConfigurationStore getDurableConfigurationStore()
{
return _host.getMessageStore();
@@ -184,4 +181,28 @@ public class DefaultExchangeRegistry imp
}
}
+ @Override
+ public void clearAndUnregisterMbeans()
+ {
+ for (final AMQShortString exchangeName : getExchangeNames())
+ {
+ final Exchange exchange = getExchange(exchangeName);
+
+ if (exchange instanceof AbstractExchange)
+ {
+ AbstractExchange abstractExchange = (AbstractExchange) exchange;
+ try
+ {
+ abstractExchange.getManagedObject().unregister();
+ }
+ catch (AMQException e)
+ {
+ LOGGER.warn("Failed to unregister mbean", e);
+ }
+ }
+ }
+ _exchangeMap.clear();
+ _exchangeMapStr.clear();
+ }
+
}
Modified: qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/DirectExchange.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/DirectExchange.java?rev=1327029&r1=1327028&r2=1327029&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/DirectExchange.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/DirectExchange.java Tue Apr 17 10:29:11 2012
@@ -133,12 +133,6 @@ public class DirectExchange extends Abst
return new DirectExchangeMBean(this);
}
- public Logger getLogger()
- {
- return _logger;
- }
-
-
public List<? extends BaseQueue> doRoute(InboundMessage payload)
{
Modified: qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/ExchangeInitialiser.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/ExchangeInitialiser.java?rev=1327029&r1=1327028&r2=1327029&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/ExchangeInitialiser.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/ExchangeInitialiser.java Tue Apr 17 10:29:11 2012
@@ -27,7 +27,8 @@ import org.apache.qpid.server.store.Dura
public class ExchangeInitialiser
{
- public void initialise(ExchangeFactory factory, ExchangeRegistry registry, DurableConfigurationStore store) throws AMQException{
+ public void initialise(ExchangeFactory factory, ExchangeRegistry registry, DurableConfigurationStore store) throws AMQException
+ {
for (ExchangeType<? extends Exchange> type : factory.getRegisteredTypes())
{
define (registry, factory, type.getDefaultExchangeName(), type.getName(), store);
@@ -44,7 +45,6 @@ public class ExchangeInitialiser
{
Exchange exchange = f.createExchange(name, type, true, false, 0);
r.registerExchange(exchange);
-
if(exchange.isDurable())
{
store.createExchange(exchange);
Modified: qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/ExchangeRegistry.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/ExchangeRegistry.java?rev=1327029&r1=1327028&r2=1327029&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/ExchangeRegistry.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/ExchangeRegistry.java Tue Apr 17 10:29:11 2012
@@ -51,7 +51,9 @@ public interface ExchangeRegistry
Exchange getExchange(String exchangeName);
- void unregisterExchange(String exchange, boolean ifUnused) throws ExchangeInUseException, AMQException;;
+ void unregisterExchange(String exchange, boolean ifUnused) throws ExchangeInUseException, AMQException;
+
+ void clearAndUnregisterMbeans();
Collection<Exchange> getExchanges();
Modified: qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/FanoutExchange.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/FanoutExchange.java?rev=1327029&r1=1327028&r2=1327029&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/FanoutExchange.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/FanoutExchange.java Tue Apr 17 10:29:11 2012
@@ -54,11 +54,6 @@ public class FanoutExchange extends Abst
return new FanoutExchangeMBean(this);
}
- public Logger getLogger()
- {
- return _logger;
- }
-
public static final ExchangeType<FanoutExchange> TYPE = new ExchangeType<FanoutExchange>()
{
Modified: qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/HeadersExchange.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/HeadersExchange.java?rev=1327029&r1=1327028&r2=1327029&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/HeadersExchange.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/HeadersExchange.java Tue Apr 17 10:29:11 2012
@@ -233,11 +233,6 @@ public class HeadersExchange extends Abs
return new HeadersExchangeMBean(this);
}
- public Logger getLogger()
- {
- return _logger;
- }
-
protected void onBind(final Binding binding)
{
String bindingKey = binding.getBindingKey();
Modified: qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/TopicExchange.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/TopicExchange.java?rev=1327029&r1=1327028&r2=1327029&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/TopicExchange.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/TopicExchange.java Tue Apr 17 10:29:11 2012
@@ -409,11 +409,6 @@ public class TopicExchange extends Abstr
return new TopicExchangeMBean(this);
}
- public Logger getLogger()
- {
- return _logger;
- }
-
private Collection<AMQQueue> getMatchedQueues(InboundMessage message, AMQShortString routingKey)
{
Modified: qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionOpenMethodHandler.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionOpenMethodHandler.java?rev=1327029&r1=1327028&r2=1327029&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionOpenMethodHandler.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionOpenMethodHandler.java Tue Apr 17 10:29:11 2012
@@ -32,6 +32,7 @@ import org.apache.qpid.server.protocol.A
import org.apache.qpid.server.state.AMQState;
import org.apache.qpid.server.state.AMQStateManager;
import org.apache.qpid.server.state.StateAwareMethodListener;
+import org.apache.qpid.server.virtualhost.State;
import org.apache.qpid.server.virtualhost.VirtualHost;
public class ConnectionOpenMethodHandler implements StateAwareMethodListener<ConnectionOpenBody>
@@ -82,6 +83,10 @@ public class ConnectionOpenMethodHandler
{
throw body.getConnectionException(AMQConstant.ACCESS_REFUSED, "Permission denied: '" + virtualHost.getName() + "'");
}
+ else if (virtualHost.getState() != State.ACTIVE)
+ {
+ throw body.getConnectionException(AMQConstant.CONNECTION_FORCED, "Virtual host '" + virtualHost.getName() + "' is not active");
+ }
session.setVirtualHost(virtualHost);
@@ -89,10 +94,10 @@ public class ConnectionOpenMethodHandler
if (session.getContextKey() == null)
{
session.setContextKey(generateClientID());
- }
+ }
MethodRegistry methodRegistry = session.getMethodRegistry();
- AMQMethodBody responseBody = methodRegistry.createConnectionOpenOkBody(body.getVirtualHost());
+ AMQMethodBody responseBody = methodRegistry.createConnectionOpenOkBody(body.getVirtualHost());
stateManager.changeState(AMQState.CONNECTION_OPEN);
Modified: qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/messages/ConfigStore_logmessages.properties
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/messages/ConfigStore_logmessages.properties?rev=1327029&r1=1327028&r2=1327029&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/messages/ConfigStore_logmessages.properties (original)
+++ qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/messages/ConfigStore_logmessages.properties Tue Apr 17 10:29:11 2012
@@ -18,8 +18,7 @@
#
# Default File used for all non-defined locales.
-# 0 - name
-CREATED = CFG-1001 : Created : {0}
+CREATED = CFG-1001 : Created
# 0 - path
STORE_LOCATION = CFG-1002 : Store location : {0}
CLOSE = CFG-1003 : Closed
Modified: qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/messages/MessageStore_logmessages.properties
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/messages/MessageStore_logmessages.properties?rev=1327029&r1=1327028&r2=1327029&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/messages/MessageStore_logmessages.properties (original)
+++ qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/messages/MessageStore_logmessages.properties Tue Apr 17 10:29:11 2012
@@ -18,11 +18,11 @@
#
# Default File used for all non-defined locales.
#
-# 0 - name
-CREATED = MST-1001 : Created : {0}
+CREATED = MST-1001 : Created
# 0 - path
STORE_LOCATION = MST-1002 : Store location : {0}
CLOSED = MST-1003 : Closed
RECOVERY_START = MST-1004 : Recovery Start
RECOVERED = MST-1005 : Recovered {0,number} messages
-RECOVERY_COMPLETE = MST-1006 : Recovery Complete
\ No newline at end of file
+RECOVERY_COMPLETE = MST-1006 : Recovery Complete
+PASSIVATE = MST-1007 : Store Passivated
Modified: qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/messages/TransactionLog_logmessages.properties
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/messages/TransactionLog_logmessages.properties?rev=1327029&r1=1327028&r2=1327029&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/messages/TransactionLog_logmessages.properties (original)
+++ qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/messages/TransactionLog_logmessages.properties Tue Apr 17 10:29:11 2012
@@ -19,8 +19,7 @@
# Default File used for all non-defined locales.
#
#
-# 0 - name
-CREATED = TXN-1001 : Created : {0}
+CREATED = TXN-1001 : Created
# 0 - path
STORE_LOCATION = TXN-1002 : Store location : {0}
CLOSED = TXN-1003 : Closed
Modified: qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/subjects/BindingLogSubject.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/subjects/BindingLogSubject.java?rev=1327029&r1=1327028&r2=1327029&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/subjects/BindingLogSubject.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/subjects/BindingLogSubject.java Tue Apr 17 10:29:11 2012
@@ -40,7 +40,8 @@ public class BindingLogSubject extends A
public BindingLogSubject(String routingKey, Exchange exchange,
AMQQueue queue)
{
- setLogStringWithFormat(BINDING_FORMAT, queue.getVirtualHost().getName(),
+ setLogStringWithFormat(BINDING_FORMAT,
+ queue.getVirtualHost().getName(),
exchange.getTypeShortString(),
exchange.getNameShortString(),
queue.getNameShortString(),
Modified: qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/subjects/MessageStoreLogSubject.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/subjects/MessageStoreLogSubject.java?rev=1327029&r1=1327028&r2=1327029&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/subjects/MessageStoreLogSubject.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/subjects/MessageStoreLogSubject.java Tue Apr 17 10:29:11 2012
@@ -20,7 +20,6 @@
*/
package org.apache.qpid.server.logging.subjects;
-import org.apache.qpid.server.store.MessageStore;
import org.apache.qpid.server.virtualhost.VirtualHost;
import static org.apache.qpid.server.logging.subjects.LogSubjectFormat.STORE_FORMAT;
@@ -28,10 +27,9 @@ import static org.apache.qpid.server.log
public class MessageStoreLogSubject extends AbstractLogSubject
{
- /** Create an ExchangeLogSubject that Logs in the following format. */
- public MessageStoreLogSubject(VirtualHost vhost, MessageStore store)
+ /** Create an MessageStoreLogSubject that Logs in the following format. */
+ public MessageStoreLogSubject(VirtualHost vhost, String messageStoreName)
{
- setLogStringWithFormat(STORE_FORMAT, vhost.getName(),
- store.getClass().getSimpleName());
+ setLogStringWithFormat(STORE_FORMAT, vhost.getName(), messageStoreName);
}
}
Propchange: qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/management/
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/management:r1307318-1307416
Modified: qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/management/AMQBrokerManagerMBean.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/management/AMQBrokerManagerMBean.java?rev=1327029&r1=1327028&r2=1327029&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/management/AMQBrokerManagerMBean.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/management/AMQBrokerManagerMBean.java Tue Apr 17 10:29:11 2012
@@ -56,8 +56,6 @@ public class AMQBrokerManagerMBean exten
private final QueueRegistry _queueRegistry;
private final ExchangeRegistry _exchangeRegistry;
private final ExchangeFactory _exchangeFactory;
- private final Exchange _defaultExchange;
- private final DurableConfigurationStore _durableConfig;
private final VirtualHostMBean _virtualHostMBean;
@@ -71,8 +69,6 @@ public class AMQBrokerManagerMBean exten
_queueRegistry = virtualHost.getQueueRegistry();
_exchangeRegistry = virtualHost.getExchangeRegistry();
- _defaultExchange = _exchangeRegistry.getDefaultExchange();
- _durableConfig = virtualHost.getMessageStore();
_exchangeFactory = virtualHost.getExchangeFactory();
}
@@ -177,7 +173,7 @@ public class AMQBrokerManagerMBean exten
_exchangeRegistry.registerExchange(exchange);
if (durable)
{
- _durableConfig.createExchange(exchange);
+ getVirtualHost().getMessageStore().createExchange(exchange);
}
}
else
@@ -271,10 +267,10 @@ public class AMQBrokerManagerMBean exten
false, false, getVirtualHost(), args);
if (queue.isDurable() && !queue.isAutoDelete())
{
- _durableConfig.createQueue(queue, args);
+ getVirtualHost().getMessageStore().createQueue(queue, args);
}
- virtualHost.getBindingFactory().addBinding(queueName, queue, _defaultExchange, null);
+ virtualHost.getBindingFactory().addBinding(queueName, queue, _exchangeRegistry.getDefaultExchange(), null);
}
catch (AMQException ex)
{
@@ -313,7 +309,7 @@ public class AMQBrokerManagerMBean exten
queue.delete();
if (queue.isDurable())
{
- _durableConfig.removeQueue(queue);
+ getVirtualHost().getMessageStore().removeQueue(queue);
}
}
catch (AMQException ex)
Propchange: qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/queue:r1307318-1307416
Modified: qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/DefaultQueueRegistry.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/DefaultQueueRegistry.java?rev=1327029&r1=1327028&r2=1327029&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/DefaultQueueRegistry.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/DefaultQueueRegistry.java Tue Apr 17 10:29:11 2012
@@ -20,7 +20,10 @@
*/
package org.apache.qpid.server.queue;
+import org.apache.log4j.Logger;
+import org.apache.qpid.AMQException;
import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.server.exchange.DefaultExchangeRegistry;
import org.apache.qpid.server.virtualhost.VirtualHost;
import java.util.ArrayList;
@@ -30,6 +33,8 @@ import java.util.concurrent.ConcurrentMa
public class DefaultQueueRegistry implements QueueRegistry
{
+ private static final Logger LOGGER = Logger.getLogger(DefaultExchangeRegistry.class);
+
private ConcurrentMap<AMQShortString, AMQQueue> _queueMap = new ConcurrentHashMap<AMQShortString, AMQQueue>();
private final VirtualHost _virtualHost;
@@ -100,4 +105,22 @@ public class DefaultQueueRegistry implem
_listeners.add(listener);
}
}
+
+ @Override
+ public void stopAllAndUnregisterMBeans()
+ {
+ for (final AMQQueue queue : getQueues())
+ {
+ queue.stop();
+ try
+ {
+ queue.getManagedObject().unregister();
+ }
+ catch (AMQException e)
+ {
+ LOGGER.warn("Failed to unregister mbean", e);
+ }
+ }
+ _queueMap.clear();
+ }
}
Modified: qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/QueueRegistry.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/QueueRegistry.java?rev=1327029&r1=1327028&r2=1327029&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/QueueRegistry.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/QueueRegistry.java Tue Apr 17 10:29:11 2012
@@ -43,6 +43,8 @@ public interface QueueRegistry
void addRegistryChangeListener(RegistryChangeListener listener);
+ void stopAllAndUnregisterMBeans();
+
interface RegistryChangeListener
{
void queueRegistered(AMQQueue queue);
Modified: qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/store/DurableConfigurationStore.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/store/DurableConfigurationStore.java?rev=1327029&r1=1327028&r2=1327029&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/store/DurableConfigurationStore.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/store/DurableConfigurationStore.java Tue Apr 17 10:29:11 2012
@@ -28,7 +28,6 @@ import org.apache.qpid.framing.FieldTabl
import org.apache.qpid.server.exchange.Exchange;
import org.apache.qpid.server.federation.Bridge;
import org.apache.qpid.server.federation.BrokerLink;
-import org.apache.qpid.server.logging.LogSubject;
import org.apache.qpid.server.queue.AMQQueue;
public interface DurableConfigurationStore
@@ -46,13 +45,11 @@ public interface DurableConfigurationSto
* @param name The name to be used by this store
* @param recoveryHandler Handler to be called as the store recovers on start up
* @param config The apache commons configuration object.
- *
* @throws Exception If any error occurs that means the store is unable to configure itself.
*/
void configureConfigStore(String name,
ConfigurationRecoveryHandler recoveryHandler,
- Configuration config,
- LogSubject logSubject) throws Exception;
+ Configuration config) throws Exception;
/**
* Makes the specified exchange persistent.
*
Modified: qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/store/MemoryMessageStore.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/store/MemoryMessageStore.java?rev=1327029&r1=1327028&r2=1327029&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/store/MemoryMessageStore.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/store/MemoryMessageStore.java Tue Apr 17 10:29:11 2012
@@ -20,103 +20,58 @@
*/
package org.apache.qpid.server.store;
-import org.apache.commons.configuration.Configuration;
-import org.apache.log4j.Logger;
-
-import org.apache.qpid.AMQException;
import org.apache.qpid.AMQStoreException;
-import org.apache.qpid.framing.AMQShortString;
-import org.apache.qpid.framing.FieldTable;
-import org.apache.qpid.server.exchange.Exchange;
-import org.apache.qpid.server.federation.Bridge;
-import org.apache.qpid.server.federation.BrokerLink;
-import org.apache.qpid.server.logging.LogSubject;
-import org.apache.qpid.server.logging.actors.CurrentActor;
-import org.apache.qpid.server.logging.messages.ConfigStoreMessages;
-import org.apache.qpid.server.logging.messages.MessageStoreMessages;
import org.apache.qpid.server.message.EnqueableMessage;
-import org.apache.qpid.server.queue.AMQQueue;
-import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
-/** A simple message store that stores the messages in a threadsafe structure in memory. */
-public class MemoryMessageStore implements MessageStore
+/** A simple message store that stores the messages in a thread-safe structure in memory. */
+public class MemoryMessageStore extends NullMessageStore
{
- private static final Logger _log = Logger.getLogger(MemoryMessageStore.class);
-
- private static final int DEFAULT_HASHTABLE_CAPACITY = 50000;
-
- private static final String HASHTABLE_CAPACITY_CONFIG = "hashtable-capacity";
-
-
private final AtomicLong _messageId = new AtomicLong(1);
- private AtomicBoolean _closed = new AtomicBoolean(false);
- private LogSubject _logSubject;
+ private final AtomicBoolean _closed = new AtomicBoolean(false);
private static final Transaction IN_MEMORY_TRANSACTION = new Transaction()
{
- public void enqueueMessage(TransactionLogResource queue, EnqueableMessage message) throws AMQStoreException
+ @Override
+ public StoreFuture commitTranAsync() throws AMQStoreException
{
+ return StoreFuture.IMMEDIATE_FUTURE;
}
- public void dequeueMessage(TransactionLogResource queue, EnqueableMessage message) throws AMQStoreException
+ @Override
+ public void enqueueMessage(TransactionLogResource queue, EnqueableMessage message) throws AMQStoreException
{
}
- public void commitTran() throws AMQStoreException
+ @Override
+ public void dequeueMessage(TransactionLogResource queue, EnqueableMessage message) throws AMQStoreException
{
}
- public StoreFuture commitTranAsync() throws AMQStoreException
+ @Override
+ public void commitTran() throws AMQStoreException
{
- return StoreFuture.IMMEDIATE_FUTURE;
}
+ @Override
public void abortTran() throws AMQStoreException
{
}
+ @Override
public void removeXid(long format, byte[] globalId, byte[] branchId)
{
}
+ @Override
public void recordXid(long format, byte[] globalId, byte[] branchId, Record[] enqueues, Record[] dequeues)
{
}
-
};
- public void configureConfigStore(String name, ConfigurationRecoveryHandler handler, Configuration configuration, LogSubject logSubject) throws Exception
- {
- _logSubject = logSubject;
- CurrentActor.get().message(_logSubject, ConfigStoreMessages.CREATED(this.getClass().getName()));
-
-
- }
-
- public void configureMessageStore(String name,
- MessageStoreRecoveryHandler recoveryHandler,
- TransactionLogRecoveryHandler tlogRecoveryHandler,
- Configuration config, LogSubject logSubject) throws Exception
- {
- if(_logSubject == null)
- {
- _logSubject = logSubject;
- }
- int hashtableCapacity = config.getInt(name + "." + HASHTABLE_CAPACITY_CONFIG, DEFAULT_HASHTABLE_CAPACITY);
- _log.info("Using capacity " + hashtableCapacity + " for hash tables");
- CurrentActor.get().message(_logSubject, MessageStoreMessages.CREATED(this.getClass().getName()));
- }
-
- public void close() throws Exception
- {
- _closed.getAndSet(true);
- CurrentActor.get().message(_logSubject,MessageStoreMessages.CLOSED());
-
- }
-
+ @Override
public StoredMessage addMessage(StorableMessageMetaData metaData)
{
final long id = _messageId.getAndIncrement();
@@ -125,96 +80,21 @@ public class MemoryMessageStore implemen
return message;
}
-
- public void createExchange(Exchange exchange) throws AMQStoreException
- {
-
- }
-
- public void removeExchange(Exchange exchange) throws AMQStoreException
- {
-
- }
-
- public void bindQueue(Exchange exchange, AMQShortString routingKey, AMQQueue queue, FieldTable args) throws AMQStoreException
- {
-
- }
-
- public void unbindQueue(Exchange exchange, AMQShortString routingKey, AMQQueue queue, FieldTable args) throws AMQStoreException
- {
-
- }
-
-
- public void createQueue(AMQQueue queue) throws AMQStoreException
- {
- // Not requred to do anything
- }
-
- public void createQueue(AMQQueue queue, FieldTable arguments) throws AMQStoreException
- {
- // Not required to do anything
- }
-
- public void removeQueue(final AMQQueue queue) throws AMQStoreException
- {
- // Not required to do anything
- }
-
- public void updateQueue(final AMQQueue queue) throws AMQStoreException
- {
- // Not required to do anything
- }
-
- public void createBrokerLink(final BrokerLink link) throws AMQStoreException
- {
-
- }
-
- public void deleteBrokerLink(final BrokerLink link) throws AMQStoreException
- {
-
- }
-
- public void createBridge(final Bridge bridge) throws AMQStoreException
- {
-
- }
-
- public void deleteBridge(final Bridge bridge) throws AMQStoreException
- {
-
- }
-
+ @Override
public Transaction newTransaction()
{
return IN_MEMORY_TRANSACTION;
}
-
- public List<AMQQueue> createQueues() throws AMQException
- {
- return null;
- }
-
- public Long getNewMessageId()
- {
- return _messageId.getAndIncrement();
- }
-
+ @Override
public boolean isPersistent()
{
return false;
}
- private void checkNotClosed() throws MessageStoreClosedException
- {
- if (_closed.get())
- {
- throw new MessageStoreClosedException();
- }
+ @Override
+ public void close() throws Exception
+ {
+ _closed.getAndSet(true);
}
-
-
}
Modified: qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/store/MessageStore.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/store/MessageStore.java?rev=1327029&r1=1327028&r2=1327029&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/store/MessageStore.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/store/MessageStore.java Tue Apr 17 10:29:11 2012
@@ -21,7 +21,6 @@
package org.apache.qpid.server.store;
import org.apache.commons.configuration.Configuration;
-import org.apache.qpid.server.logging.LogSubject;
/**
* MessageStore defines the interface to a storage area, which can be used to preserve the state of messages.
@@ -42,9 +41,9 @@ public interface MessageStore extends Du
void configureMessageStore(String name,
MessageStoreRecoveryHandler messageRecoveryHandler,
TransactionLogRecoveryHandler tlogRecoveryHandler,
- Configuration config, LogSubject logSubject) throws Exception;
-
+ Configuration config) throws Exception;
+ void activate() throws Exception;
public <T extends StorableMessageMetaData> StoredMessage<T> addMessage(T metaData);
@@ -65,4 +64,8 @@ public interface MessageStore extends Du
*/
void close() throws Exception;
+ void addEventListener(EventListener eventListener, Event event);
+
+ MessageStore getUnderlyingStore();
+
}
Modified: qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/store/StoreFuture.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/store/StoreFuture.java?rev=1327029&r1=1327028&r2=1327029&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/store/StoreFuture.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/store/StoreFuture.java Tue Apr 17 10:29:11 2012
@@ -31,11 +31,10 @@ public interface StoreFuture
public void waitForCompletion()
{
-
}
};
boolean isComplete();
void waitForCompletion();
-}
\ No newline at end of file
+}
Modified: qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnectionDelegate.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnectionDelegate.java?rev=1327029&r1=1327028&r2=1327029&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnectionDelegate.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnectionDelegate.java Tue Apr 17 10:29:11 2012
@@ -33,6 +33,7 @@ import org.apache.qpid.server.security.S
import org.apache.qpid.server.security.auth.AuthenticationResult;
import org.apache.qpid.server.security.auth.AuthenticationResult.AuthenticationStatus;
import org.apache.qpid.server.subscription.Subscription_0_10;
+import org.apache.qpid.server.virtualhost.State;
import org.apache.qpid.server.virtualhost.VirtualHost;
import org.apache.qpid.transport.*;
import org.slf4j.Logger;
@@ -177,6 +178,11 @@ public class ServerConnectionDelegate ex
sconn.setState(Connection.State.CLOSING);
sconn.invoke(new ConnectionClose(ConnectionCloseCode.CONNECTION_FORCED, "Permission denied '"+vhostName+"'"));
}
+ else if (vhost.getState() != State.ACTIVE)
+ {
+ sconn.setState(Connection.State.CLOSING);
+ sconn.invoke(new ConnectionClose(ConnectionCloseCode.CONNECTION_FORCED, "Virtual host '"+vhostName+"' is not active"));
+ }
else
{
sconn.setState(Connection.State.OPEN);
Propchange: qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost:r1307318-1307416
Modified: qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java?rev=1327029&r1=1327028&r2=1327029&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java Tue Apr 17 10:29:11 2012
@@ -97,4 +97,6 @@ public interface VirtualHost extends Dur
void removeBrokerConnection(BrokerLink brokerLink);
ScheduledFuture<?> scheduleTask(long delay, Runnable timeoutTask);
+
+ State getState();
}
Modified: qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHostConfigRecoveryHandler.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHostConfigRecoveryHandler.java?rev=1327029&r1=1327028&r2=1327029&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHostConfigRecoveryHandler.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHostConfigRecoveryHandler.java Tue Apr 17 10:29:11 2012
@@ -27,6 +27,7 @@ import java.util.HashMap;
import java.util.Map;
import java.util.TreeMap;
import java.util.UUID;
+
import org.apache.log4j.Logger;
import org.apache.qpid.AMQException;
import org.apache.qpid.AMQStoreException;
@@ -92,7 +93,7 @@ public class VirtualHostConfigRecoveryHa
public VirtualHostConfigRecoveryHandler begin(MessageStore store)
{
- _logSubject = new MessageStoreLogSubject(_virtualHost,store);
+ _logSubject = new MessageStoreLogSubject(_virtualHost,store.getClass().getSimpleName());
_store = store;
CurrentActor.get().message(_logSubject, TransactionLogMessages.RECOVERY_START(null, false));
@@ -355,31 +356,6 @@ public class VirtualHostConfigRecoveryHa
CurrentActor.get().message(_logSubject, TransactionLogMessages.RECOVERY_COMPLETE(null, false));
}
- private static final class ProcessAction
- {
- private final AMQQueue _queue;
- private final AMQMessage _message;
-
- public ProcessAction(AMQQueue queue, AMQMessage message)
- {
- _queue = queue;
- _message = message;
- }
-
- public void process()
- {
- try
- {
- _queue.enqueue(_message);
- }
- catch(AMQException e)
- {
- throw new RuntimeException(e);
- }
- }
-
- }
-
public void binding(String exchangeName, String queueName, String bindingKey, ByteBuffer buf)
{
try
Modified: qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHostImpl.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHostImpl.java?rev=1327029&r1=1327028&r2=1327029&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHostImpl.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHostImpl.java Tue Apr 17 10:29:11 2012
@@ -20,15 +20,11 @@
*/
package org.apache.qpid.server.virtualhost;
-import java.util.concurrent.ScheduledFuture;
-import org.apache.commons.configuration.Configuration;
+
import org.apache.commons.configuration.ConfigurationException;
import org.apache.log4j.Logger;
-
import org.apache.qpid.AMQException;
-import org.apache.qpid.AMQStoreException;
import org.apache.qpid.framing.AMQShortString;
-import org.apache.qpid.framing.FieldTable;
import org.apache.qpid.server.management.AMQBrokerManagerMBean;
import org.apache.qpid.server.binding.BindingFactory;
import org.apache.qpid.server.configuration.BrokerConfig;
@@ -45,9 +41,7 @@ import org.apache.qpid.server.exchange.D
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.federation.Bridge;
import org.apache.qpid.server.federation.BrokerLink;
-import org.apache.qpid.server.logging.LogSubject;
import org.apache.qpid.server.logging.actors.CurrentActor;
import org.apache.qpid.server.logging.messages.VirtualHostMessages;
import org.apache.qpid.server.logging.subjects.MessageStoreLogSubject;
@@ -62,13 +56,10 @@ import org.apache.qpid.server.queue.Queu
import org.apache.qpid.server.registry.IApplicationRegistry;
import org.apache.qpid.server.security.SecurityManager;
import org.apache.qpid.server.stats.StatisticsCounter;
-import org.apache.qpid.server.store.ConfigurationRecoveryHandler;
+import org.apache.qpid.server.store.Event;
+import org.apache.qpid.server.store.EventListener;
import org.apache.qpid.server.store.MessageStore;
-import org.apache.qpid.server.store.MessageStoreRecoveryHandler;
-import org.apache.qpid.server.store.StorableMessageMetaData;
-import org.apache.qpid.server.store.StoredMessage;
-import org.apache.qpid.server.store.Transaction;
-import org.apache.qpid.server.store.TransactionLogRecoveryHandler;
+import org.apache.qpid.server.store.MessageStoreFactory;
import org.apache.qpid.server.txn.DtxRegistry;
import org.apache.qpid.server.virtualhost.plugins.VirtualHostPlugin;
import org.apache.qpid.server.virtualhost.plugins.VirtualHostPluginFactory;
@@ -76,10 +67,14 @@ import org.apache.qpid.server.virtualhos
import java.util.List;
import java.util.Map;
import java.util.UUID;
+import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
+import javax.management.JMException;
+
public class VirtualHostImpl implements VirtualHost
{
private static final Logger _logger = Logger.getLogger(VirtualHostImpl.class);
@@ -102,29 +97,31 @@ public class VirtualHostImpl implements
private final BrokerConfig _brokerConfig;
- private final VirtualHostConfiguration _configuration;
+ private final VirtualHostConfiguration _vhostConfig;
- private ConnectionRegistry _connectionRegistry;
+ private final VirtualHostMBean _virtualHostMBean;
- private QueueRegistry _queueRegistry;
+ private final AMQBrokerManagerMBean _brokerMBean;
- private ExchangeRegistry _exchangeRegistry;
+ private final QueueRegistry _queueRegistry;
- private ExchangeFactory _exchangeFactory;
+ private final ExchangeRegistry _exchangeRegistry;
- private MessageStore _messageStore;
+ private final ExchangeFactory _exchangeFactory;
- private DtxRegistry _dtxRegistry;
+ private final ConnectionRegistry _connectionRegistry;
- private VirtualHostMBean _virtualHostMBean;
+ private final BindingFactory _bindingFactory;
- private AMQBrokerManagerMBean _brokerMBean;
+ private final DtxRegistry _dtxRegistry;
- private BindingFactory _bindingFactory;
+ private final MessageStore _messageStore;
+
+ private State _state = State.INITIALISING;
private boolean _statisticsEnabled = false;
- private StatisticsCounter _messagesDelivered, _dataDelivered, _messagesReceived, _dataReceived;
+ private StatisticsCounter _messagesDelivered, _dataDelivered, _messagesReceived, _dataReceived;
public VirtualHostImpl(IApplicationRegistry appRegistry, VirtualHostConfiguration hostConfig) throws Exception
{
@@ -133,53 +130,45 @@ public class VirtualHostImpl implements
throw new IllegalArgumentException("HostConfig cannot be null");
}
+ if (hostConfig.getName() == null || hostConfig.getName().length() == 0)
+ {
+ throw new IllegalArgumentException("Illegal name (" + hostConfig.getName() + ") for virtualhost.");
+ }
+
_appRegistry = appRegistry;
_brokerConfig = _appRegistry.getBroker();
- _configuration = hostConfig;
- _name = _configuration.getName();
+ _vhostConfig = hostConfig;
+ _name = _vhostConfig.getName();
_dtxRegistry = new DtxRegistry();
_id = _appRegistry.getConfigStore().createId();
CurrentActor.get().message(VirtualHostMessages.CREATED(_name));
- if (_name == null || _name.length() == 0)
- {
- throw new IllegalArgumentException("Illegal name (" + _name + ") for virtualhost.");
- }
-
- _securityManager = new SecurityManager(_appRegistry.getSecurityManager());
- _securityManager.configureHostPlugins(_configuration);
-
_virtualHostMBean = new VirtualHostMBean(this);
+ _securityManager = new SecurityManager(_appRegistry.getSecurityManager());
+ _securityManager.configureHostPlugins(_vhostConfig);
_connectionRegistry = new ConnectionRegistry();
- _houseKeepingTasks = new ScheduledThreadPoolExecutor(_configuration.getHouseKeepingThreadCount());
+ _houseKeepingTasks = new ScheduledThreadPoolExecutor(_vhostConfig.getHouseKeepingThreadCount());
_queueRegistry = new DefaultQueueRegistry(this);
_exchangeFactory = new DefaultExchangeFactory(this);
- _exchangeFactory.initialise(_configuration);
+ _exchangeFactory.initialise(_vhostConfig);
_exchangeRegistry = new DefaultExchangeRegistry(this);
- StartupRoutingTable configFileRT = new StartupRoutingTable();
-
- _messageStore = configFileRT;
-
- // This needs to be after the RT has been defined as it creates the default durable exchanges.
- _exchangeRegistry.initialise();
-
_bindingFactory = new BindingFactory(this);
- initialiseModel(_configuration);
+ _brokerMBean = new AMQBrokerManagerMBean(_virtualHostMBean);
- initialiseMessageStore(hostConfig);
+ _messageStore = initialiseMessageStore(hostConfig.getMessageStoreFactoryClass());
- _brokerMBean = new AMQBrokerManagerMBean(_virtualHostMBean);
- _brokerMBean.register();
- initialiseHouseKeeping(hostConfig.getHousekeepingCheckPeriod());
+ configureMessageStore(hostConfig);
+
+ activateNonHAMessageStore();
initialiseStatistics();
}
@@ -191,7 +180,7 @@ public class VirtualHostImpl implements
public VirtualHostConfiguration getConfiguration()
{
- return _configuration;
+ return _vhostConfig;
}
public UUID getId()
@@ -220,12 +209,11 @@ public class VirtualHostImpl implements
*
* @param period
*/
- private void initialiseHouseKeeping(long period)
+ private void initialiseHouseKeeping(long period)
{
+
if (period != 0L)
{
-
-
scheduleHouseKeepingTask(period, new VirtualHostHouseKeepingTask());
Map<String, VirtualHostPluginFactory> plugins = _appRegistry.getPluginManager().getVirtualHostPlugins();
@@ -258,50 +246,30 @@ public class VirtualHostImpl implements
}
}
- private class VirtualHostHouseKeepingTask extends HouseKeepingTask
+ private void shutdownHouseKeeping()
{
- public VirtualHostHouseKeepingTask()
- {
- super(VirtualHostImpl.this);
- }
+ _houseKeepingTasks.shutdown();
- public void execute()
+ try
+ {
+ if (!_houseKeepingTasks.awaitTermination(HOUSEKEEPING_SHUTDOWN_TIMEOUT, TimeUnit.SECONDS))
+ {
+ _houseKeepingTasks.shutdownNow();
+ }
+ }
+ catch (InterruptedException e)
+ {
+ _logger.warn("Interrupted during Housekeeping shutdown:", e);
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ private void removeHouseKeepingTasks()
+ {
+ BlockingQueue<Runnable> taskQueue = _houseKeepingTasks.getQueue();
+ for (final Runnable runnable : taskQueue)
{
- for (AMQQueue q : _queueRegistry.getQueues())
- {
- _logger.debug("Checking message status for queue: "
- + q.getName());
- try
- {
- q.checkMessageStatus();
- }
- catch (Exception e)
- {
- _logger.error("Exception in housekeeping for queue: "
- + q.getNameShortString().toString(), e);
- //Don't throw exceptions as this will stop the
- // house keeping task from running.
- }
- }
- for (AMQConnectionModel connection : getConnectionRegistry().getConnections())
- {
- _logger.debug("Checking for long running open transactions on connection " + connection);
- for (AMQSessionModel session : connection.getSessionModels())
- {
- _logger.debug("Checking for long running open transactions on session " + session);
- try
- {
- session.checkTransactionStatus(_configuration.getTransactionTimeoutOpenWarn(),
- _configuration.getTransactionTimeoutOpenClose(),
- _configuration.getTransactionTimeoutIdleWarn(),
- _configuration.getTransactionTimeoutIdleClose());
- }
- catch (Exception e)
- {
- _logger.error("Exception in housekeeping for connection: " + connection.toString(), e);
- }
- }
- }
+ _houseKeepingTasks.remove(runnable);
}
}
@@ -349,36 +317,43 @@ public class VirtualHostImpl implements
}
- private void initialiseMessageStore(VirtualHostConfiguration hostConfig) throws Exception
+ private MessageStore initialiseMessageStore(final String messageStoreFactoryClass) throws Exception
{
- String messageStoreClass = hostConfig.getMessageStoreClass();
-
- Class<?> clazz = Class.forName(messageStoreClass);
- Object o = clazz.newInstance();
+ final Class<?> clazz = Class.forName(messageStoreFactoryClass);
+ final Object o = clazz.newInstance();
- if (!(o instanceof MessageStore))
+ if (!(o instanceof MessageStoreFactory))
{
- throw new ClassCastException("Message store class must implement " + MessageStore.class + ". Class " + clazz +
- " does not.");
+ throw new ClassCastException("Message store factory class must implement " + MessageStoreFactory.class +
+ ". Class " + clazz + " does not.");
}
- MessageStore messageStore = (MessageStore) o;
- VirtualHostConfigRecoveryHandler recoveryHandler = new VirtualHostConfigRecoveryHandler(this);
- MessageStoreLogSubject storeLogSubject = new MessageStoreLogSubject(this, messageStore);
+ final MessageStoreFactory messageStoreFactory = (MessageStoreFactory) o;
+ final MessageStoreLogSubject storeLogSubject = new MessageStoreLogSubject(this, messageStoreFactory.getStoreClassName());
+ final MessageStore messageStore = messageStoreFactory.createMessageStore(storeLogSubject);
- messageStore.configureConfigStore(this.getName(),
- recoveryHandler,
- hostConfig.getStoreConfiguration(),
- storeLogSubject);
-
- messageStore.configureMessageStore(this.getName(),
- recoveryHandler,
- recoveryHandler,
- hostConfig.getStoreConfiguration(), storeLogSubject);
+ messageStore.addEventListener(new BeforeActivationListener(), Event.BEFORE_ACTIVATE);
+ messageStore.addEventListener(new AfterActivationListener(), Event.AFTER_ACTIVATE);
+ messageStore.addEventListener(new BeforeCloseListener(), Event.BEFORE_CLOSE);
+ messageStore.addEventListener(new BeforePassivationListener(), Event.BEFORE_PASSIVATE);
- _messageStore = messageStore;
+ return messageStore;
+ }
+ private void configureMessageStore(VirtualHostConfiguration hostConfig) throws Exception
+ {
+
+ VirtualHostConfigRecoveryHandler recoveryHandler = new VirtualHostConfigRecoveryHandler(this);
+ // TODO perhaps pass config on construction??
+ _messageStore.configureConfigStore(getName(), recoveryHandler, hostConfig.getStoreConfiguration());
+ _messageStore.configureMessageStore(getName(), recoveryHandler, recoveryHandler, hostConfig.getStoreConfiguration());
+
+ }
+
+ private void activateNonHAMessageStore() throws Exception
+ {
+ _messageStore.activate();
}
private void initialiseModel(VirtualHostConfiguration config) throws ConfigurationException, AMQException
@@ -392,7 +367,7 @@ public class VirtualHostImpl implements
configureExchange(config.getExchangeConfiguration(exchangeName));
}
- String[] queueNames = config.getQueueNames();
+ String[] queueNames = config.getQueueNames();
for (Object queueNameObj : queueNames)
{
@@ -403,16 +378,16 @@ public class VirtualHostImpl implements
private void configureExchange(ExchangeConfiguration exchangeConfiguration) throws AMQException
{
- AMQShortString exchangeName = new AMQShortString(exchangeConfiguration.getName());
+ AMQShortString exchangeName = new AMQShortString(exchangeConfiguration.getName());
Exchange exchange;
exchange = _exchangeRegistry.getExchange(exchangeName);
if (exchange == null)
{
- AMQShortString type = new AMQShortString(exchangeConfiguration.getType());
- boolean durable = exchangeConfiguration.getDurable();
- boolean autodelete = exchangeConfiguration.getAutoDelete();
+ AMQShortString type = new AMQShortString(exchangeConfiguration.getType());
+ boolean durable = exchangeConfiguration.getDurable();
+ boolean autodelete = exchangeConfiguration.getAutoDelete();
Exchange newExchange = _exchangeFactory.createExchange(exchangeName, type, durable, autodelete, 0);
_exchangeRegistry.registerExchange(newExchange);
@@ -426,7 +401,7 @@ public class VirtualHostImpl implements
private void configureQueue(QueueConfiguration queueConfiguration) throws AMQException, ConfigurationException
{
- AMQQueue queue = AMQQueueFactory.createAMQQueueImpl(queueConfiguration, this);
+ AMQQueue queue = AMQQueueFactory.createAMQQueueImpl(queueConfiguration, this);
String queueName = queue.getName();
if (queue.isDurable())
@@ -435,13 +410,13 @@ public class VirtualHostImpl implements
}
//get the exchange name (returns default exchange name if none was specified)
- String exchangeName = queueConfiguration.getExchange();
+ String exchangeName = queueConfiguration.getExchange();
Exchange exchange = _exchangeRegistry.getExchange(exchangeName);
- if (exchange == null)
- {
+ if (exchange == null)
+ {
throw new ConfigurationException("Attempt to bind queue '" + queueName + "' to unknown exchange:" + exchangeName);
- }
+ }
Exchange defaultExchange = _exchangeRegistry.getDefaultExchange();
@@ -529,39 +504,8 @@ public class VirtualHostImpl implements
{
//Stop Connections
_connectionRegistry.close();
-
- //Stop the Queues processing
- if (_queueRegistry != null)
- {
- for (AMQQueue queue : _queueRegistry.getQueues())
- {
- queue.stop();
- }
- }
-
- //Stop Housekeeping
- if (_houseKeepingTasks != null)
- {
- _houseKeepingTasks.shutdown();
-
- try
- {
- if (!_houseKeepingTasks.awaitTermination(HOUSEKEEPING_SHUTDOWN_TIMEOUT, TimeUnit.SECONDS))
- {
- _houseKeepingTasks.shutdownNow();
- }
- }
- catch (InterruptedException e)
- {
- _logger.warn("Interrupted during Housekeeping shutdown:" + e.getMessage());
- // Swallowing InterruptedException ok as we are shutting down.
- }
- }
-
- if(_dtxRegistry != null)
- {
- _dtxRegistry.close();
- }
+ _queueRegistry.stopAllAndUnregisterMBeans();
+ _dtxRegistry.close();
//Close MessageStore
if (_messageStore != null)
@@ -577,6 +521,8 @@ public class VirtualHostImpl implements
}
}
+ _state = State.STOPPED;
+
CurrentActor.get().message(VirtualHostMessages.CLOSED());
}
@@ -688,7 +634,6 @@ public class VirtualHostImpl implements
return blink;
}
-
public void createBrokerConnection(final String transport,
final String host,
final int port,
@@ -735,105 +680,135 @@ public class VirtualHostImpl implements
return _dtxRegistry;
}
- /**
- * Temporary Startup RT class to record the creation of persistent queues / exchanges.
- *
- *
- * This is so we can replay the creation of queues/exchanges in to the real _RT after it has been loaded.
- * This should be removed after the _RT has been fully split from the the TL
- */
- private static class StartupRoutingTable implements MessageStore
+ @Override
+ public String toString()
{
- public void configureConfigStore(String name,
- ConfigurationRecoveryHandler recoveryHandler,
- Configuration config,
- LogSubject logSubject) throws Exception
- {
- }
-
- public void createExchange(Exchange exchange) throws AMQStoreException
- {
- }
-
- public void removeExchange(Exchange exchange) throws AMQStoreException
- {
- }
-
- public void bindQueue(Exchange exchange, AMQShortString routingKey, AMQQueue queue, FieldTable args) throws AMQStoreException
- {
- }
-
- public void unbindQueue(Exchange exchange, AMQShortString routingKey, AMQQueue queue, FieldTable args) throws AMQStoreException
- {
- }
-
- public void createQueue(AMQQueue queue) throws AMQStoreException
- {
- }
-
- public void createQueue(AMQQueue queue, FieldTable arguments) throws AMQStoreException
- {
- }
-
- public void removeQueue(AMQQueue queue) throws AMQStoreException
- {
- }
-
- public void updateQueue(AMQQueue queue) throws AMQStoreException
- {
- }
-
- public void createBrokerLink(final BrokerLink link) throws AMQStoreException
- {
- }
-
- public void deleteBrokerLink(final BrokerLink link) throws AMQStoreException
- {
- }
+ return _name;
+ }
- public void createBridge(final Bridge bridge) throws AMQStoreException
- {
- }
+ @Override
+ public State getState()
+ {
+ return _state;
+ }
- public void deleteBridge(final Bridge bridge) throws AMQStoreException
- {
- }
+ private final class BeforeActivationListener implements EventListener
+ {
+ @Override
+ public void event(Event event)
+ {
+ try
+ {
+ _exchangeRegistry.initialise();
+ initialiseModel(_vhostConfig);
+ }
+ catch (Exception e)
+ {
+ throw new RuntimeException("Failed to initialise virtual host after state change", e);
+ }
+ }
+ }
+
+ private final class AfterActivationListener implements EventListener
+ {
+ @Override
+ public void event(Event event)
+ {
+ initialiseHouseKeeping(_vhostConfig.getHousekeepingCheckPeriod());
+ try
+ {
+ _brokerMBean.register();
+ }
+ catch (JMException e)
+ {
+ throw new RuntimeException("Failed to register virtual host mbean for virtual host " + getName(), e);
+ }
+
+ _state = State.ACTIVE;
+ }
+ }
- @Override
- public void configureMessageStore(String name,
- MessageStoreRecoveryHandler recoveryHandler,
- TransactionLogRecoveryHandler tlogRecoveryHandler, Configuration config, LogSubject logSubject) throws Exception
- {
- }
-
- @Override
- public void close() throws Exception
- {
- }
+ public class BeforePassivationListener implements EventListener
+ {
@Override
- public <T extends StorableMessageMetaData> StoredMessage<T> addMessage(
- T metaData)
+ public void event(Event event)
{
- return null;
- }
+ _connectionRegistry.close(IConnectionRegistry.VHOST_PASSIVATE_REPLY_TEXT);
+ _brokerMBean.unregister();
+ removeHouseKeepingTasks();
- @Override
- public boolean isPersistent()
- {
- return false;
- }
+ _queueRegistry.stopAllAndUnregisterMBeans();
+ _exchangeRegistry.clearAndUnregisterMbeans();
+ _dtxRegistry.close();
- @Override
- public Transaction newTransaction()
- {
- return null;
+ _state = State.PASSIVE;
}
}
- @Override
- public String toString()
- {
- return _name;
- }
+ private final class BeforeCloseListener implements EventListener
+ {
+ @Override
+ public void event(Event event)
+ {
+ _brokerMBean.unregister();
+ shutdownHouseKeeping();
+ }
+ }
+
+ private class VirtualHostHouseKeepingTask extends HouseKeepingTask
+ {
+ public VirtualHostHouseKeepingTask()
+ {
+ super(VirtualHostImpl.this);
+ }
+
+ public void execute()
+ {
+ for (AMQQueue q : _queueRegistry.getQueues())
+ {
+ if (_logger.isDebugEnabled())
+ {
+ _logger.debug("Checking message status for queue: "
+ + q.getName());
+ }
+ try
+ {
+ q.checkMessageStatus();
+ }
+ catch (Exception e)
+ {
+ _logger.error("Exception in housekeeping for queue: "
+ + q.getNameShortString().toString(), e);
+ //Don't throw exceptions as this will stop the
+ // house keeping task from running.
+ }
+ }
+ for (AMQConnectionModel connection : getConnectionRegistry().getConnections())
+ {
+ if (_logger.isDebugEnabled())
+ {
+ _logger.debug("Checking for long running open transactions on connection " + connection);
+ }
+ for (AMQSessionModel session : connection.getSessionModels())
+ {
+ if (_logger.isDebugEnabled())
+ {
+ _logger.debug("Checking for long running open transactions on session " + session);
+ }
+ try
+ {
+ session.checkTransactionStatus(_vhostConfig.getTransactionTimeoutOpenWarn(),
+ _vhostConfig.getTransactionTimeoutOpenClose(),
+ _vhostConfig.getTransactionTimeoutIdleWarn(),
+ _vhostConfig.getTransactionTimeoutIdleClose());
+ }
+ catch (Exception e)
+ {
+ _logger.error("Exception in housekeeping for connection: " + connection.toString(), e);
+ }
+ }
+ }
+ }
+ }
}
Modified: qpid/branches/java-config-and-management/qpid/java/broker/src/test/java/org/apache/qpid/server/AMQBrokerManagerMBeanTest.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/broker/src/test/java/org/apache/qpid/server/AMQBrokerManagerMBeanTest.java?rev=1327029&r1=1327028&r2=1327029&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/broker/src/test/java/org/apache/qpid/server/AMQBrokerManagerMBeanTest.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/broker/src/test/java/org/apache/qpid/server/AMQBrokerManagerMBeanTest.java Tue Apr 17 10:29:11 2012
@@ -39,6 +39,7 @@ import org.apache.qpid.server.queue.Queu
import org.apache.qpid.server.registry.ApplicationRegistry;
import org.apache.qpid.server.registry.IApplicationRegistry;
import org.apache.qpid.server.store.TestableMemoryMessageStore;
+import org.apache.qpid.server.store.TestableMemoryMessageStoreFactory;
import org.apache.qpid.server.util.TestApplicationRegistry;
import org.apache.qpid.server.virtualhost.VirtualHost;
import org.apache.qpid.test.utils.QpidTestCase;
@@ -166,7 +167,7 @@ public class AMQBrokerManagerMBeanTest e
XMLConfiguration configXml = new XMLConfiguration();
configXml.addProperty("virtualhosts.virtualhost(-1).name", "test");
- configXml.addProperty("virtualhosts.virtualhost(-1).test.store.class", TestableMemoryMessageStore.class.getName());
+ configXml.addProperty("virtualhosts.virtualhost(-1).test.store.factoryclass", TestableMemoryMessageStoreFactory.class.getName());
ServerConfiguration configuration = new ServerConfiguration(configXml);
Modified: qpid/branches/java-config-and-management/qpid/java/broker/src/test/java/org/apache/qpid/server/configuration/VirtualHostConfigurationTest.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/broker/src/test/java/org/apache/qpid/server/configuration/VirtualHostConfigurationTest.java?rev=1327029&r1=1327028&r2=1327029&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/broker/src/test/java/org/apache/qpid/server/configuration/VirtualHostConfigurationTest.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/broker/src/test/java/org/apache/qpid/server/configuration/VirtualHostConfigurationTest.java Tue Apr 17 10:29:11 2012
@@ -27,6 +27,7 @@ import org.apache.qpid.server.queue.AMQP
import org.apache.qpid.server.queue.AMQQueue;
import org.apache.qpid.server.registry.ApplicationRegistry;
import org.apache.qpid.server.store.TestableMemoryMessageStore;
+import org.apache.qpid.server.store.TestableMemoryMessageStoreFactory;
import org.apache.qpid.server.util.InternalBrokerBaseCase;
import org.apache.qpid.server.virtualhost.VirtualHost;
@@ -161,7 +162,7 @@ public class VirtualHostConfigurationTes
getConfigXml().addProperty("virtualhosts.virtualhost." + getName() + "Extra.queues(-1).queue(-1).name", "r2d2");
getConfigXml().addProperty("virtualhosts.virtualhost." + getName() + "Extra.queues.queue.r2d2.deadLetterQueues", "true");
getConfigXml().addProperty("virtualhosts.virtualhost." + getName() + "Extra.queues(-1).queue(-1).name", "c3p0");
- getConfigXml().addProperty("virtualhosts.virtualhost." + getName() + "Extra.store.class", TestableMemoryMessageStore.class.getName());
+ getConfigXml().addProperty("virtualhosts.virtualhost." + getName() + "Extra.store.factoryclass", TestableMemoryMessageStoreFactory.class.getName());
// Start the broker now.
super.createBroker();
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org