You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ra...@apache.org on 2007/05/02 18:49:05 UTC
svn commit: r534541 [3/3] - in /incubator/qpid/trunk/qpid/java: broker/etc/
broker/src/main/java/org/apache/qpid/server/
broker/src/main/java/org/apache/qpid/server/configuration/
broker/src/main/java/org/apache/qpid/server/exception/ broker/src/main/j...
Added: incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/txn/TransactionRecord.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/txn/TransactionRecord.java?view=auto&rev=534541
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/txn/TransactionRecord.java (added)
+++ incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/txn/TransactionRecord.java Wed May 2 09:49:03 2007
@@ -0,0 +1,72 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.qpid.server.txn;
+
+import org.apache.qpid.server.exception.*;
+import org.apache.qpid.server.messageStore.MessageStore;
+
+import javax.transaction.xa.Xid;
+
+/**
+ * Created by Arnaud Simon
+ * Date: 25-Apr-2007
+ * Time: 14:12:17
+ */
+public interface TransactionRecord
+{
+ /**
+ * Commit this record.
+ *
+ * @param store the store to be used during commit
+ * @param xid the xid of the tx branch
+ * @throws org.apache.qpid.server.exception.InternalErrorException in case of internal problem
+ * @throws org.apache.qpid.server.exception.QueueDoesntExistException the queue does not exist
+ * @throws org.apache.qpid.server.exception.InvalidXidException the xid is invalid
+ * @throws org.apache.qpid.server.exception.UnknownXidException the xid is unknonw
+ * @throws org.apache.qpid.server.exception.MessageDoesntExistException the message does not exist
+ */
+ public abstract void commit(MessageStore store, Xid xid)
+ throws
+ InternalErrorException,
+ QueueDoesntExistException,
+ InvalidXidException,
+ UnknownXidException,
+ MessageDoesntExistException;
+
+ /**
+ * rollback this record
+ *
+ * @param store the store to be used
+ * @throws InternalErrorException In case of internal error
+ */
+ public abstract void rollback(MessageStore store)
+ throws
+ InternalErrorException;
+
+ /**
+ * Prepare this record
+ *
+ * @param store the store to be used
+ * @throws InternalErrorException In case of internal error
+ */
+ public abstract void prepare(MessageStore store)
+ throws
+ InternalErrorException;
+
+}
Added: incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/txn/XAFlag.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/txn/XAFlag.java?view=auto&rev=534541
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/txn/XAFlag.java (added)
+++ incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/txn/XAFlag.java Wed May 2 09:49:03 2007
@@ -0,0 +1,50 @@
+package org.apache.qpid.server.txn;
+
+/**
+ * Created by Arnaud Simon
+ * Date: 29-Mar-2007
+ * Time: 14:57:01
+ */
+public enum XAFlag
+{
+ rbrollback(1, "XA-RBROLLBACK", "The rollback was caused by an unspecified reason"),
+ rbtimeout(2, "XA-RBTIMEOUT", "The transaction branch took too long"),
+ heurhaz(3, "XA-HEURHAZ", "The transaction branch may have been heuristically completed"),
+ heurcom(4, "XA-HEURCOM", "The transaction branch has been heuristically committed"),
+ heurrb(5, "XA-HEURRB", "The transaction branch has been heuristically rolled back"),
+ heurmix(6, "XA-HEURMIX", "The transaction branch has been heuristically committed and rolled back"),
+ rdonly(7, "XA-RDONLY", "The transaction branch was read-only and has been committed"),
+ ok(8, "XA-OK", "Normal execution");
+
+ private final int _code;
+
+ private final String _name;
+
+ private final String _description;
+
+ XAFlag(int code, String name, String description)
+ {
+ _code = code;
+ _name = name;
+ _description = description;
+ }
+
+ //==============================================
+ // Getter methods
+ //==============================================
+
+ public int getCode()
+ {
+ return _code;
+ }
+
+ public String getName()
+ {
+ return _name;
+ }
+
+ public String getDescription()
+ {
+ return _description;
+ }
+}
Added: incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/txn/XidImpl.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/txn/XidImpl.java?view=auto&rev=534541
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/txn/XidImpl.java (added)
+++ incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/txn/XidImpl.java Wed May 2 09:49:03 2007
@@ -0,0 +1,210 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.qpid.server.txn;
+
+
+import org.apache.log4j.Logger;
+
+import javax.transaction.xa.Xid;
+
+/**
+ * Created by Arnaud Simon
+ * Date: 03-Apr-2007
+ * Time: 20:32:55
+ */
+public class XidImpl implements Xid
+{
+ //========================================================================
+ // Static Constants
+ //========================================================================
+ // The logger for this class
+ private static final Logger _log = Logger.getLogger(XidImpl.class);
+
+ //========================================================================
+ // Instance Fields
+ //========================================================================
+
+ //the transaction branch identifier part of XID as an array of bytes
+ private byte[] m_branchQualifier;
+
+ // the format identifier part of the XID.
+ private int m_formatID;
+
+ // the global transaction identifier part of XID as an array of bytes.
+ private byte[] m_globalTransactionID;
+
+ //========================================================================
+ // Constructor(s)
+ //========================================================================
+
+ /**
+ * Create new Xid.
+ */
+ public XidImpl()
+ {
+
+ }
+
+ /**
+ * Create new XidImpl from an existing Xid.
+ * <p/>
+ * This is usually called when an application server provides some implementation
+ * of the Xid interface and we need to cast this into our own XidImpl.
+ *
+ * @param xid the xid to cloning
+ */
+ public XidImpl(Xid xid)
+ {
+ if (_log.isDebugEnabled())
+ {
+ _log.debug("Cloning Xid");
+ }
+ m_branchQualifier = xid.getBranchQualifier();
+ m_formatID = xid.getFormatId();
+ m_globalTransactionID = xid.getGlobalTransactionId();
+ }
+
+ /**
+ * Create a new Xid.
+ *
+ * @param branchQualifier The transaction branch identifier part of XID as an array of bytes.
+ * @param format The format identifier part of the XID.
+ * @param globalTransactionID The global transaction identifier part of XID as an array of bytes.
+ */
+ public XidImpl(byte[] branchQualifier, int format, byte[] globalTransactionID)
+ {
+ if (_log.isDebugEnabled())
+ {
+ _log.debug("creating Xid");
+ }
+ m_branchQualifier = branchQualifier;
+ m_formatID = format;
+ m_globalTransactionID = globalTransactionID;
+ }
+
+//========================================================================
+
+ // Xid interface implementation
+ //========================================================================
+ /**
+ * Format identifier. O means the OSI CCR format.
+ *
+ * @return Global transaction identifier.
+ */
+ public byte[] getGlobalTransactionId()
+ {
+ return m_globalTransactionID;
+ }
+
+ /**
+ * Obtain the transaction branch identifier part of XID as an array of bytes.
+ *
+ * @return Branch identifier part of XID.
+ */
+ public byte[] getBranchQualifier()
+ {
+ return m_branchQualifier;
+ }
+
+ /**
+ * Obtain the format identifier part of the XID.
+ *
+ * @return Format identifier. O means the OSI CCR format.
+ */
+ public int getFormatId()
+ {
+ return m_formatID;
+ }
+
+//========================================================================
+// Object operations
+//========================================================================
+
+ /**
+ * Indicates whether some other Xid is "equal to" this one.
+ * <p/>
+ * Two Xids are equal if and only if their three elementary parts are equal
+ *
+ * @param o the object to compare this <code>XidImpl</code> against.
+ * @return code>true</code> if the <code>XidImpl</code> are equal; <code>false</code> otherwise.
+ */
+ public boolean equals(Object o)
+ {
+ if (this == o)
+ {
+ return true;
+ }
+ if (o instanceof XidImpl)
+ {
+ XidImpl other = (XidImpl) o;
+ if (m_formatID == other.getFormatId())
+ {
+ if (m_branchQualifier.length == other.getBranchQualifier().length)
+ {
+ for (int i = 0; i < m_branchQualifier.length; i++)
+ {
+ if (m_branchQualifier[i] != other.getBranchQualifier()[i])
+ {
+ return false;
+ }
+ }
+
+ if (m_globalTransactionID.length == other.getGlobalTransactionId().length)
+ {
+ for (int i = 0; i < m_globalTransactionID.length; i++)
+ {
+ if (m_globalTransactionID[i] != other.getGlobalTransactionId()[i])
+ {
+ return false;
+ }
+ }
+ // everithing is equal
+ return true;
+ }
+ }
+ }
+ }
+ return false;
+ }
+
+ /**
+ * Returns a hash code for this Xid.
+ * <p/>
+ * As this object is used as a key entry in a hashMap it is necessary to provide an implementation
+ * of hashcode in order to fulfill the following aspect of the general contract of
+ * {@link Object#hashCode()} that is:
+ * <ul>
+ * <li>If two objects are equal according to the <tt>equals(Object)</tt>
+ * method, then calling the <code>hashCode</code> method on each of
+ * the two objects must produce the same integer result.
+ * </ul>
+ * <p/>
+ * The hash code for a
+ * <code>XidImpl</code> object is computed as
+ * <blockquote><pre>
+ * hashcode( globalTransactionID ) + hashcode( branchQualifier ) + formatID
+ * </pre></blockquote>
+ *
+ * @return a hash code value for this object.
+ */
+ public int hashCode()
+ {
+ return (new String(m_globalTransactionID)).hashCode() + (new String(m_branchQualifier)).hashCode() + m_formatID;
+ }
+
+}
Modified: incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/util/NullApplicationRegistry.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/util/NullApplicationRegistry.java?view=diff&rev=534541&r1=534540&r2=534541
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/util/NullApplicationRegistry.java (original)
+++ incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/util/NullApplicationRegistry.java Wed May 2 09:49:03 2007
@@ -59,7 +59,8 @@
public void initialise() throws Exception
{
- _configuration.addProperty("store.class", "org.apache.qpid.server.store.MemoryMessageStore");
+ _configuration.addProperty("store.class", "org.apache.qpid.server.messageStore.MemoryMessageStore");
+ _configuration.addProperty("txn.class", "org.apache.qpid.server.txn.MemoryTransactionManager");
Properties users = new Properties();
Modified: incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java?view=diff&rev=534541&r1=534540&r2=534541
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java (original)
+++ incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java Wed May 2 09:49:03 2007
@@ -1,249 +1,272 @@
-/*
- *
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- *
- */
-package org.apache.qpid.server.virtualhost;
-
-import javax.management.NotCompliantMBeanException;
-
-import org.apache.commons.configuration.Configuration;
-import org.apache.log4j.Logger;
-import org.apache.qpid.server.AMQBrokerManagerMBean;
-import org.apache.qpid.server.security.access.AccessManager;
-import org.apache.qpid.server.security.access.AccessManagerImpl;
-import org.apache.qpid.server.security.access.Accessable;
-import org.apache.qpid.server.security.auth.manager.PrincipalDatabaseAuthenticationManager;
-import org.apache.qpid.server.security.auth.manager.AuthenticationManager;
-import org.apache.qpid.server.configuration.Configurator;
-import org.apache.qpid.server.exchange.DefaultExchangeFactory;
-import org.apache.qpid.server.exchange.DefaultExchangeRegistry;
-import org.apache.qpid.server.exchange.ExchangeFactory;
-import org.apache.qpid.server.exchange.ExchangeRegistry;
-import org.apache.qpid.server.management.AMQManagedObject;
-import org.apache.qpid.server.management.ManagedObject;
-import org.apache.qpid.server.queue.DefaultQueueRegistry;
-import org.apache.qpid.server.queue.QueueRegistry;
-import org.apache.qpid.server.registry.ApplicationRegistry;
-import org.apache.qpid.server.store.MessageStore;
-
-public class VirtualHost implements Accessable
-{
- private static final Logger _logger = Logger.getLogger(VirtualHost.class);
-
-
- private final String _name;
-
- private QueueRegistry _queueRegistry;
-
- private ExchangeRegistry _exchangeRegistry;
-
- private ExchangeFactory _exchangeFactory;
-
- private MessageStore _messageStore;
-
- protected VirtualHostMBean _virtualHostMBean;
-
- private AMQBrokerManagerMBean _brokerMBean;
-
- private AuthenticationManager _authenticationManager;
-
- private AccessManager _accessManager;
-
-
- public void setAccessableName(String name)
- {
- _logger.warn("Setting Accessable Name for VirualHost is not allowed. ("
- + name + ") ignored remains :" + getAccessableName());
- }
-
- public String getAccessableName()
- {
- return _name;
- }
-
-
- /**
- * Abstract MBean class. This has some of the methods implemented from management intrerface for exchanges. Any
- * implementaion of an Exchange MBean should extend this class.
- */
- public class VirtualHostMBean extends AMQManagedObject implements ManagedVirtualHost
- {
- public VirtualHostMBean() throws NotCompliantMBeanException
- {
- super(ManagedVirtualHost.class, "VirtualHost");
- }
-
- public String getObjectInstanceName()
- {
- return _name.toString();
- }
-
- public String getName()
- {
- return _name.toString();
- }
-
- public VirtualHost getVirtualHost()
- {
- return VirtualHost.this;
- }
-
-
- } // End of MBean class
-
-
- public VirtualHost(String name, MessageStore store) throws Exception
- {
- this(name, null, store);
- }
-
- public VirtualHost(String name, Configuration hostConfig) throws Exception
- {
- this(name, hostConfig, null);
- }
-
- private VirtualHost(String name, Configuration hostConfig, MessageStore store) throws Exception
- {
- _name = name;
-
- _virtualHostMBean = new VirtualHostMBean();
- // This isn't needed to be registered
- //_virtualHostMBean.register();
-
- _queueRegistry = new DefaultQueueRegistry(this);
- _exchangeFactory = new DefaultExchangeFactory(this);
- _exchangeRegistry = new DefaultExchangeRegistry(this);
-
- if (store != null)
- {
- _messageStore = store;
- }
- else
- {
- if (hostConfig == null)
- {
- throw new IllegalAccessException("HostConfig and MessageStore cannot be null");
- }
- initialiseMessageStore(hostConfig);
- }
-
- _exchangeRegistry.initialise();
-
- _logger.warn("VirtualHost authentication Managers require spec change to be operational.");
- _authenticationManager = new PrincipalDatabaseAuthenticationManager(name, hostConfig);
-
- _accessManager = new AccessManagerImpl(name, hostConfig);
-
- _brokerMBean = new AMQBrokerManagerMBean(_virtualHostMBean);
- _brokerMBean.register();
- }
-
- private void initialiseMessageStore(Configuration config) throws Exception
- {
- String messageStoreClass = config.getString("store.class");
-
- Class clazz = Class.forName(messageStoreClass);
- Object o = clazz.newInstance();
-
- if (!(o instanceof MessageStore))
- {
- throw new ClassCastException("Message store class must implement " + MessageStore.class + ". Class " + clazz +
- " does not.");
- }
- _messageStore = (MessageStore) o;
- _messageStore.configure(this, "store", config);
- }
-
-
- public <T> T getConfiguredObject(Class<T> instanceType, Configuration config)
- {
- T instance;
- try
- {
- instance = instanceType.newInstance();
- }
- catch (Exception e)
- {
- _logger.error("Unable to instantiate configuration class " + instanceType + " - ensure it has a public default constructor");
- throw new IllegalArgumentException("Unable to instantiate configuration class " + instanceType + " - ensure it has a public default constructor", e);
- }
- Configurator.configure(instance);
-
- return instance;
- }
-
-
- public String getName()
- {
- return _name;
- }
-
- public QueueRegistry getQueueRegistry()
- {
- return _queueRegistry;
- }
-
- public ExchangeRegistry getExchangeRegistry()
- {
- return _exchangeRegistry;
- }
-
- public ExchangeFactory getExchangeFactory()
- {
- return _exchangeFactory;
- }
-
- public ApplicationRegistry getApplicationRegistry()
- {
- throw new UnsupportedOperationException();
- }
-
- public MessageStore getMessageStore()
- {
- return _messageStore;
- }
-
- public AuthenticationManager getAuthenticationManager()
- {
- return _authenticationManager;
- }
-
- public AccessManager getAccessManager()
- {
- return _accessManager;
- }
-
- public void close() throws Exception
- {
- if (_messageStore != null)
- {
- _messageStore.close();
- }
- }
-
- public ManagedObject getBrokerMBean()
- {
- return _brokerMBean;
- }
-
- public ManagedObject getManagedObject()
- {
- return _virtualHostMBean;
- }
-}
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.server.virtualhost;
+
+import javax.management.NotCompliantMBeanException;
+
+import org.apache.commons.configuration.Configuration;
+import org.apache.log4j.Logger;
+import org.apache.qpid.server.AMQBrokerManagerMBean;
+import org.apache.qpid.server.txn.TransactionManager;
+import org.apache.qpid.server.security.access.AccessManager;
+import org.apache.qpid.server.security.access.AccessManagerImpl;
+import org.apache.qpid.server.security.access.Accessable;
+import org.apache.qpid.server.security.auth.manager.PrincipalDatabaseAuthenticationManager;
+import org.apache.qpid.server.security.auth.manager.AuthenticationManager;
+import org.apache.qpid.server.configuration.Configurator;
+import org.apache.qpid.server.exchange.DefaultExchangeFactory;
+import org.apache.qpid.server.exchange.DefaultExchangeRegistry;
+import org.apache.qpid.server.exchange.ExchangeFactory;
+import org.apache.qpid.server.exchange.ExchangeRegistry;
+import org.apache.qpid.server.management.AMQManagedObject;
+import org.apache.qpid.server.management.ManagedObject;
+import org.apache.qpid.server.queue.DefaultQueueRegistry;
+import org.apache.qpid.server.queue.QueueRegistry;
+import org.apache.qpid.server.registry.ApplicationRegistry;
+import org.apache.qpid.server.messageStore.MessageStore;
+
+public class VirtualHost implements Accessable
+{
+ private static final Logger _logger = Logger.getLogger(VirtualHost.class);
+
+
+ private final String _name;
+
+ private QueueRegistry _queueRegistry;
+
+ private ExchangeRegistry _exchangeRegistry;
+
+ private ExchangeFactory _exchangeFactory;
+
+ private MessageStore _messageStore;
+
+ private TransactionManager _transactionManager;
+
+ protected VirtualHostMBean _virtualHostMBean;
+
+ private AMQBrokerManagerMBean _brokerMBean;
+
+ private AuthenticationManager _authenticationManager;
+
+ private AccessManager _accessManager;
+
+
+ public void setAccessableName(String name)
+ {
+ _logger.warn("Setting Accessable Name for VirualHost is not allowed. ("
+ + name + ") ignored remains :" + getAccessableName());
+ }
+
+ public String getAccessableName()
+ {
+ return _name;
+ }
+
+
+ /**
+ * Abstract MBean class. This has some of the methods implemented from management intrerface for exchanges. Any
+ * implementaion of an Exchange MBean should extend this class.
+ */
+ public class VirtualHostMBean extends AMQManagedObject implements ManagedVirtualHost
+ {
+ public VirtualHostMBean() throws NotCompliantMBeanException
+ {
+ super(ManagedVirtualHost.class, "VirtualHost");
+ }
+
+ public String getObjectInstanceName()
+ {
+ return _name.toString();
+ }
+
+ public String getName()
+ {
+ return _name.toString();
+ }
+
+ public VirtualHost getVirtualHost()
+ {
+ return VirtualHost.this;
+ }
+
+
+ } // End of MBean class
+
+
+ public VirtualHost(String name, MessageStore store) throws Exception
+ {
+ this(name, null, store);
+ }
+
+ public VirtualHost(String name, Configuration hostConfig) throws Exception
+ {
+ this(name, hostConfig, null);
+ }
+
+ private VirtualHost(String name, Configuration hostConfig, MessageStore store) throws Exception
+ {
+ _name = name;
+
+ _virtualHostMBean = new VirtualHostMBean();
+ // This isn't needed to be registered
+ //_virtualHostMBean.register();
+
+ _queueRegistry = new DefaultQueueRegistry(this);
+ _exchangeFactory = new DefaultExchangeFactory(this);
+ _exchangeRegistry = new DefaultExchangeRegistry(this);
+
+ if (store != null)
+ {
+ _messageStore = store;
+ }
+ else
+ {
+ if (hostConfig == null)
+ {
+ throw new IllegalAccessException("HostConfig and MessageStore cannot be null");
+ }
+ initialiseTransactionManager(hostConfig);
+ initialiseMessageStore(hostConfig);
+ }
+
+ _exchangeRegistry.initialise();
+
+ _logger.warn("VirtualHost authentication Managers require spec change to be operational.");
+ _authenticationManager = new PrincipalDatabaseAuthenticationManager(name, hostConfig);
+
+ _accessManager = new AccessManagerImpl(name, hostConfig);
+
+ _brokerMBean = new AMQBrokerManagerMBean(_virtualHostMBean);
+ _brokerMBean.register();
+ }
+
+ private void initialiseMessageStore(Configuration config) throws Exception
+ {
+ String messageStoreClass = config.getString("store.class");
+
+ Class clazz = Class.forName(messageStoreClass);
+ Object o = clazz.newInstance();
+
+ if (!(o instanceof MessageStore))
+ {
+ throw new ClassCastException("Message store class must implement " + MessageStore.class + ". Class " + clazz +
+ " does not.");
+ }
+ _messageStore = (MessageStore) o;
+ _messageStore.configure(this, _transactionManager, "store", config);
+ }
+
+ private void initialiseTransactionManager(Configuration config) throws Exception
+ {
+ String transactionManagerClass = config.getString("txn.class");
+ Class clazz = Class.forName(transactionManagerClass);
+ Object o = clazz.newInstance();
+
+ if (!(o instanceof TransactionManager))
+ {
+ throw new ClassCastException("Transaction Manager class must implement " + TransactionManager.class + ". Class " + clazz +
+ " does not.");
+ }
+ _transactionManager = (TransactionManager) o;
+ }
+
+
+ public <T> T getConfiguredObject(Class<T> instanceType, Configuration config)
+ {
+ T instance;
+ try
+ {
+ instance = instanceType.newInstance();
+ }
+ catch (Exception e)
+ {
+ _logger.error("Unable to instantiate configuration class " + instanceType + " - ensure it has a public default constructor");
+ throw new IllegalArgumentException("Unable to instantiate configuration class " + instanceType + " - ensure it has a public default constructor", e);
+ }
+ Configurator.configure(instance);
+
+ return instance;
+ }
+
+
+ public String getName()
+ {
+ return _name;
+ }
+
+ public QueueRegistry getQueueRegistry()
+ {
+ return _queueRegistry;
+ }
+
+ public ExchangeRegistry getExchangeRegistry()
+ {
+ return _exchangeRegistry;
+ }
+
+ public ExchangeFactory getExchangeFactory()
+ {
+ return _exchangeFactory;
+ }
+
+ public ApplicationRegistry getApplicationRegistry()
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ public MessageStore getMessageStore()
+ {
+ return _messageStore;
+ }
+
+ public TransactionManager getTransactionManager()
+ {
+ return _transactionManager;
+ }
+
+ public AuthenticationManager getAuthenticationManager()
+ {
+ return _authenticationManager;
+ }
+
+ public AccessManager getAccessManager()
+ {
+ return _accessManager;
+ }
+
+ public void close() throws Exception
+ {
+ if (_messageStore != null)
+ {
+ _messageStore.close();
+ }
+ }
+
+ public ManagedObject getBrokerMBean()
+ {
+ return _brokerMBean;
+ }
+
+ public ManagedObject getManagedObject()
+ {
+ return _virtualHostMBean;
+ }
+}
Modified: incubator/qpid/trunk/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueAlertTest.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueAlertTest.java?view=diff&rev=534541&r1=534540&r2=534541
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueAlertTest.java (original)
+++ incubator/qpid/trunk/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueAlertTest.java Wed May 2 09:49:03 2007
@@ -19,14 +19,16 @@
import junit.framework.TestCase;
import org.apache.qpid.AMQException;
-import org.apache.qpid.server.store.MessageStore;
-import org.apache.qpid.server.store.MemoryMessageStore;
+import org.apache.qpid.server.messageStore.MessageStore;
+import org.apache.qpid.server.messageStore.MemoryMessageStore;
import org.apache.qpid.server.store.StoreContext;
import org.apache.qpid.server.virtualhost.VirtualHost;
import org.apache.qpid.server.registry.IApplicationRegistry;
import org.apache.qpid.server.registry.ApplicationRegistry;
import org.apache.qpid.server.txn.TransactionalContext;
import org.apache.qpid.server.txn.NonTransactionalContext;
+import org.apache.qpid.server.txn.TransactionManager;
+import org.apache.qpid.server.txn.MemoryTransactionManager;
import org.apache.qpid.server.RequiredDeliveryException;
import org.apache.qpid.server.AMQChannel;
import org.apache.qpid.server.protocol.TestMinaProtocolSession;
@@ -51,6 +53,8 @@
private VirtualHost _virtualHost;
private AMQMinaProtocolSession protocolSession = null;
private MessageStore _messageStore = new MemoryMessageStore();
+ private TransactionManager _txm = new MemoryTransactionManager();
+
private StoreContext _storeContext = new StoreContext();
private TransactionalContext _transactionalContext = new NonTransactionalContext(_messageStore, _storeContext,
null,
@@ -170,7 +174,7 @@
public void testQueueDepthAlertWithSubscribers() throws Exception
{
protocolSession = new TestMinaProtocolSession();
- AMQChannel channel = new AMQChannel(protocolSession, 2, _messageStore, null);
+ AMQChannel channel = new AMQChannel(protocolSession, 2,_txm, _messageStore, null);
protocolSession.addChannel(channel);
// Create queue
Modified: incubator/qpid/trunk/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueMBeanTest.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueMBeanTest.java?view=diff&rev=534541&r1=534540&r2=534541
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueMBeanTest.java (original)
+++ incubator/qpid/trunk/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueMBeanTest.java Wed May 2 09:49:03 2007
@@ -30,9 +30,11 @@
import org.apache.qpid.server.registry.ApplicationRegistry;
import org.apache.qpid.server.txn.TransactionalContext;
import org.apache.qpid.server.txn.NonTransactionalContext;
-import org.apache.qpid.server.store.MessageStore;
+import org.apache.qpid.server.txn.TransactionManager;
+import org.apache.qpid.server.txn.MemoryTransactionManager;
+import org.apache.qpid.server.messageStore.MessageStore;
import org.apache.qpid.server.store.StoreContext;
-import org.apache.qpid.server.store.MemoryMessageStore;
+import org.apache.qpid.server.messageStore.MemoryMessageStore;
import javax.management.JMException;
import java.util.LinkedList;
@@ -47,6 +49,7 @@
private AMQQueue _queue;
private AMQQueueMBean _queueMBean;
private MessageStore _messageStore = new MemoryMessageStore();
+ private TransactionManager _txm = new MemoryTransactionManager();
private StoreContext _storeContext = new StoreContext();
private TransactionalContext _transactionalContext = new NonTransactionalContext(_messageStore, _storeContext,
null,
@@ -80,7 +83,7 @@
TestMinaProtocolSession protocolSession = new TestMinaProtocolSession();
- AMQChannel channel = new AMQChannel(protocolSession, 1, _messageStore, null);
+ AMQChannel channel = new AMQChannel(protocolSession, 1,_txm, _messageStore, null);
protocolSession.addChannel(channel);
_queue.registerProtocolSession(protocolSession, 1, new AMQShortString("test"), false, null,false,false);
Modified: incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/ClusteredProtocolSession.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/ClusteredProtocolSession.java?view=diff&rev=534541&r1=534540&r2=534541
==============================================================================
--- incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/ClusteredProtocolSession.java (original)
+++ incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/ClusteredProtocolSession.java Wed May 2 09:49:03 2007
@@ -108,6 +108,7 @@
throws AMQException
{
super(ClusteredProtocolSession.this,channelId,
+ virtualHost.getTransactionManager(),
virtualHost.getMessageStore(),
virtualHost.getExchangeRegistry());
}
Modified: incubator/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/ack/TxAckTest.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/ack/TxAckTest.java?view=diff&rev=534541&r1=534540&r2=534541
==============================================================================
--- incubator/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/ack/TxAckTest.java (original)
+++ incubator/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/ack/TxAckTest.java Wed May 2 09:49:03 2007
@@ -1,232 +0,0 @@
-/*
- *
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- *
- */
-package org.apache.qpid.server.ack;
-
-import junit.framework.TestCase;
-import org.apache.qpid.AMQException;
-import org.apache.qpid.framing.BasicPublishBody;
-import org.apache.qpid.framing.AMQShortString;
-import org.apache.qpid.framing.ContentHeaderBody;
-import org.apache.qpid.framing.AMQFrameDecodingException;
-import org.apache.qpid.framing.abstraction.MessagePublishInfo;
-import org.apache.qpid.server.RequiredDeliveryException;
-import org.apache.qpid.server.queue.AMQMessage;
-import org.apache.qpid.server.store.TestableMemoryMessageStore;
-import org.apache.qpid.server.store.StoreContext;
-import org.apache.qpid.server.txn.NonTransactionalContext;
-import org.apache.qpid.server.txn.TransactionalContext;
-
-import java.util.*;
-
-public class TxAckTest extends TestCase
-{
- private Scenario individual;
- private Scenario multiple;
- private Scenario combined;
-
- protected void setUp() throws Exception
- {
- super.setUp();
-
- //ack only 5th msg
- individual = new Scenario(10, Arrays.asList(5l), Arrays.asList(1l, 2l, 3l, 4l, 6l, 7l, 8l, 9l, 10l));
- individual.update(5, false);
-
- //ack all up to and including 5th msg
- multiple = new Scenario(10, Arrays.asList(1l, 2l, 3l, 4l, 5l), Arrays.asList(6l, 7l, 8l, 9l, 10l));
- multiple.update(5, true);
-
- //leave only 8th and 9th unacked
- combined = new Scenario(10, Arrays.asList(1l, 2l, 3l, 4l, 5l, 6l, 7l, 10l), Arrays.asList(8l, 9l));
- combined.update(3, false);
- combined.update(5, true);
- combined.update(7, true);
- combined.update(2, true);//should be ignored
- combined.update(1, false);//should be ignored
- combined.update(10, false);
- }
-
- public void testPrepare() throws AMQException
- {
- individual.prepare();
- multiple.prepare();
- combined.prepare();
- }
-
- public void testUndoPrepare() throws AMQException
- {
- individual.undoPrepare();
- multiple.undoPrepare();
- combined.undoPrepare();
- }
-
- public void testCommit() throws AMQException
- {
- individual.commit();
- multiple.commit();
- combined.commit();
- }
-
- public static junit.framework.Test suite()
- {
- return new junit.framework.TestSuite(TxAckTest.class);
- }
-
- private class Scenario
- {
- private final UnacknowledgedMessageMap _map = new UnacknowledgedMessageMapImpl(5000);
- private final TxAck _op = new TxAck(_map);
- private final List<Long> _acked;
- private final List<Long> _unacked;
- private StoreContext _storeContext = new StoreContext();
-
- Scenario(int messageCount, List<Long> acked, List<Long> unacked)
- {
- TransactionalContext txnContext = new NonTransactionalContext(new TestableMemoryMessageStore(),
- _storeContext, null,
- new LinkedList<RequiredDeliveryException>(),
- new HashSet<Long>());
- for (int i = 0; i < messageCount; i++)
- {
- long deliveryTag = i + 1;
-
- MessagePublishInfo info = new MessagePublishInfo()
- {
-
- public AMQShortString getExchange()
- {
- return null;
- }
-
- public boolean isImmediate()
- {
- return false;
- }
-
- public boolean isMandatory()
- {
- return false;
- }
-
- public AMQShortString getRoutingKey()
- {
- return null;
- }
- };
-
- TestMessage message = new TestMessage(deliveryTag, i, info, txnContext);
- _map.add(deliveryTag, new UnacknowledgedMessage(null, message, null, deliveryTag));
- }
- _acked = acked;
- _unacked = unacked;
- }
-
- void update(long deliverytag, boolean multiple)
- {
- _op.update(deliverytag, multiple);
- }
-
- private void assertCount(List<Long> tags, int expected)
- {
- for (long tag : tags)
- {
- UnacknowledgedMessage u = _map.get(tag);
- assertTrue("Message not found for tag " + tag, u != null);
- ((TestMessage) u.message).assertCountEquals(expected);
- }
- }
-
- void prepare() throws AMQException
- {
- _op.consolidate();
- _op.prepare(_storeContext);
-
- assertCount(_acked, -1);
- assertCount(_unacked, 0);
-
- }
-
- void undoPrepare()
- {
- _op.consolidate();
- _op.undoPrepare();
-
- assertCount(_acked, 1);
- assertCount(_unacked, 0);
- }
-
- void commit()
- {
- _op.consolidate();
- _op.commit(_storeContext);
-
- //check acked messages are removed from map
- Set<Long> keys = new HashSet<Long>(_map.getDeliveryTags());
- keys.retainAll(_acked);
- assertTrue("Expected messages with following tags to have been removed from map: " + keys, keys.isEmpty());
- //check unacked messages are still in map
- keys = new HashSet<Long>(_unacked);
- keys.removeAll(_map.getDeliveryTags());
- assertTrue("Expected messages with following tags to still be in map: " + keys, keys.isEmpty());
- }
- }
-
- private class TestMessage extends AMQMessage
- {
- private final long _tag;
- private int _count;
-
- TestMessage(long tag, long messageId, MessagePublishInfo publishBody, TransactionalContext txnContext)
- {
- super(messageId, publishBody, txnContext);
- try
- {
- setContentHeaderBody(new ContentHeaderBody()
- {
- public int getSize()
- {
- return 1;
- }
- });
- }
- catch (AMQException e)
- {
- // won't happen
- }
- _tag = tag;
- }
-
- public void incrementReference()
- {
- _count++;
- }
-
- public void decrementReference(StoreContext context)
- {
- _count--;
- }
-
- void assertCountEquals(int expected)
- {
- assertEquals("Wrong count for message with tag " + _tag, expected, _count);
- }
- }
-}
Modified: incubator/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/channel/MaxChannelsTest.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/channel/MaxChannelsTest.java?view=diff&rev=534541&r1=534540&r2=534541
==============================================================================
--- incubator/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/channel/MaxChannelsTest.java (original)
+++ incubator/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/channel/MaxChannelsTest.java Wed May 2 09:49:03 2007
@@ -1,75 +0,0 @@
-/*
- *
- * Copyright (c) 2006 The Apache Software Foundation
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- *
- */
-package org.apache.qpid.server.protocol;
-
-import junit.framework.TestCase;
-import org.apache.mina.common.IoSession;
-import org.apache.qpid.codec.AMQCodecFactory;
-import org.apache.qpid.server.AMQChannel;
-import org.apache.qpid.server.virtualhost.VirtualHost;
-import org.apache.qpid.server.registry.IApplicationRegistry;
-import org.apache.qpid.server.registry.ApplicationRegistry;
-import org.apache.qpid.server.exchange.ExchangeRegistry;
-import org.apache.qpid.server.queue.QueueRegistry;
-import org.apache.qpid.server.queue.AMQQueue;
-import org.apache.qpid.server.store.MessageStore;
-import org.apache.qpid.server.store.SkeletonMessageStore;
-import org.apache.qpid.AMQException;
-import org.apache.qpid.protocol.AMQConstant;
-import org.apache.qpid.framing.AMQShortString;
-
-import javax.management.JMException;
-
-/** Test class to test MBean operations for AMQMinaProtocolSession. */
-public class MaxChannelsTest extends TestCase
-{
-// private MessageStore _messageStore = new SkeletonMessageStore();
-
- public void testChannels() throws Exception
- {
- IApplicationRegistry appRegistry = ApplicationRegistry.getInstance();
- AMQMinaProtocolSession _protocolSession = new AMQMinaProtocolSession(new MockIoSession(),
- appRegistry.getVirtualHostRegistry(),
- new AMQCodecFactory(true),
- null);
- _protocolSession.setVirtualHost(appRegistry.getVirtualHostRegistry().getVirtualHost("test"));
-
- // check the channel count is correct
- int channelCount = _protocolSession.getChannels().size();
- assertEquals("Initial channel count wrong", 0, channelCount);
-
- long maxChannels = 10L;
- _protocolSession.setMaximumNumberOfChannels(maxChannels);
- assertEquals("Number of channels not correctly set.", new Long(maxChannels), _protocolSession.getMaximumNumberOfChannels());
-
-
- try
- {
- for (long currentChannel = 0L; currentChannel < maxChannels; currentChannel++)
- {
- _protocolSession.addChannel(new AMQChannel(_protocolSession, (int) currentChannel, null, null));
- }
- }
- catch (AMQException e)
- {
- assertEquals("Wrong exception recevied.", e.getErrorCode(), AMQConstant.NOT_ALLOWED);
- }
- assertEquals("Maximum number of channels not set.", new Long(maxChannels), new Long(_protocolSession.getChannels().size()));
- }
-
-}
Modified: incubator/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/exchange/AbstractHeadersExchangeTestBase.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/exchange/AbstractHeadersExchangeTestBase.java?view=diff&rev=534541&r1=534540&r2=534541
==============================================================================
--- incubator/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/exchange/AbstractHeadersExchangeTestBase.java (original)
+++ incubator/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/exchange/AbstractHeadersExchangeTestBase.java Wed May 2 09:49:03 2007
@@ -28,9 +28,9 @@
import org.apache.qpid.server.queue.AMQQueue;
import org.apache.qpid.server.queue.MessageHandleFactory;
import org.apache.qpid.server.registry.ApplicationRegistry;
-import org.apache.qpid.server.store.MessageStore;
+import org.apache.qpid.server.messageStore.MessageStore;
import org.apache.qpid.server.store.SkeletonMessageStore;
-import org.apache.qpid.server.store.MemoryMessageStore;
+import org.apache.qpid.server.messageStore.MemoryMessageStore;
import org.apache.qpid.server.store.StoreContext;
import org.apache.qpid.server.txn.NonTransactionalContext;
import org.apache.qpid.server.txn.TransactionalContext;
@@ -261,7 +261,7 @@
*/
static class Message extends AMQMessage
{
- private static MessageStore _messageStore = new SkeletonMessageStore();
+ private static MessageStore _messageStore = new MemoryMessageStore();
private static StoreContext _storeContext = new StoreContext();
Modified: incubator/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSessionMBeanTest.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSessionMBeanTest.java?view=diff&rev=534541&r1=534540&r2=534541
==============================================================================
--- incubator/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSessionMBeanTest.java (original)
+++ incubator/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSessionMBeanTest.java Wed May 2 09:49:03 2007
@@ -21,25 +21,30 @@
import org.apache.mina.common.IoSession;
import org.apache.qpid.codec.AMQCodecFactory;
import org.apache.qpid.server.AMQChannel;
+import org.apache.qpid.server.txn.MemoryTransactionManager;
+import org.apache.qpid.server.txn.TransactionManager;
import org.apache.qpid.server.virtualhost.VirtualHost;
import org.apache.qpid.server.registry.IApplicationRegistry;
import org.apache.qpid.server.registry.ApplicationRegistry;
import org.apache.qpid.server.exchange.ExchangeRegistry;
import org.apache.qpid.server.queue.QueueRegistry;
import org.apache.qpid.server.queue.AMQQueue;
-import org.apache.qpid.server.store.MessageStore;
+import org.apache.qpid.server.messageStore.MessageStore;
+import org.apache.qpid.server.messageStore.MemoryMessageStore;
import org.apache.qpid.server.store.SkeletonMessageStore;
import org.apache.qpid.AMQException;
import org.apache.qpid.framing.AMQShortString;
import javax.management.JMException;
+
/**
* Test class to test MBean operations for AMQMinaProtocolSession.
*/
public class AMQProtocolSessionMBeanTest extends TestCase
{
- private MessageStore _messageStore = new SkeletonMessageStore();
+ private MessageStore _messageStore = new MemoryMessageStore();
+ private TransactionManager _txm = new MemoryTransactionManager();
private AMQMinaProtocolSession _protocolSession;
private AMQChannel _channel;
private AMQProtocolSessionMBean _mbean;
@@ -54,7 +59,7 @@
new AMQShortString("test"),
true,
_protocolSession.getVirtualHost());
- AMQChannel channel = new AMQChannel(_protocolSession,2, _messageStore, null);
+ AMQChannel channel = new AMQChannel(_protocolSession,2,_txm, _messageStore, null);
channel.setDefaultQueue(queue);
_protocolSession.addChannel(channel);
channelCount = _mbean.channels().size();
@@ -65,7 +70,7 @@
assertTrue(_mbean.getMaximumNumberOfChannels() == 1000L);
// check APIs
- AMQChannel channel3 = new AMQChannel(_protocolSession,3, _messageStore, null);
+ AMQChannel channel3 = new AMQChannel(_protocolSession,3,_txm, _messageStore, null);
channel3.setLocalTransactional();
_protocolSession.addChannel(channel3);
_mbean.rollbackTransactions(2);
@@ -85,14 +90,14 @@
}
// check if closing of session works
- _protocolSession.addChannel(new AMQChannel(_protocolSession,5, _messageStore, null));
+ _protocolSession.addChannel(new AMQChannel(_protocolSession,5, _txm, _messageStore, null));
_mbean.closeConnection();
try
{
channelCount = _mbean.channels().size();
assertTrue(channelCount == 0);
// session is now closed so adding another channel should throw an exception
- _protocolSession.addChannel(new AMQChannel(_protocolSession,6, _messageStore, null));
+ _protocolSession.addChannel(new AMQChannel(_protocolSession,6, _txm, _messageStore, null));
fail();
}
catch(AMQException ex)
@@ -112,7 +117,7 @@
new AMQCodecFactory(true),
null);
_protocolSession.setVirtualHost(appRegistry.getVirtualHostRegistry().getVirtualHost("test"));
- _channel = new AMQChannel(_protocolSession,1, _messageStore, null);
+ _channel = new AMQChannel(_protocolSession,1, _txm, _messageStore, null);
_protocolSession.addChannel(_channel);
_mbean = (AMQProtocolSessionMBean)_protocolSession.getManagedObject();
}
Modified: incubator/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/queue/AckTest.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/queue/AckTest.java?view=diff&rev=534541&r1=534540&r2=534541
==============================================================================
--- incubator/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/queue/AckTest.java (original)
+++ incubator/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/queue/AckTest.java Wed May 2 09:49:03 2007
@@ -1,356 +0,0 @@
-/*
- *
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- *
- */
-package org.apache.qpid.server.queue;
-
-import junit.framework.TestCase;
-import org.apache.log4j.Logger;
-import org.apache.qpid.AMQException;
-import org.apache.qpid.framing.BasicContentHeaderProperties;
-import org.apache.qpid.framing.BasicPublishBody;
-import org.apache.qpid.framing.ContentHeaderBody;
-import org.apache.qpid.framing.AMQShortString;
-import org.apache.qpid.framing.abstraction.MessagePublishInfo;
-import org.apache.qpid.server.AMQChannel;
-import org.apache.qpid.server.RequiredDeliveryException;
-import org.apache.qpid.server.ack.UnacknowledgedMessage;
-import org.apache.qpid.server.ack.UnacknowledgedMessageMap;
-import org.apache.qpid.server.registry.ApplicationRegistry;
-import org.apache.qpid.server.store.TestableMemoryMessageStore;
-import org.apache.qpid.server.store.StoreContext;
-import org.apache.qpid.server.txn.NonTransactionalContext;
-import org.apache.qpid.server.txn.TransactionalContext;
-import org.apache.qpid.server.util.TestApplicationRegistry;
-import org.apache.qpid.server.util.NullApplicationRegistry;
-
-import java.util.LinkedList;
-import java.util.Set;
-import java.util.HashSet;
-
-/**
- * Tests that acknowledgements are handled correctly.
- */
-public class AckTest extends TestCase
-{
- private static final Logger _log = Logger.getLogger(AckTest.class);
-
- private SubscriptionImpl _subscription;
-
- private MockProtocolSession _protocolSession;
-
- private TestableMemoryMessageStore _messageStore;
-
- private StoreContext _storeContext = new StoreContext();
-
- private AMQChannel _channel;
-
- private SubscriptionSet _subscriptionManager;
-
- private AMQQueue _queue;
-
- private static final AMQShortString DEFAULT_CONSUMER_TAG = new AMQShortString("conTag");
-
- public AckTest() throws Exception
- {
- ApplicationRegistry.initialise(new NullApplicationRegistry());
- }
-
- protected void setUp() throws Exception
- {
- super.setUp();
- _messageStore = new TestableMemoryMessageStore();
- _protocolSession = new MockProtocolSession(_messageStore);
- _channel = new AMQChannel(_protocolSession,5, _messageStore, null/*dont need exchange registry*/);
-
- _protocolSession.addChannel(_channel);
- _subscriptionManager = new SubscriptionSet();
- _queue = new AMQQueue(new AMQShortString("myQ"), false, new AMQShortString("guest"), true, ApplicationRegistry.getInstance().getVirtualHostRegistry().getVirtualHost("test"), _subscriptionManager);
- }
-
- private void publishMessages(int count) throws AMQException
- {
- publishMessages(count, false);
- }
-
- private void publishMessages(int count, boolean persistent) throws AMQException
- {
- TransactionalContext txnContext = new NonTransactionalContext(_messageStore, _storeContext, null,
- new LinkedList<RequiredDeliveryException>(),
- new HashSet<Long>());
- MessageHandleFactory factory = new MessageHandleFactory();
- for (int i = 1; i <= count; i++)
- {
- // AMQP version change: Hardwire the version to 0-8 (major=8, minor=0)
- // TODO: Establish some way to determine the version for the test.
- MessagePublishInfo publishBody = new MessagePublishInfo()
- {
-
- public AMQShortString getExchange()
- {
- return new AMQShortString("someExchange");
- }
-
- public boolean isImmediate()
- {
- return false;
- }
-
- public boolean isMandatory()
- {
- return false;
- }
-
- public AMQShortString getRoutingKey()
- {
- return new AMQShortString("rk");
- }
- };
- AMQMessage msg = new AMQMessage(_messageStore.getNewMessageId(), publishBody, txnContext);
- if (persistent)
- {
- BasicContentHeaderProperties b = new BasicContentHeaderProperties();
- //This is DeliveryMode.PERSISTENT
- b.setDeliveryMode((byte) 2);
- ContentHeaderBody cb = new ContentHeaderBody();
- cb.properties = b;
- msg.setContentHeaderBody(cb);
- }
- else
- {
- msg.setContentHeaderBody(new ContentHeaderBody());
- }
- // we increment the reference here since we are not delivering the messaging to any queues, which is where
- // the reference is normally incremented. The test is easier to construct if we have direct access to the
- // subscription
- msg.incrementReference();
- msg.routingComplete(_messageStore, _storeContext, factory);
- // we manually send the message to the subscription
- _subscription.send(msg, _queue);
- }
- }
-
- /**
- * Tests that the acknowledgements are correctly associated with a channel and
- * order is preserved when acks are enabled
- */
- public void testAckChannelAssociationTest() throws AMQException
- {
- _subscription = new SubscriptionImpl(5, _protocolSession, DEFAULT_CONSUMER_TAG, true);
- final int msgCount = 10;
- publishMessages(msgCount, true);
-
- UnacknowledgedMessageMap map = _channel.getUnacknowledgedMessageMap();
- assertTrue(map.size() == msgCount);
- assertTrue(_messageStore.getMessageMetaDataMap().size() == msgCount);
-
- Set<Long> deliveryTagSet = map.getDeliveryTags();
- int i = 1;
- for (long deliveryTag : deliveryTagSet)
- {
- assertTrue(deliveryTag == i);
- i++;
- UnacknowledgedMessage unackedMsg = map.get(deliveryTag);
- assertTrue(unackedMsg.queue == _queue);
- }
-
- assertTrue(map.size() == msgCount);
- assertTrue(_messageStore.getMessageMetaDataMap().size() == msgCount);
- }
-
- /**
- * Tests that in no-ack mode no messages are retained
- */
- public void testNoAckMode() throws AMQException
- {
- // false arg means no acks expected
- _subscription = new SubscriptionImpl(5, _protocolSession, DEFAULT_CONSUMER_TAG, false);
- final int msgCount = 10;
- publishMessages(msgCount);
-
- UnacknowledgedMessageMap map = _channel.getUnacknowledgedMessageMap();
- assertTrue(map.size() == 0);
- assertTrue(_messageStore.getMessageMetaDataMap().size() == 0);
- }
-
- /**
- * Tests that a single acknowledgement is handled correctly (i.e multiple flag not
- * set case)
- */
- public void testSingleAckReceivedTest() throws AMQException
- {
- _subscription = new SubscriptionImpl(5, _protocolSession, DEFAULT_CONSUMER_TAG, true);
- final int msgCount = 10;
- publishMessages(msgCount);
-
- _channel.acknowledgeMessage(5, false);
- UnacknowledgedMessageMap map = _channel.getUnacknowledgedMessageMap();
- assertTrue(map.size() == msgCount - 1);
-
- Set<Long> deliveryTagSet = map.getDeliveryTags();
- int i = 1;
- for (long deliveryTag : deliveryTagSet)
- {
- assertTrue(deliveryTag == i);
- UnacknowledgedMessage unackedMsg = map.get(deliveryTag);
- assertTrue(unackedMsg.queue == _queue);
- // 5 is the delivery tag of the message that *should* be removed
- if (++i == 5)
- {
- ++i;
- }
- }
- }
-
- /**
- * Tests that a single acknowledgement is handled correctly (i.e multiple flag not
- * set case)
- */
- public void testMultiAckReceivedTest() throws AMQException
- {
- _subscription = new SubscriptionImpl(5, _protocolSession, DEFAULT_CONSUMER_TAG, true);
- final int msgCount = 10;
- publishMessages(msgCount);
-
- _channel.acknowledgeMessage(5, true);
- UnacknowledgedMessageMap map = _channel.getUnacknowledgedMessageMap();
- assertTrue(map.size() == 5);
-
- Set<Long> deliveryTagSet = map.getDeliveryTags();
- int i = 1;
- for (long deliveryTag : deliveryTagSet)
- {
- assertTrue(deliveryTag == i + 5);
- UnacknowledgedMessage unackedMsg = map.get(deliveryTag);
- assertTrue(unackedMsg.queue == _queue);
- ++i;
- }
- }
-
- /**
- * Tests that a multiple acknowledgement is handled correctly. When ack'ing all pending msgs.
- */
- public void testMultiAckAllReceivedTest() throws AMQException
- {
- _subscription = new SubscriptionImpl(5, _protocolSession, DEFAULT_CONSUMER_TAG, true);
- final int msgCount = 10;
- publishMessages(msgCount);
-
- _channel.acknowledgeMessage(0, true);
- UnacknowledgedMessageMap map = _channel.getUnacknowledgedMessageMap();
- assertTrue(map.size() == 0);
-
- Set<Long> deliveryTagSet = map.getDeliveryTags();
- int i = 1;
- for (long deliveryTag : deliveryTagSet)
- {
- assertTrue(deliveryTag == i + 5);
- UnacknowledgedMessage unackedMsg = map.get(deliveryTag);
- assertTrue(unackedMsg.queue == _queue);
- ++i;
- }
- }
-
- public void testPrefetchHighLow() throws AMQException
- {
- int lowMark = 5;
- int highMark = 10;
-
- _subscription = new SubscriptionImpl(5, _protocolSession, DEFAULT_CONSUMER_TAG, true);
- _channel.setPrefetchLowMarkCount(lowMark);
- _channel.setPrefetchHighMarkCount(highMark);
-
- assertTrue(_channel.getPrefetchLowMarkCount() == lowMark);
- assertTrue(_channel.getPrefetchHighMarkCount() == highMark);
-
- publishMessages(highMark);
-
- // at this point we should have sent out only highMark messages
- // which have not bee received so will be queued up in the channel
- // which should be suspended
- assertTrue(_subscription.isSuspended());
- UnacknowledgedMessageMap map = _channel.getUnacknowledgedMessageMap();
- assertTrue(map.size() == highMark);
-
- //acknowledge messages so we are just above lowMark
- _channel.acknowledgeMessage(lowMark - 1, true);
-
- //we should still be suspended
- assertTrue(_subscription.isSuspended());
- assertTrue(map.size() == lowMark + 1);
-
- //acknowledge one more message
- _channel.acknowledgeMessage(lowMark, true);
-
- //and suspension should be lifted
- assertTrue(!_subscription.isSuspended());
-
- //pubilsh more msgs so we are just below the limit
- publishMessages(lowMark - 1);
-
- //we should not be suspended
- assertTrue(!_subscription.isSuspended());
-
- //acknowledge all messages
- _channel.acknowledgeMessage(0, true);
- try
- {
- Thread.sleep(3000);
- }
- catch (InterruptedException e)
- {
- _log.error("Error: " + e, e);
- }
- //map will be empty
- assertTrue(map.size() == 0);
- }
-
- public void testPrefetch() throws AMQException
- {
- _subscription = new SubscriptionImpl(5, _protocolSession, DEFAULT_CONSUMER_TAG, true);
- _channel.setPrefetchCount(5);
-
- assertTrue(_channel.getPrefetchCount() == 5);
-
- final int msgCount = 5;
- publishMessages(msgCount);
-
- // at this point we should have sent out only 5 messages with a further 5 queued
- // up in the channel which should now be suspended
- assertTrue(_subscription.isSuspended());
- UnacknowledgedMessageMap map = _channel.getUnacknowledgedMessageMap();
- assertTrue(map.size() == 5);
- _channel.acknowledgeMessage(5, true);
- assertTrue(!_subscription.isSuspended());
- try
- {
- Thread.sleep(3000);
- }
- catch (InterruptedException e)
- {
- _log.error("Error: " + e, e);
- }
- assertTrue(map.size() == 0);
- }
-
- public static junit.framework.Test suite()
- {
- return new junit.framework.TestSuite(AckTest.class);
- }
-}
Modified: incubator/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/queue/MessageTestHelper.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/queue/MessageTestHelper.java?view=diff&rev=534541&r1=534540&r2=534541
==============================================================================
--- incubator/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/queue/MessageTestHelper.java (original)
+++ incubator/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/queue/MessageTestHelper.java Wed May 2 09:49:03 2007
@@ -24,7 +24,8 @@
import org.apache.qpid.framing.ContentHeaderBody;
import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.framing.abstraction.MessagePublishInfo;
-import org.apache.qpid.server.store.MessageStore;
+import org.apache.qpid.server.messageStore.MessageStore;
+import org.apache.qpid.server.messageStore.MemoryMessageStore;
import org.apache.qpid.server.store.SkeletonMessageStore;
import org.apache.qpid.server.store.StoreContext;
import org.apache.qpid.server.registry.ApplicationRegistry;
@@ -42,7 +43,7 @@
class MessageTestHelper extends TestCase
{
- private final MessageStore _messageStore = new SkeletonMessageStore();
+ private final MessageStore _messageStore = new MemoryMessageStore();
private final StoreContext _storeContext = new StoreContext();
Modified: incubator/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/store/TestReferenceCounting.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/store/TestReferenceCounting.java?view=diff&rev=534541&r1=534540&r2=534541
==============================================================================
--- incubator/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/store/TestReferenceCounting.java (original)
+++ incubator/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/store/TestReferenceCounting.java Wed May 2 09:49:03 2007
@@ -1,146 +0,0 @@
-/*
- *
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- *
- */
-package org.apache.qpid.server.store;
-
-import junit.framework.TestCase;
-import org.apache.qpid.AMQException;
-import org.apache.qpid.framing.BasicContentHeaderProperties;
-import org.apache.qpid.framing.BasicPublishBody;
-import org.apache.qpid.framing.ContentHeaderBody;
-import org.apache.qpid.framing.AMQShortString;
-import org.apache.qpid.framing.abstraction.MessagePublishInfo;
-import org.apache.qpid.server.queue.AMQMessage;
-import org.apache.qpid.server.queue.MessageHandleFactory;
-import org.apache.qpid.server.txn.NonTransactionalContext;
-
-/**
- * Tests that reference counting works correctly with AMQMessage and the message store
- */
-public class TestReferenceCounting extends TestCase
-{
- private TestableMemoryMessageStore _store;
-
- private StoreContext _storeContext = new StoreContext();
-
- protected void setUp() throws Exception
- {
- super.setUp();
- _store = new TestableMemoryMessageStore();
- }
-
- /**
- * Check that when the reference count is decremented the message removes itself from the store
- */
- public void testMessageGetsRemoved() throws AMQException
- {
- createPersistentContentHeader();
-
- MessagePublishInfo info = new MessagePublishInfo()
- {
-
- public AMQShortString getExchange()
- {
- return null;
- }
-
- public boolean isImmediate()
- {
- return false;
- }
-
- public boolean isMandatory()
- {
- return false;
- }
-
- public AMQShortString getRoutingKey()
- {
- return null;
- }
- };
-
- AMQMessage message = new AMQMessage(_store.getNewMessageId(), info,
- new NonTransactionalContext(_store, _storeContext, null, null, null),
- createPersistentContentHeader());
- message = message.takeReference();
-
- // we call routing complete to set up the handle
- message.routingComplete(_store, _storeContext, new MessageHandleFactory());
- assertTrue(_store.getMessageMetaDataMap().size() == 1);
- message.decrementReference(_storeContext);
- assertTrue(_store.getMessageMetaDataMap().size() == 0);
- }
-
- private ContentHeaderBody createPersistentContentHeader()
- {
- ContentHeaderBody chb = new ContentHeaderBody();
- BasicContentHeaderProperties bchp = new BasicContentHeaderProperties();
- bchp.setDeliveryMode((byte)2);
- chb.properties = bchp;
- return chb;
- }
-
- public void testMessageRemains() throws AMQException
- {
-
- MessagePublishInfo info = new MessagePublishInfo()
- {
-
- public AMQShortString getExchange()
- {
- return null;
- }
-
- public boolean isImmediate()
- {
- return false;
- }
-
- public boolean isMandatory()
- {
- return false;
- }
-
- public AMQShortString getRoutingKey()
- {
- return null;
- }
- };
-
- AMQMessage message = new AMQMessage(_store.getNewMessageId(),
- info,
- new NonTransactionalContext(_store, _storeContext, null, null, null),
- createPersistentContentHeader());
-
- message = message.takeReference();
- // we call routing complete to set up the handle
- message.routingComplete(_store, _storeContext, new MessageHandleFactory());
- assertTrue(_store.getMessageMetaDataMap().size() == 1);
- message = message.takeReference();
- message.decrementReference(_storeContext);
- assertTrue(_store.getMessageMetaDataMap().size() == 1);
- }
-
- public static junit.framework.Test suite()
- {
- return new junit.framework.TestSuite(TestReferenceCounting.class);
- }
-}