[jboss-cvs] JBoss Messaging SVN: r4976 - in branches/Branch_JBossMessaging_1_4_0_SP3_CP: src/main/org/jboss/jms/server and 6 other directories.
jboss-cvs-commits at lists.jboss.org
jboss-cvs-commits at lists.jboss.org
Wed Sep 17 22:08:07 EDT 2008
Author: gaohoward
Date: 2008-09-17 22:08:06 -0400 (Wed, 17 Sep 2008)
New Revision: 4976
Added:
branches/Branch_JBossMessaging_1_4_0_SP3_CP/tests/src/org/jboss/test/messaging/core/impl/
branches/Branch_JBossMessaging_1_4_0_SP3_CP/tests/src/org/jboss/test/messaging/core/impl/NullPersistenceManagerTest.java
branches/Branch_JBossMessaging_1_4_0_SP3_CP/tests/src/org/jboss/test/messaging/jms/clustering/NullPersistenceClusterTest.java
Modified:
branches/Branch_JBossMessaging_1_4_0_SP3_CP/src/etc/server/default/deploy/null-persistence-service.xml
branches/Branch_JBossMessaging_1_4_0_SP3_CP/src/main/org/jboss/jms/server/ServerPeer.java
branches/Branch_JBossMessaging_1_4_0_SP3_CP/src/main/org/jboss/messaging/core/impl/NullPersistenceManager.java
branches/Branch_JBossMessaging_1_4_0_SP3_CP/src/main/org/jboss/messaging/core/jmx/MessagingPostOfficeService.java
branches/Branch_JBossMessaging_1_4_0_SP3_CP/tests/etc/container.xml
Log:
JBMESSAGING-1376
Modified: branches/Branch_JBossMessaging_1_4_0_SP3_CP/src/etc/server/default/deploy/null-persistence-service.xml
===================================================================
--- branches/Branch_JBossMessaging_1_4_0_SP3_CP/src/etc/server/default/deploy/null-persistence-service.xml 2008-09-17 20:27:12 UTC (rev 4975)
+++ branches/Branch_JBossMessaging_1_4_0_SP3_CP/src/etc/server/default/deploy/null-persistence-service.xml 2008-09-18 02:08:06 UTC (rev 4976)
@@ -32,7 +32,7 @@
<!-- This post office is clustered. If you don't want a clustered post office then set to false -->
- <attribute name="Clustered">false</attribute>
+ <attribute name="Clustered">true</attribute>
<!-- All the remaining properties only have to be specified if the post office is clustered.
You can safely comment them out if your post office is non clustered -->
Modified: branches/Branch_JBossMessaging_1_4_0_SP3_CP/src/main/org/jboss/jms/server/ServerPeer.java
===================================================================
--- branches/Branch_JBossMessaging_1_4_0_SP3_CP/src/main/org/jboss/jms/server/ServerPeer.java 2008-09-17 20:27:12 UTC (rev 4975)
+++ branches/Branch_JBossMessaging_1_4_0_SP3_CP/src/main/org/jboss/jms/server/ServerPeer.java 2008-09-18 02:08:06 UTC (rev 4976)
@@ -1,24 +1,24 @@
/*
- * JBoss, Home of Professional Open Source
- * Copyright 2005, JBoss Inc., and individual contributors as indicated
- * by the @authors tag. See the copyright.txt in the distribution for a
- * full listing of individual contributors.
- *
- * This is free software; you can redistribute it and/or modify it
- * under the terms of the GNU Lesser General Public License as
- * published by the Free Software Foundation; either version 2.1 of
- * the License, or (at your option) any later version.
- *
- * This software is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
- * Lesser General Public License for more details.
- *
- * You should have received a copy of the GNU Lesser General Public
- * License along with this software; if not, write to the Free
- * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
- * 02110-1301 USA, or see the FSF site: http://www.fsf.org.
- */
+ * JBoss, Home of Professional Open Source Copyright 2005, JBoss Inc., and
+ * individual contributors as indicated by the @authors tag. See the
+ * copyright.txt in the distribution for a full listing of individual
+ * contributors.
+ *
+ * This is free software; you can redistribute it and/or modify it under the
+ * terms of the GNU Lesser General Public License as published by the Free
+ * Software Foundation; either version 2.1 of the License, or (at your option)
+ * any later version.
+ *
+ * This software is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
+ * FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more
+ * details.
+ *
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with this software; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA, or see the FSF
+ * site: http://www.fsf.org.
+ */
package org.jboss.jms.server;
import java.io.ByteArrayOutputStream;
@@ -68,6 +68,7 @@
import org.jboss.messaging.core.impl.FailoverWaiter;
import org.jboss.messaging.core.impl.IDManager;
import org.jboss.messaging.core.impl.JDBCPersistenceManager;
+import org.jboss.messaging.core.impl.NullPersistenceManager;
import org.jboss.messaging.core.impl.RotatingID;
import org.jboss.messaging.core.impl.clusterconnection.ClusterConnectionManager;
import org.jboss.messaging.core.impl.memory.SimpleMemoryManager;
@@ -87,44 +88,58 @@
import EDU.oswego.cs.dl.util.concurrent.ConcurrentReaderHashMap;
-/**
+/*
* A JMS server peer.
- *
+ *
* @author <a href="mailto:ovidiu at feodorov.com">Ovidiu Feodorov</a>
+ *
* @author <a href="mailto:tim.fox at jboss.com">Tim Fox</a>
+ *
* @author <a href="mailto:juha at jboss.org">Juha Lindfors</a>
+ *
* @author <a href="mailto:aslak at conduct.no">Aslak Knutsen</a>
- *
+ *
* @version <tt>$Revision$</tt>
- *
+ *
* $Id$
*/
public class ServerPeer extends ServiceMBeanSupport
{
- // Constants ------------------------------------------------------------------------------------
+ // Constants
+ // ------------------------------------------------------------------
+ // ------------------
private static final Logger log = Logger.getLogger(ServerPeer.class);
- // The "subsystem" label this ServerPeer uses to register its ServerInvocationHandler with the
+ // The "subsystem" label this ServerPeer uses to register its
+ // ServerInvocationHandler with the
// Remoting connector
public static final String REMOTING_JMS_SUBSYSTEM = "JMS";
- // Static ---------------------------------------------------------------------------------------
+ // Static
+ // --------------------------------------------------------------------
+ // -------------------
- // Attributes -----------------------------------------------------------------------------------
+ // Attributes
+ // ----------------------------------------------------------------
+ // -------------------
private int serverPeerID = -1;
+
private byte[] clientAOPStack;
+
private Version version;
private String defaultQueueJNDIContext = "";
+
private String defaultTopicJNDIContext = "";
private boolean started;
private boolean supportsFailover = true;
- // The default maximum number of delivery attempts before sending to DLQ - can be overridden on
+ // The default maximum number of delivery attempts before sending to DLQ -
+ // can be overridden on
// the destination
private int defaultMaxDeliveryAttempts = 10;
@@ -139,9 +154,9 @@
private long defaultRedeliveryDelay;
private long messageCounterSamplePeriod = 10000;
-
+
private boolean enableMessageCounters;
-
+
private int defaultMessageCounterHistoryDayLimit;
private String clusterPullConnectionFactoryName;
@@ -153,49 +168,73 @@
private long recoverDeliveriesTimeout = 5 * 60 * 1000;
private String suckerPassword;
-
- //Global override for strict behaviour
+
+ // Global override for strict behaviour
private boolean strictTck;
- //From a system property - this overrides
+ // From a system property - this overrides
private boolean strictTckProperty;
+ // For generating unique Channel ID for cluster without a shared DB
+ private long serverStartTime;
+
// wired components
private DestinationJNDIMapper destinationJNDIMapper;
+
private SecurityMetadataStore securityStore;
+
private ConnectionFactoryJNDIMapper connFactoryJNDIMapper;
+
private TransactionRepository txRepository;
+
private SimpleConnectionManager connectionManager;
+
private ConnectorManager connectorManager;
+
private IDManager messageIDManager;
+
private IDManager channelIDManager;
+
private MemoryManager memoryManager;
+
private MessageStore messageStore;
+
private MessageCounterManager messageCounterManager;
+
private ClusterConnectionManager clusterConnectionManager;
+
private ClusterNotifier clusterNotifier;
+
private FailoverWaiter failoverWaiter;
+
private RotatingID messageIDMgr;
// plugins
protected ObjectName persistenceManagerObjectName;
+
protected PersistenceManager persistenceManager;
protected ObjectName postOfficeObjectName;
+
protected PostOffice postOffice;
protected ObjectName jmsUserManagerObjectName;
+
protected JMSUserManager jmsUserManager;
protected ObjectName defaultDLQObjectName;
+
protected Queue defaultDLQ;
protected ObjectName defaultExpiryQueueObjectName;
+
protected Queue defaultExpiryQueue;
- // Constructors ---------------------------------------------------------------------------------
+ // Constructors
+ // --------------------------------------------------------------
+ // -------------------
public ServerPeer() throws Exception
{
// Some wired components need to be started here
@@ -208,7 +247,8 @@
started = false;
}
- // ServiceMBeanSupport overrides ----------------------------------------------------------------
+ // ServiceMBeanSupport overrides
+ // ----------------------------------------------------------------
public synchronized void startService() throws Exception
{
@@ -228,28 +268,41 @@
log.debug(this + " starting");
+ serverStartTime = System.currentTimeMillis();
+
loadClientAOPConfig();
loadServerAOPConfig();
MBeanServer mbeanServer = getServer();
- // Acquire references to plugins. Each plug-in will be accessed directly via a reference
- // circumventing the MBeanServer. However, they are installed as services to take advantage
+ // Acquire references to plugins. Each plug-in will be accessed
+ // directly via a reference
+ // circumventing the MBeanServer. However, they are installed as
+ // services to take advantage
// of their automatically-creating management interface.
- persistenceManager = (PersistenceManager)JMXAccessor.getJMXAttributeOverSecurity(mbeanServer, persistenceManagerObjectName, "Instance");
-
+ persistenceManager = (PersistenceManager)JMXAccessor.getJMXAttributeOverSecurity(mbeanServer,
+ persistenceManagerObjectName,
+ "Instance");
+
if (persistenceManager instanceof JDBCPersistenceManager)
{
- ((JDBCPersistenceManager)persistenceManager).injectNodeID(serverPeerID);
+ ((JDBCPersistenceManager)persistenceManager).injectNodeID(serverPeerID);
}
+ else if (persistenceManager instanceof NullPersistenceManager)
+ {
+ ((NullPersistenceManager)persistenceManager).initCounter(serverPeerID, serverStartTime);
+ }
- jmsUserManager = (JMSUserManager)JMXAccessor.getJMXAttributeOverSecurity(mbeanServer, jmsUserManagerObjectName, "Instance");
+ jmsUserManager = (JMSUserManager)JMXAccessor.getJMXAttributeOverSecurity(mbeanServer,
+ jmsUserManagerObjectName,
+ "Instance");
strictTckProperty = "true".equalsIgnoreCase(System.getProperty("jboss.messaging.stricttck"));
- // We get references to some plugins lazily to avoid problems with circular MBean
+ // We get references to some plugins lazily to avoid problems with
+ // circular MBean
// dependencies
// Create the wired components
@@ -262,8 +315,7 @@
memoryManager = new SimpleMemoryManager();
messageStore = new SimpleMessageStore();
messageIDMgr = new RotatingID(serverPeerID);
- txRepository =
- new TransactionRepository(persistenceManager, messageStore, serverPeerID);
+ txRepository = new TransactionRepository(persistenceManager, messageStore, serverPeerID);
messageCounterManager = new MessageCounterManager(messageCounterSamplePeriod);
clusterNotifier = new DefaultClusterNotifier();
@@ -271,17 +323,19 @@
clusterNotifier.registerListener(connFactoryJNDIMapper);
failoverWaiter = new FailoverWaiter(serverPeerID, failoverStartTimeout, failoverCompleteTimeout, txRepository);
clusterNotifier.registerListener(failoverWaiter);
-
+
if (suckerPassword == null)
- {
- suckerPassword = SecurityMetadataStore.DEFAULT_SUCKER_USER_PASSWORD;
- }
+ {
+ suckerPassword = SecurityMetadataStore.DEFAULT_SUCKER_USER_PASSWORD;
+ }
if (clusterPullConnectionFactoryName != null)
- {
- clusterConnectionManager = new ClusterConnectionManager(serverPeerID,
- clusterPullConnectionFactoryName, defaultPreserveOrdering,
- SecurityMetadataStore.SUCKER_USER, suckerPassword);
- clusterNotifier.registerListener(clusterConnectionManager);
+ {
+ clusterConnectionManager = new ClusterConnectionManager(serverPeerID,
+ clusterPullConnectionFactoryName,
+ defaultPreserveOrdering,
+ SecurityMetadataStore.SUCKER_USER,
+ suckerPassword);
+ clusterNotifier.registerListener(clusterConnectionManager);
}
// Start the wired components
@@ -293,7 +347,7 @@
connectionManager.start();
connectorManager.start();
memoryManager.start();
- messageStore.start();
+ messageStore.start();
securityStore.setSuckerPassword(suckerPassword);
securityStore.start();
txRepository.start();
@@ -302,9 +356,12 @@
clusterConnectionManager.start();
}
- // Note we do not start the message counter manager by default. This must be done
- // explicitly by the user by calling enableMessageCounters(). This is because message
- // counter history takes up growing memory to store the stats and could theoretically
+ // Note we do not start the message counter manager by default. This
+ // must be done
+ // explicitly by the user by calling enableMessageCounters(). This is
+ // because message
+ // counter history takes up growing memory to store the stats and could
+ // theoretically
// eventually cause the server to run out of RAM
txRepository.loadPreparedTransactions();
@@ -312,19 +369,23 @@
JMSWireFormat wf = new JMSWireFormat();
MarshalFactory.addMarshaller("jms", wf, wf);
- //Now everything is started we can tell the invocation handler to start handling invocations
- //We do this right at the end otherwise it can start handling invocations before we are properly started
+ // Now everything is started we can tell the invocation handler to
+ // start handling invocations
+ // We do this right at the end otherwise it can start handling
+ // invocations before we are properly started
JMSServerInvocationHandler.setClosed(false);
-
+
if (enableMessageCounters)
{
- startMessageCounters();
+ startMessageCounters();
}
started = true;
- log.info("JBoss Messaging " + getVersion().getProviderVersion() + " server [" +
- getServerPeerID()+ "] started");
+ log.info("JBoss Messaging " + getVersion().getProviderVersion() +
+ " server [" +
+ getServerPeerID() +
+ "] started");
}
catch (Throwable t)
{
@@ -343,9 +404,11 @@
started = false;
- //Tell the invocation handler we are closed - this is so we don't attempt to handle
- //any invocations when we are in a partial closing down state - which can give strange
- //"object not found with id" exceptions and stuff like that
+ // Tell the invocation handler we are closed - this is so we don't
+ // attempt to handle
+ // any invocations when we are in a partial closing down state - which
+ // can give strange
+ // "object not found with id" exceptions and stuff like that
JMSServerInvocationHandler.setClosed(true);
// Stop the wired components
@@ -367,7 +430,8 @@
messageStore.stop();
messageStore = null;
securityStore.stop();
- //securityStore = null; - if securitySTore is set to null, The ServerPeer won't survive a restart of the service (stop/start)
+ // securityStore = null; - if securitySTore is set to null, The
+ // ServerPeer won't survive a restart of the service (stop/start)
txRepository.stop();
txRepository = null;
messageCounterManager.stop();
@@ -393,7 +457,9 @@
}
}
- // JMX Attributes -------------------------------------------------------------------------------
+ // JMX Attributes
+ // ------------------------------------------------------------
+ // -------------------
// Plugins
@@ -469,7 +535,7 @@
return this;
}
- //read only JMX attributes
+ // read only JMX attributes
public String getJMSVersion()
{
@@ -506,7 +572,7 @@
return version.getProviderMinorVersion();
}
- //Read - write attributes
+ // Read - write attributes
public synchronized void setSecurityDomain(String securityDomain) throws Exception
{
@@ -534,7 +600,7 @@
{
return securityStore.getDefaultSecurityConfig();
}
-
+
public synchronized long getFailoverStartTimeout()
{
return this.failoverStartTimeout;
@@ -612,7 +678,7 @@
public String getClusterPullConnectionFactoryName()
{
- return clusterPullConnectionFactoryName;
+ return clusterPullConnectionFactoryName;
}
public void setClusterPullConnectionFactoryName(String name)
@@ -621,12 +687,12 @@
{
throw new IllegalStateException("Cannot set ClusterPullConnectionFactoryName while the service is running");
}
- this.clusterPullConnectionFactoryName = name;
+ this.clusterPullConnectionFactoryName = name;
}
public boolean isUseXAForMessagePull()
{
- return useXAForMessagePull;
+ return useXAForMessagePull;
}
public void setUseXAForMessagePull(boolean useXA) throws Exception
@@ -636,12 +702,12 @@
throw new IllegalStateException("Cannot set UseXAForMessagePull while the service is running");
}
- this.useXAForMessagePull = useXA;
+ this.useXAForMessagePull = useXA;
}
public boolean isDefaultPreserveOrdering()
{
- return defaultPreserveOrdering;
+ return defaultPreserveOrdering;
}
public void setDefaultPreserveOrdering(boolean preserve) throws Exception
@@ -651,17 +717,17 @@
throw new IllegalStateException("Cannot set DefaultPreserveOrdering while the service is running");
}
- this.defaultPreserveOrdering = preserve;
+ this.defaultPreserveOrdering = preserve;
}
public long getRecoverDeliveriesTimeout()
{
- return this.recoverDeliveriesTimeout;
+ return this.recoverDeliveriesTimeout;
}
public void setRecoverDeliveriesTimeout(long timeout)
{
- this.recoverDeliveriesTimeout = timeout;
+ this.recoverDeliveriesTimeout = timeout;
}
public synchronized void setServerPeerID(int serverPeerID)
@@ -714,55 +780,55 @@
public synchronized void setSuckerPassword(String password)
{
- if (started)
+ if (started)
{
throw new IllegalStateException("Cannot set SuckerPassword while the service is running");
}
- if (password == null)
- {
- throw new IllegalArgumentException("SuckerPassword cannot be null");
- }
+ if (password == null)
+ {
+ throw new IllegalArgumentException("SuckerPassword cannot be null");
+ }
- this.suckerPassword = password;
+ this.suckerPassword = password;
}
public void setStrictTck(boolean strictTck)
{
- this.strictTck = strictTck || strictTckProperty;
+ this.strictTck = strictTck || strictTckProperty;
}
public boolean isStrictTck()
{
- return strictTck || strictTckProperty;
+ return strictTck || strictTckProperty;
}
-
+
public RotatingID getMessageIDMgr()
{
return messageIDMgr;
}
-
- public synchronized void setEnableMessageCounters(boolean enable)
+
+ public synchronized void setEnableMessageCounters(boolean enable)
{
- if (started)
- {
- if (enableMessageCounters && !enable)
- {
- stopMessageCounters();
- }
- else if (!enableMessageCounters && enable)
- {
- startMessageCounters();
- }
- }
- enableMessageCounters = enable;
+ if (started)
+ {
+ if (enableMessageCounters && !enable)
+ {
+ stopMessageCounters();
+ }
+ else if (!enableMessageCounters && enable)
+ {
+ startMessageCounters();
+ }
+ }
+ enableMessageCounters = enable;
}
-
+
public boolean isEnableMessageCounters()
{
- return enableMessageCounters;
- }
-
+ return enableMessageCounters;
+ }
+
public void enableMessageCounters()
{
setEnableMessageCounters(true);
@@ -773,7 +839,9 @@
setEnableMessageCounters(false);
}
- // JMX Operations -------------------------------------------------------------------------------
+ // JMX Operations
+ // ------------------------------------------------------------
+ // -------------------
public String deployQueue(String name, String jndiName) throws Exception
{
@@ -899,27 +967,27 @@
{
List counters = getMessageCounters();
- Collections.sort(counters, new Comparator() {
- public int compare(Object o1, Object o2) {
- MessageCounter m1 = (MessageCounter)o1;
- MessageCounter m2 = (MessageCounter)o2;
- return m1.getDestinationName().compareTo(m2.getDestinationName());
- }
+ Collections.sort(counters, new Comparator()
+ {
+ public int compare(Object o1, Object o2)
+ {
+ MessageCounter m1 = (MessageCounter)o1;
+ MessageCounter m2 = (MessageCounter)o2;
+ return m1.getDestinationName().compareTo(m2.getDestinationName());
+ }
});
- String ret =
- "<table width=\"100%\" border=\"1\" cellpadding=\"1\" cellspacing=\"1\">"
- + "<tr>"
- + "<th>Type</th>"
- + "<th>Name</th>"
- + "<th>Subscription</th>"
- + "<th>Durable</th>"
- + "<th>Count</th>"
- + "<th>CountDelta</th>"
- + "<th>Depth</th>"
- + "<th>DepthDelta</th>"
- + "<th>Last Add</th>"
- + "</tr>";
+ String ret = "<table width=\"100%\" border=\"1\" cellpadding=\"1\" cellspacing=\"1\">" + "<tr>"
+ + "<th>Type</th>"
+ + "<th>Name</th>"
+ + "<th>Subscription</th>"
+ + "<th>Durable</th>"
+ + "<th>Count</th>"
+ + "<th>CountDelta</th>"
+ + "<th>Depth</th>"
+ + "<th>DepthDelta</th>"
+ + "<th>Last Add</th>"
+ + "</tr>";
String strNameLast = null;
String strTypeLast = null;
@@ -974,7 +1042,7 @@
// output destination counter data as HTML table row
// ( for topics with multiple subscriptions output
- // type + name field as rowspans, looks better )
+ // type + name field as rowspans, looks better )
if (strDestLast != null && strDestLast.equals(strDest))
{
// still same destination -> append destination subscription data
@@ -987,18 +1055,17 @@
if (strDestLast != null)
{
// store last destination data string
- ret += "<tr bgcolor=\"#"
- + ((i % 2) == 0 ? "FFFFFF" : "F0F0F0")
- + "\"><td rowspan=\""
- + destCount
- + "\">"
- + strTypeLast
- + "</td><td rowspan=\""
- + destCount
- + "\">"
- + strNameLast
- + "</td>"
- + destData;
+ ret += "<tr bgcolor=\"#" + ((i % 2) == 0 ? "FFFFFF" : "F0F0F0") +
+ "\"><td rowspan=\"" +
+ destCount +
+ "\">" +
+ strTypeLast +
+ "</td><td rowspan=\"" +
+ destCount +
+ "\">" +
+ strNameLast +
+ "</td>" +
+ destData;
destData = "";
}
@@ -1007,27 +1074,26 @@
}
// counter data row
- destData += "<td>"
- + strSub
- + "</td>"
- + "<td>"
- + strDurable
- + "</td>"
- + "<td>"
- + strCount
- + "</td>"
- + "<td>"
- + strCountDelta
- + "</td>"
- + "<td>"
- + strDepth
- + "</td>"
- + "<td>"
- + strDepthDelta
- + "</td>"
- + "<td>"
- + strDate
- + "</td>";
+ destData += "<td>" + strSub +
+ "</td>" +
+ "<td>" +
+ strDurable +
+ "</td>" +
+ "<td>" +
+ strCount +
+ "</td>" +
+ "<td>" +
+ strCountDelta +
+ "</td>" +
+ "<td>" +
+ strDepth +
+ "</td>" +
+ "<td>" +
+ strDepthDelta +
+ "</td>" +
+ "<td>" +
+ strDate +
+ "</td>";
// store current destination data for change detection
strTypeLast = strType;
@@ -1038,32 +1104,30 @@
if (strDestLast != null)
{
// store last module data string
- ret += "<tr bgcolor=\"#"
- + ((i % 2) == 0 ? "FFFFFF" : "F0F0F0")
- + "\"><td rowspan=\""
- + destCount
- + "\">"
- + strTypeLast
- + "</td><td rowspan=\""
- + destCount
- + "\">"
- + strNameLast
- + "</td>"
- + destData;
+ ret += "<tr bgcolor=\"#" + ((i % 2) == 0 ? "FFFFFF" : "F0F0F0") +
+ "\"><td rowspan=\"" +
+ destCount +
+ "\">" +
+ strTypeLast +
+ "</td><td rowspan=\"" +
+ destCount +
+ "\">" +
+ strNameLast +
+ "</td>" +
+ destData;
}
// append summation info
- ret += "<tr>"
- + "<td><![CDATA[ ]]></td><td><![CDATA[ ]]></td>"
- + "<td><![CDATA[ ]]></td><td><![CDATA[ ]]></td><td>"
- + countTotal
- + "</td><td>"
- + (countDeltaTotal == 0 ? "-" : Integer.toString(countDeltaTotal))
- + "</td><td>"
- + depthTotal
- + "</td><td>"
- + (depthDeltaTotal == 0 ? "-" : Integer.toString(depthDeltaTotal))
- + "</td><td>Total</td></tr></table>";
+ ret += "<tr>" + "<td><![CDATA[ ]]></td><td><![CDATA[ ]]></td>" +
+ "<td><![CDATA[ ]]></td><td><![CDATA[ ]]></td><td>" +
+ countTotal +
+ "</td><td>" +
+ (countDeltaTotal == 0 ? "-" : Integer.toString(countDeltaTotal)) +
+ "</td><td>" +
+ depthTotal +
+ "</td><td>" +
+ (depthDeltaTotal == 0 ? "-" : Integer.toString(depthDeltaTotal)) +
+ "</td><td>Total</td></tr></table>";
return ret;
}
@@ -1113,11 +1177,12 @@
out.println("<table><tr><td>ID</td><td>Host</td><td>User</td><td>#Sessions</td></tr>");
for (Iterator iter = endpoints.iterator(); iter.hasNext();)
{
- ServerConnectionEndpoint endpoint = (ServerConnectionEndpoint) iter.next();
+ ServerConnectionEndpoint endpoint = (ServerConnectionEndpoint)iter.next();
out.println("<tr>");
out.println("<td>" + endpoint.toString() + "</td>");
- out.println("<td>" + endpoint.getCallbackHandler().getCallbackClient().getInvoker().getLocator().getHost() + "</td>");
+ out.println("<td>" + endpoint.getCallbackHandler().getCallbackClient().getInvoker().getLocator().getHost() +
+ "</td>");
out.println("<td>" + endpoint.getUsername() + "</td>");
out.println("<td>" + endpoint.getSessions().size() + "</td>");
out.println("</tr>");
@@ -1125,11 +1190,12 @@
out.println("</table>");
-
return charArray.toString();
}
- // Public ---------------------------------------------------------------------------------------
+ // Public
+ // --------------------------------------------------------------------
+ // -------------------
public void resetAllSuckers()
{
@@ -1166,7 +1232,7 @@
public Collection getSessions()
{
- return sessions.values();
+ return sessions.values();
}
public void addSession(String id, ServerSessionEndpoint session)
@@ -1190,8 +1256,9 @@
{
ManagedQueue dest = null;
- // This can be null... JMXAccessor will return null if InstanceNotFoundException is caught
- dest = (ManagedQueue) JMXAccessor.getJMXAttributeOverSecurity(getServer(), defaultDLQObjectName, "Instance");
+ // This can be null... JMXAccessor will return null if
+ // InstanceNotFoundException is caught
+ dest = (ManagedQueue)JMXAccessor.getJMXAttributeOverSecurity(getServer(), defaultDLQObjectName, "Instance");
if (dest != null && dest.getName() != null)
{
@@ -1199,14 +1266,14 @@
if (binding == null)
{
- throw new IllegalStateException("Cannot find binding for queue " + dest.getName());
+ throw new IllegalStateException("Cannot find binding for queue " + dest.getName());
}
Queue queue = binding.queue;
if (queue.isActive())
{
- dlq = queue;
+ dlq = queue;
}
}
}
@@ -1225,27 +1292,29 @@
try
{
- dest = (ManagedQueue)JMXAccessor.getJMXAttributeOverSecurity(getServer(), defaultExpiryQueueObjectName, "Instance");
+ dest = (ManagedQueue)JMXAccessor.getJMXAttributeOverSecurity(getServer(),
+ defaultExpiryQueueObjectName,
+ "Instance");
}
catch (InstanceNotFoundException e)
{
- //Ok
+ // Ok
}
if (dest != null && dest.getName() != null)
{
- Binding binding = postOffice.getBindingForQueueName(dest.getName());
+ Binding binding = postOffice.getBindingForQueueName(dest.getName());
if (binding == null)
{
- throw new IllegalStateException("Cannot find binding for queue " + dest.getName());
+ throw new IllegalStateException("Cannot find binding for queue " + dest.getName());
}
Queue queue = binding.queue;
if (queue.isActive())
{
- expiryQueue = queue;
+ expiryQueue = queue;
}
}
}
@@ -1319,12 +1388,15 @@
public PostOffice getPostOfficeInstance() throws Exception
{
- // We get the reference lazily to avoid problems with MBean circular dependencies
+ // We get the reference lazily to avoid problems with MBean circular
+ // dependencies
if (postOffice == null)
{
postOffice = (PostOffice)JMXAccessor.getJMXAttributeOverSecurity(getServer(), postOfficeObjectName, "Instance");
- // We also inject the replicator dependency into the ConnectionFactoryJNDIMapper. This is
- // a bit messy but we have a circular dependency POJOContainer should be able to help us
+ // We also inject the replicator dependency into the
+ // ConnectionFactoryJNDIMapper. This is
+ // a bit messy but we have a circular dependency POJOContainer should
+ // be able to help us
// here. Yes, this is nasty.
if (postOffice.isClustered())
@@ -1355,12 +1427,12 @@
public ClusterNotifier getClusterNotifier()
{
- return clusterNotifier;
+ return clusterNotifier;
}
public FailoverWaiter getFailoverWaiter()
{
- return failoverWaiter;
+ return failoverWaiter;
}
public boolean isSupportsFailover()
@@ -1372,8 +1444,7 @@
{
if (started)
{
- throw new IllegalAccessException("supportsFailover can only be changed when " +
- "server peer is stopped");
+ throw new IllegalAccessException("supportsFailover can only be changed when " + "server peer is stopped");
}
this.supportsFailover = supportsFailover;
}
@@ -1383,26 +1454,32 @@
return "ServerPeer[" + getServerPeerID() + "]";
}
- // Package protected ----------------------------------------------------------------------------
+ // Package protected
+ // ----------------------------------------------------------
+ // ------------------
- // Protected ------------------------------------------------------------------------------------
+ // Protected
+ // ------------------------------------------------------------------
+ // ------------------
- // Private --------------------------------------------------------------------------------------
-
+ // Private
+ // --------------------------------------------------------------------
+ // ------------------
+
private void startMessageCounters()
{
- messageCounterManager.start();
+ messageCounterManager.start();
}
-
+
private void stopMessageCounters()
{
- messageCounterManager.stop();
-
+ messageCounterManager.stop();
+
messageCounterManager.resetAllCounters();
messageCounterManager.resetAllCounterHistories();
}
-
+
private void loadServerAOPConfig() throws Exception
{
URL url = this.getClass().getClassLoader().getResource("aop-messaging-server.xml");
@@ -1417,8 +1494,10 @@
private void loadClientAOPConfig() throws Exception
{
- // Note the file is called aop-messaging-client.xml NOT messaging-client-aop.xml. This is
- // because the JBoss will automatically deploy any files ending with aop.xml; we do not want
+ // Note the file is called aop-messaging-client.xml NOT
+ // messaging-client-aop.xml. This is
+ // because the JBoss will automatically deploy any files ending with
+ // aop.xml; we do not want
// this to happen for the client config
URL url = this.getClass().getClassLoader().getResource("aop-messaging-client.xml");
@@ -1448,73 +1527,107 @@
}
}
- private String deployDestinationDefault(boolean isQueue, String name, String jndiName)
- throws Exception
+ private String deployDestinationDefault(boolean isQueue, String name, String jndiName) throws Exception
{
//
- // TODO - THIS IS A TEMPORARY IMPLEMENTATION; WILL BE REPLACED WITH INTEGRATION-CONSISTENT ONE
- // TODO - if I find a way not using UnifiedClassLoader3 directly, then get rid of
- // <path refid="jboss.jmx.classpath"/> from jms/build.xml dependentmodule.classpath
+ // TODO - THIS IS A TEMPORARY IMPLEMENTATION; WILL BE REPLACED WITH
+ // INTEGRATION-CONSISTENT ONE
+ // TODO - if I find a way not using UnifiedClassLoader3 directly, then get
+ // rid of
+ // <path refid="jboss.jmx.classpath"/> from jms/build.xml
+ // dependentmodule.classpath
//
- //TODO - Yes this is super-ugly - there must be an easier way of doing it
- //also in LocalTestServer is doing the same thing in a slightly different way
- //this should be combined
+ // TODO - Yes this is super-ugly - there must be an easier way of doing it
+ // also in LocalTestServer is doing the same thing in a slightly different
+ // way
+ // this should be combined
String destType = isQueue ? "Queue" : "Topic";
String className = "org.jboss.jms.server.destination." + destType + "Service";
- String ons ="jboss.messaging.destination:service="+ destType + ",name=" + name;
+ String ons = "jboss.messaging.destination:service=" + destType + ",name=" + name;
ObjectName on = new ObjectName(ons);
- String destinationMBeanConfig =
- "<mbean code=\"" + className + "\" " +
- " name=\"" + ons + "\" " +
- " xmbean-dd=\"xmdesc/" + destType + "-xmbean.xml\">\n" +
- " <constructor>" +
- " <arg type=\"boolean\" value=\"true\"/>" +
- " </constructor>" +
- "</mbean>";
+ String destinationMBeanConfig = "<mbean code=\"" + className +
+ "\" " +
+ " name=\"" +
+ ons +
+ "\" " +
+ " xmbean-dd=\"xmdesc/" +
+ destType +
+ "-xmbean.xml\">\n" +
+ " <constructor>" +
+ " <arg type=\"boolean\" value=\"true\"/>" +
+ " </constructor>" +
+ "</mbean>";
return deployDestinationInternal(destinationMBeanConfig, on, jndiName, false, -1, -1, -1);
}
- private String deployDestination(boolean isQueue, String name, String jndiName, int fullSize,
- int pageSize, int downCacheSize) throws Exception
+ private String deployDestination(boolean isQueue,
+ String name,
+ String jndiName,
+ int fullSize,
+ int pageSize,
+ int downCacheSize) throws Exception
{
//
- // TODO - THIS IS A TEMPORARY IMPLEMENTATION; WILL BE REPLACED WITH INTEGRATION-CONSISTENT ONE
- // TODO - if I find a way not using UnifiedClassLoader3 directly, then get rid of
- // <path refid="jboss.jmx.classpath"/> from jms/build.xml dependentmodule.classpath
+ // TODO - THIS IS A TEMPORARY IMPLEMENTATION; WILL BE REPLACED WITH
+ // INTEGRATION-CONSISTENT ONE
+ // TODO - if I find a way not using UnifiedClassLoader3 directly, then get
+ // rid of
+ // <path refid="jboss.jmx.classpath"/> from jms/build.xml
+ // dependentmodule.classpath
//
String destType = isQueue ? "Queue" : "Topic";
String className = "org.jboss.jms.server.destination." + destType + "Service";
- String ons ="jboss.messaging.destination:service="+ destType + ",name=" + name;
+ String ons = "jboss.messaging.destination:service=" + destType + ",name=" + name;
ObjectName on = new ObjectName(ons);
- String destinationMBeanConfig =
- "<mbean code=\"" + className + "\" " +
- " name=\"" + ons + "\" " +
- " xmbean-dd=\"xmdesc/" + destType + "-xmbean.xml\">\n" +
- " <constructor>" +
- " <arg type=\"boolean\" value=\"true\"/>" +
- " </constructor>" +
- " <attribute name=\"FullSize\">" + fullSize + "</attribute>" +
- " <attribute name=\"PageSize\">" + pageSize + "</attribute>" +
- " <attribute name=\"DownCacheSize\">" + downCacheSize + "</attribute>" +
- "</mbean>";
+ String destinationMBeanConfig = "<mbean code=\"" + className +
+ "\" " +
+ " name=\"" +
+ ons +
+ "\" " +
+ " xmbean-dd=\"xmdesc/" +
+ destType +
+ "-xmbean.xml\">\n" +
+ " <constructor>" +
+ " <arg type=\"boolean\" value=\"true\"/>" +
+ " </constructor>" +
+ " <attribute name=\"FullSize\">" +
+ fullSize +
+ "</attribute>" +
+ " <attribute name=\"PageSize\">" +
+ pageSize +
+ "</attribute>" +
+ " <attribute name=\"DownCacheSize\">" +
+ downCacheSize +
+ "</attribute>" +
+ "</mbean>";
- return deployDestinationInternal(destinationMBeanConfig, on, jndiName, true, fullSize,
- pageSize, downCacheSize);
+ return deployDestinationInternal(destinationMBeanConfig, on, jndiName, true, fullSize, pageSize, downCacheSize);
}
- private String deployDestinationInternal(String destinationMBeanConfig, ObjectName on,
- String jndiName, boolean params, int fullSize,
- int pageSize, int downCacheSize) throws Exception
+ private String deployDestinationInternal(String destinationMBeanConfig,
+ ObjectName on,
+ String jndiName,
+ boolean params,
+ int fullSize,
+ int pageSize,
+ int downCacheSize) throws Exception
{
- log.trace("Deploying destination" + destinationMBeanConfig + " jndiName: " + jndiName +
- "fullSize: " + fullSize + " pageSize: " + pageSize + " downCacheSize: " + downCacheSize);
+ log.trace("Deploying destination" + destinationMBeanConfig +
+ " jndiName: " +
+ jndiName +
+ "fullSize: " +
+ fullSize +
+ " pageSize: " +
+ pageSize +
+ " downCacheSize: " +
+ downCacheSize);
MBeanServer mbeanServer = getServer();
@@ -1546,20 +1659,19 @@
return (String)JMXAccessor.getJMXAttributeOverSecurity(mbeanServer, on, "JNDIName");
}
-
-
/*
* Undeploy the MBean but don't delete the underlying data
*/
private boolean undeployDestination(boolean isQueue, String name) throws Exception
{
String destType = isQueue ? "Queue" : "Topic";
- String ons ="jboss.messaging.destination:service=" + destType + ",name=" + name;
+ String ons = "jboss.messaging.destination:service=" + destType + ",name=" + name;
ObjectName on = new ObjectName(ons);
MBeanServer mbeanServer = getServer();
- // we can only undeploy destinations that exist AND that have been created programatically
+ // we can only undeploy destinations that exist AND that have been created
+ // programatically
if (!mbeanServer.isRegistered(on))
{
return false;
@@ -1582,12 +1694,13 @@
private boolean destroyDestination(boolean isQueue, String name) throws Throwable
{
String destType = isQueue ? "Queue" : "Topic";
- String ons ="jboss.messaging.destination:service=" + destType + ",name=" + name;
+ String ons = "jboss.messaging.destination:service=" + destType + ",name=" + name;
ObjectName on = new ObjectName(ons);
MBeanServer mbeanServer = getServer();
- // we can only destroy destinations that exist AND that have been created programatically
+ // we can only destroy destinations that exist AND that have been created
+ // programatically
if (!mbeanServer.isRegistered(on))
{
return false;
@@ -1606,13 +1719,13 @@
queue.removeAllReferences();
}
- //undeploy the mbean
+ // undeploy the mbean
if (!undeployDestination(isQueue, name))
{
return false;
}
- //Unbind the destination's queues
+ // Unbind the destination's queues
while (iter.hasNext())
{
@@ -1620,7 +1733,7 @@
queue.removeAllReferences();
- //Durable subs need to be removed on all nodes
+ // Durable subs need to be removed on all nodes
boolean all = !isQueue && queue.isRecoverable();
postOffice.removeBinding(queue.getName(), all);
@@ -1629,8 +1742,8 @@
return true;
}
+ // Inner classes
+ // --------------------------------------------------------------
+ // ------------------
- // Inner classes --------------------------------------------------------------------------------
-
-
}
Modified: branches/Branch_JBossMessaging_1_4_0_SP3_CP/src/main/org/jboss/messaging/core/impl/NullPersistenceManager.java
===================================================================
--- branches/Branch_JBossMessaging_1_4_0_SP3_CP/src/main/org/jboss/messaging/core/impl/NullPersistenceManager.java 2008-09-17 20:27:12 UTC (rev 4975)
+++ branches/Branch_JBossMessaging_1_4_0_SP3_CP/src/main/org/jboss/messaging/core/impl/NullPersistenceManager.java 2008-09-18 02:08:06 UTC (rev 4976)
@@ -1,3 +1,24 @@
+/*
+ * JBoss, Home of Professional Open Source Copyright 2005, JBoss Inc., and
+ * individual contributors as indicated by the @authors tag. See the
+ * copyright.txt in the distribution for a full listing of individual
+ * contributors.
+ *
+ * This is free software; you can redistribute it and/or modify it under the
+ * terms of the GNU Lesser General Public License as published by the Free
+ * Software Foundation; either version 2.1 of the License, or (at your option)
+ * any later version.
+ *
+ * This software is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
+ * FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more
+ * details.
+ *
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with this software; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA, or see the FSF
+ * site: http://www.fsf.org.
+ */
package org.jboss.messaging.core.impl;
import java.util.Collections;
@@ -4,151 +25,270 @@
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.atomic.AtomicLong;
import org.jboss.messaging.core.contract.MessageReference;
import org.jboss.messaging.core.contract.PersistenceManager;
import org.jboss.messaging.core.impl.tx.Transaction;
-/**
+/*
*
* A NullPersistenceManager
*
* @author <a href="mailto:tim.fox at jboss.com">Tim Fox</a>
- *
*/
public class NullPersistenceManager implements PersistenceManager
{
- private ConcurrentMap<String, AtomicLong> counters = new ConcurrentHashMap<String, AtomicLong>();
+ private static final int MAX_PEER_ID = 255;
- public void addReference(long channelID, MessageReference ref, Transaction tx)
- throws Exception
- {
- //NOOP
- }
+ private ConcurrentMap<String, IDCounter> counters = new ConcurrentHashMap<String, IDCounter>();
- public List getMessageChannelPairAcksForTx(long transactionId)
- throws Exception
- {
- return Collections.EMPTY_LIST;
- }
+ private int peerID; // 0 - 255
- public List getMessageChannelPairRefsForTx(long transactionId)
- throws Exception
- {
- return Collections.EMPTY_LIST;
- }
+ private long timeMark;
- public List getMessages(List messageIds) throws Exception
- {
- return Collections.EMPTY_LIST;
- }
+ public void addReference(long channelID, MessageReference ref, Transaction tx) throws Exception
+ {
+ // NOOP
+ }
- public List getPagedReferenceInfos(long channelID, long orderStart,
- int number) throws Exception
- {
- return Collections.EMPTY_LIST;
- }
+ public List getMessageChannelPairAcksForTx(long transactionId) throws Exception
+ {
+ return Collections.EMPTY_LIST;
+ }
- public boolean idExists(String messageID) throws Exception
- {
- return false;
- }
+ public List getMessageChannelPairRefsForTx(long transactionId) throws Exception
+ {
+ return Collections.EMPTY_LIST;
+ }
- public InitialLoadInfo loadFromStart(long channelID, int fullSize)
- throws Exception
- {
- return new InitialLoadInfo(null, null, Collections.EMPTY_LIST);
- }
+ public List getMessages(List messageIds) throws Exception
+ {
+ return Collections.EMPTY_LIST;
+ }
- public InitialLoadInfo mergeAndLoad(long fromChannelID, long toChannelID,
- int numberToLoad, long firstPagingOrder, long nextPagingOrder)
- throws Exception
- {
- return new InitialLoadInfo(null, null, Collections.EMPTY_LIST);
- }
+ public List getPagedReferenceInfos(long channelID, long orderStart, int number) throws Exception
+ {
+ return Collections.EMPTY_LIST;
+ }
- public void mergeTransactions(int fromNodeID, int toNodeID) throws Exception
- {
- //NOOP
- }
+ public boolean idExists(String messageID) throws Exception
+ {
+ return false;
+ }
- public void moveReference(long sourceChannelID, long destChannelID,
- MessageReference ref) throws Exception
- {
- //NOOP
- }
+ public InitialLoadInfo loadFromStart(long channelID, int fullSize) throws Exception
+ {
+ return new InitialLoadInfo(null, null, Collections.EMPTY_LIST);
+ }
- public void pageReferences(long channelID, List references, boolean paged)
- throws Exception
- {
- //NOOP
- }
+ public InitialLoadInfo mergeAndLoad(long fromChannelID,
+ long toChannelID,
+ int numberToLoad,
+ long firstPagingOrder,
+ long nextPagingOrder) throws Exception
+ {
+ return new InitialLoadInfo(null, null, Collections.EMPTY_LIST);
+ }
- public void removeDepagedReferences(long channelID, List refs)
- throws Exception
- {
- //NOOP
- }
+ public void mergeTransactions(int fromNodeID, int toNodeID) throws Exception
+ {
+ // NOOP
+ }
- public void removeReference(long channelID, MessageReference ref,
- Transaction tx) throws Exception
- {
- //NOOP
- }
-
- public long reserveIDBlock(String counterName, int size) throws Exception
- {
- AtomicLong counter = counters.get(counterName);
-
- if (counter == null)
- {
- counter = new AtomicLong(0);
-
- AtomicLong oldCounter = counters.putIfAbsent(counterName, counter);
-
- if (oldCounter != null)
- {
- counter = oldCounter;
- }
- }
-
- long idStart = counter.addAndGet(size) - size;
-
- return idStart;
- }
+ public void moveReference(long sourceChannelID, long destChannelID, MessageReference ref) throws Exception
+ {
+ // NOOP
+ }
- public List retrievePreparedTransactions() throws Exception
- {
- return Collections.EMPTY_LIST;
- }
+ public void pageReferences(long channelID, List references, boolean paged) throws Exception
+ {
+ // NOOP
+ }
- public void updateDeliveryCount(long channelID, MessageReference ref)
- throws Exception
- {
- //NOOP
- }
+ public void removeDepagedReferences(long channelID, List refs) throws Exception
+ {
+ // NOOP
+ }
- public void updatePageOrder(long channelID, List references)
- throws Exception
- {
- ///NOOP
- }
+ public void removeReference(long channelID, MessageReference ref, Transaction tx) throws Exception
+ {
+ // NOOP
+ }
- public void updateReferencesNotPagedInRange(long channelID, long orderStart,
- long orderEnd, long num) throws Exception
- {
- //NOOP
- }
+ public long reserveIDBlock(String counterName, int size) throws Exception
+ {
+ checkServerID();
+ IDCounter counter = counters.get(counterName);
- public void start() throws Exception
- {
- //NOOP
- }
+ if (counter == null)
+ {
+ synchronized (counters)
+ {
+ counter = counters.get(counterName);
+ if (counter == null)
+ {
+ counter = new IDCounter(this);
+ counters.put(counterName, counter);
+ }
+ }
+ }
+ long idStart = counter.reserveAndGetNextId(size);
- public void stop() throws Exception
- {
- //NOOP
- }
+ return idStart;
+ }
+ private void checkServerID() throws Exception
+ {
+ if (peerID > MAX_PEER_ID)
+ {
+ throw new Exception("ServerPeerID " + peerID + " exceeding 255");
+ }
+ if (peerID < 0)
+ {
+ throw new Exception("ServerPeerID cannot have negative values");
+ }
+ }
+
+ public List retrievePreparedTransactions() throws Exception
+ {
+ return Collections.EMPTY_LIST;
+ }
+
+ public void updateDeliveryCount(long channelID, MessageReference ref) throws Exception
+ {
+ // NOOP
+ }
+
+ public void updatePageOrder(long channelID, List references) throws Exception
+ {
+ // /NOOP
+ }
+
+ public void updateReferencesNotPagedInRange(long channelID, long orderStart, long orderEnd, long num) throws Exception
+ {
+ // NOOP
+ }
+
+ public void start() throws Exception
+ {
+ // NOOP
+ }
+
+ public void stop() throws Exception
+ {
+ // NOOP
+ }
+
+ public void initCounter(int serverPeerID, long serverStartTime)
+ {
+ setPeerID(serverPeerID);
+ setTimeMark(serverStartTime);
+ }
+
+ public void setPeerID(int peerID)
+ {
+ this.peerID = peerID;
+ }
+
+ public int getPeerID()
+ {
+ return peerID;
+ }
+
+ public void setTimeMark(long timeMark)
+ {
+ this.timeMark = timeMark;
+ }
+
+ public long getTimeMark()
+ {
+ return timeMark;
+ }
+
}
+
+class IDCounter
+{
+
+ private NullPersistenceManager manager;
+
+ private short counter;
+
+ private long tmMark;
+
+ private long peerIDBit;
+
+ public IDCounter(NullPersistenceManager pManager)
+ {
+ manager = pManager;
+ counter = 0;
+ tmMark = manager.getTimeMark() & MASK_TIME;
+ peerIDBit = (((long)manager.getPeerID()) & MASK_SERVER_PEER_ID) << 56;
+ recalculate();// avoid quick restart conflict
+ }
+
+ /**
+ * for each named counter, we generate it using the following algorithm:
+ * <8-bit ServerPeerID> + <40-bit time bit> + <16-bit counter>
+ * the 16-bit counter starts from zero and increases by 1. If the counter
+ * wraps to zero, we re-calculate the time using current time. Check will be
+ * performed when the calculated current time value is the same as the old
+ * value. If so, sleep for a while and get the current time value again.
+ * That will make sure the generated ID will always be unique even
+ * if server peer gets restarted from a previous shutting down.
+ *
+ * Note: the block size is limited by the counter (which is a short).
+ * if the block size is greater than Short.MAX_VALUE - counter, we
+ * will discard the counter and do a recalculate op because we cannot
+ * return a consecutive block of long values.
+ *
+ * @param size : size of the block to be reserved.
+ * @return
+ */
+ private static final long MASK_SERVER_PEER_ID = 0x00000000000000FFL;
+
+ private static final long MASK_TIME = 0x00000FFFFFFFFFF0L;
+
+ public synchronized long reserveAndGetNextId(int size) throws Exception
+ {
+ if (size > Short.MAX_VALUE)
+ {
+ throw new Exception("The block size exceeds " + Short.MAX_VALUE);
+ }
+ if (size > (Short.MAX_VALUE - counter))
+ {
+ recalculate();
+ }
+ long nextID = assembleID();
+ counter += size;
+
+ return nextID;
+ }
+
+ private long assembleID()
+ {
+ long id = peerIDBit;
+ id += tmMark << 12;
+ id += counter;
+ return id;
+ }
+
+ private void recalculate()
+ {
+ counter = 0;
+ long newTm = System.currentTimeMillis() & MASK_TIME;
+
+ while (newTm == tmMark)
+ {
+ try
+ {
+ Thread.sleep(20);
+ }
+ catch (InterruptedException e)
+ {
+ }
+ newTm = System.currentTimeMillis() & MASK_TIME;
+ }
+ tmMark = newTm;
+ }
+}
Modified: branches/Branch_JBossMessaging_1_4_0_SP3_CP/src/main/org/jboss/messaging/core/jmx/MessagingPostOfficeService.java
===================================================================
--- branches/Branch_JBossMessaging_1_4_0_SP3_CP/src/main/org/jboss/messaging/core/jmx/MessagingPostOfficeService.java 2008-09-17 20:27:12 UTC (rev 4975)
+++ branches/Branch_JBossMessaging_1_4_0_SP3_CP/src/main/org/jboss/messaging/core/jmx/MessagingPostOfficeService.java 2008-09-18 02:08:06 UTC (rev 4976)
@@ -1,23 +1,23 @@
/*
- * JBoss, Home of Professional Open Source
- * Copyright 2005, JBoss Inc., and individual contributors as indicated
- * by the @authors tag. See the copyright.txt in the distribution for a
- * full listing of individual contributors.
- *
- * This is free software; you can redistribute it and/or modify it
- * under the terms of the GNU Lesser General Public License as
- * published by the Free Software Foundation; either version 2.1 of
- * the License, or (at your option) any later version.
- *
- * This software is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
- * Lesser General Public License for more details.
- *
- * You should have received a copy of the GNU Lesser General Public
- * License along with this software; if not, write to the Free
- * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
- * 02110-1301 USA, or see the FSF site: http://www.fsf.org.
+ * JBoss, Home of Professional Open Source Copyright 2005, JBoss Inc., and
+ * individual contributors as indicated by the @authors tag. See the
+ * copyright.txt in the distribution for a full listing of individual
+ * contributors.
+ *
+ * This is free software; you can redistribute it and/or modify it under the
+ * terms of the GNU Lesser General Public License as published by the Free
+ * Software Foundation; either version 2.1 of the License, or (at your option)
+ * any later version.
+ *
+ * This software is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
+ * FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more
+ * details.
+ *
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with this software; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA, or see the FSF
+ * site: http://www.fsf.org.
*/
package org.jboss.messaging.core.jmx;
@@ -49,17 +49,18 @@
import org.jboss.messaging.util.JMXAccessor;
import org.w3c.dom.Element;
-/**
+/*
* A MessagingPostOfficeService
*
* MBean wrapper for a messaging post office
- *
+ *
* @author <a href="mailto:tim.fox at jboss.com">Tim Fox</a>
+ *
* @author <a href="mailto:ovidiu at feodorov.com">Ovidiu Feodorov</a>
+ *
* @version <tt>$Revision: 2684 $</tt>
- *
+ *
* $Id: ClusteredPostOfficeService.java 2684 2007-05-15 07:31:30Z timfox $
- *
*/
public class MessagingPostOfficeService extends JDBCServiceSupport
{
@@ -73,31 +74,31 @@
// This group of properties is used on JGroups Channel configuration
private Element controlChannelConfig;
-
+
private Element dataChannelConfig;
-
+
private ObjectName channelFactoryName;
-
+
private String controlChannelName;
-
+
private String dataChannelName;
-
+
private String channelPartitionName;
private ObjectName serverPeerObjectName;
private String officeName;
-
+
private long stateTimeout = 5000;
-
+
private long castTimeout = 5000;
-
+
private String groupName;
-
+
private boolean clustered;
-
+
private int maxConcurrentReplications = 25;
-
+
private boolean failoverOnNodeLeave;
private MessagingPostOffice postOffice;
@@ -108,20 +109,18 @@
public MessagingComponent getInstance()
{
+ Runtime.getRuntime();
return postOffice;
}
// NotificationBroadcaster implementation ------------------------
- public void addNotificationListener(NotificationListener listener,
- NotificationFilter filter,
- Object object) throws IllegalArgumentException
+ public void addNotificationListener(NotificationListener listener, NotificationFilter filter, Object object) throws IllegalArgumentException
{
postOffice.addNotificationListener(listener, filter, object);
}
- public void removeNotificationListener(NotificationListener listener)
- throws ListenerNotFoundException
+ public void removeNotificationListener(NotificationListener listener) throws ListenerNotFoundException
{
postOffice.removeNotificationListener(listener);
}
@@ -131,7 +130,6 @@
return postOffice.getNotificationInfo();
}
-
// MBean attributes ----------------------------------------------
public synchronized ObjectName getServerPeer()
@@ -293,61 +291,60 @@
{
return groupName;
}
-
+
public boolean isClustered()
{
- return clustered;
+ return clustered;
}
-
+
public void setClustered(boolean clustered)
{
- if (started)
- {
- log.warn("Cannot set attribute when service is started");
- return;
- }
- this.clustered = clustered;
+ if (started)
+ {
+ log.warn("Cannot set attribute when service is started");
+ return;
+ }
+ this.clustered = clustered;
}
-
+
public int getMaxConcurrentReplications()
{
- return maxConcurrentReplications;
+ return maxConcurrentReplications;
}
-
+
public void setMaxConcurrentReplications(int number)
{
- if (started)
+ if (started)
{
log.warn("Cannot set attribute when service is started");
return;
}
- this.maxConcurrentReplications = number;
+ this.maxConcurrentReplications = number;
}
-
- public boolean isFailoverOnNodeLeave()
- {
- return failoverOnNodeLeave;
- }
- public void setFailoverOnNodeLeave(boolean failoverOnNodeLeave)
- {
- if (started)
+ public boolean isFailoverOnNodeLeave()
+ {
+ return failoverOnNodeLeave;
+ }
+
+ public void setFailoverOnNodeLeave(boolean failoverOnNodeLeave)
+ {
+ if (started)
{
log.warn("Cannot set attribute when service is started");
return;
}
- this.failoverOnNodeLeave = failoverOnNodeLeave;
- }
+ this.failoverOnNodeLeave = failoverOnNodeLeave;
+ }
-
public String listBindings()
{
return postOffice.printBindingInformation();
}
-
+
public Set getNodeIDView()
{
- return postOffice.nodeIDView();
+ return postOffice.nodeIDView();
}
// Public --------------------------------------------------------
@@ -369,29 +366,31 @@
{
TransactionManager tm = getTransactionManagerReference();
- ServerPeer serverPeer = (ServerPeer) JMXAccessor.getJMXAttributeOverSecurity(server, serverPeerObjectName, "Instance");
+ ServerPeer serverPeer = (ServerPeer)JMXAccessor.getJMXAttributeOverSecurity(server,
+ serverPeerObjectName,
+ "Instance");
MessageStore ms = serverPeer.getMessageStore();
-
+
PersistenceManager pm = serverPeer.getPersistenceManagerInstance();
-
+
TransactionRepository tr = serverPeer.getTxRepository();
-
+
IDManager idManager = serverPeer.getChannelIDManager();
-
+
int nodeId = serverPeer.getServerPeerID();
-
+
ClusterNotifier clusterNotifier = serverPeer.getClusterNotifier();
ConditionFactory cf = new JMSConditionFactory();
-
+
FilterFactory ff = new SelectorFactory();
-
+
if (clustered)
- {
+ {
ChannelFactory jChannelFactory = null;
- if (channelFactoryName != null)
+ if (channelFactoryName != null)
{
Object info = null;
try
@@ -401,7 +400,8 @@
catch (Exception e)
{
// log.error("Error", e);
- // noop... means we couldn't find the channel hence we should use regular
+ // noop... means we couldn't find the channel hence we should
+ // use regular
// XMLChannelFactories
}
@@ -409,9 +409,11 @@
{
log.debug(this + " uses MultiplexerJChannelFactory");
- jChannelFactory =
- new MultiplexerChannelFactory(server, channelFactoryName, channelPartitionName,
- controlChannelName, dataChannelName);
+ jChannelFactory = new MultiplexerChannelFactory(server,
+ channelFactoryName,
+ channelPartitionName,
+ controlChannelName,
+ dataChannelName);
}
else
{
@@ -424,28 +426,43 @@
log.debug(this + " uses XMLJChannelFactory");
jChannelFactory = new XMLChannelFactory(controlChannelConfig, dataChannelConfig);
}
-
- postOffice = new MessagingPostOffice(ds, tm, sqlProperties,
- createTablesOnStartup,
- nodeId, officeName, ms,
- pm,
- tr, ff, cf, idManager,
- clusterNotifier,
- groupName,
- jChannelFactory,
- stateTimeout, castTimeout,
- serverPeer.isSupportsFailover(),
- maxConcurrentReplications,
- failoverOnNodeLeave);
+
+ postOffice = new MessagingPostOffice(ds,
+ tm,
+ sqlProperties,
+ createTablesOnStartup,
+ nodeId,
+ officeName,
+ ms,
+ pm,
+ tr,
+ ff,
+ cf,
+ idManager,
+ clusterNotifier,
+ groupName,
+ jChannelFactory,
+ stateTimeout,
+ castTimeout,
+ serverPeer.isSupportsFailover(),
+ maxConcurrentReplications,
+ failoverOnNodeLeave);
}
else
{
- postOffice = new MessagingPostOffice(ds, tm, sqlProperties,
- createTablesOnStartup,
- nodeId, officeName, ms,
- pm,
- tr, ff, cf, idManager,
- clusterNotifier);
+ postOffice = new MessagingPostOffice(ds,
+ tm,
+ sqlProperties,
+ createTablesOnStartup,
+ nodeId,
+ officeName,
+ ms,
+ pm,
+ tr,
+ ff,
+ cf,
+ idManager,
+ clusterNotifier);
}
postOffice.start();
@@ -490,4 +507,3 @@
// Inner classes -------------------------------------------------
}
-
Modified: branches/Branch_JBossMessaging_1_4_0_SP3_CP/tests/etc/container.xml
===================================================================
--- branches/Branch_JBossMessaging_1_4_0_SP3_CP/tests/etc/container.xml 2008-09-17 20:27:12 UTC (rev 4975)
+++ branches/Branch_JBossMessaging_1_4_0_SP3_CP/tests/etc/container.xml 2008-09-18 02:08:06 UTC (rev 4976)
@@ -13,6 +13,12 @@
<database-configurations>
+ <database-configuration name="null">
+ <url>jdbc:hsqldb:mem:test</url>
+ <driver>org.hsqldb.jdbcDriver</driver>
+ <username>sa</username>
+ </database-configuration>
+
<database-configuration name="hsqldb">
<url>jdbc:hsqldb:mem:test</url>
<driver>org.hsqldb.jdbcDriver</driver>
Added: branches/Branch_JBossMessaging_1_4_0_SP3_CP/tests/src/org/jboss/test/messaging/core/impl/NullPersistenceManagerTest.java
===================================================================
--- branches/Branch_JBossMessaging_1_4_0_SP3_CP/tests/src/org/jboss/test/messaging/core/impl/NullPersistenceManagerTest.java (rev 0)
+++ branches/Branch_JBossMessaging_1_4_0_SP3_CP/tests/src/org/jboss/test/messaging/core/impl/NullPersistenceManagerTest.java 2008-09-18 02:08:06 UTC (rev 4976)
@@ -0,0 +1,157 @@
+/*
+ * JBoss, Home of Professional Open Source Copyright 2005, JBoss Inc., and
+ * individual contributors as indicated by the @authors tag. See the
+ * copyright.txt in the distribution for a full listing of individual
+ * contributors.
+ *
+ * This is free software; you can redistribute it and/or modify it under the
+ * terms of the GNU Lesser General Public License as published by the Free
+ * Software Foundation; either version 2.1 of the License, or (at your option)
+ * any later version.
+ *
+ * This software is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
+ * FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more
+ * details.
+ *
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with this software; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA, or see the FSF
+ * site: http://www.fsf.org.
+ */
+package org.jboss.test.messaging.core.impl;
+
+import java.util.Hashtable;
+import java.util.Random;
+
+import org.jboss.messaging.core.impl.NullPersistenceManager;
+import org.jboss.test.messaging.MessagingTestCase;
+
+public class NullPersistenceManagerTest extends MessagingTestCase
+{
+
+ public NullPersistenceManagerTest(String name)
+ {
+ super(name);
+ }
+
+ /**
+ * To make sure that
+ * 1. the server peer ID must be between 0 and 255
+ * 2. the size should not exceed Short.MAX_VALUE
+ * 3. the block generated should be unique even
+ * after the manager was reset due to failures.
+ */
+ public void testReserveIDBlock()
+ {
+ NullPersistenceManager manager = new NullPersistenceManager();
+ manager.setPeerID(256);
+ manager.setTimeMark(System.currentTimeMillis());
+
+ try
+ {
+ manager.reserveIDBlock("testCounter", 255);
+ fail("Failed to throw exception when server peer ID exceeding 255.");
+ }
+ catch (Exception e)
+ {
+ log.debug(e.getMessage());
+ }
+
+ manager.setPeerID(-1);
+ try
+ {
+ manager.reserveIDBlock("testCounter", 255);
+ fail("Failed to throw exception when server peer ID negative.");
+ }
+ catch (Exception e)
+ {
+ log.debug(e.getMessage());
+ }
+
+ manager.setPeerID(255);
+ try
+ {
+ manager.reserveIDBlock("testCounter", Short.MAX_VALUE + 1);
+ fail("Failed to throw exception when block size exceeds Short.MAX_VALUE");
+ }
+ catch (Exception e)
+ {
+ log.debug(e.getMessage());
+ }
+
+ long nextID = 0;
+ int ftimes = 0;
+ Hashtable<String, Integer> data = new Hashtable<String, Integer>();
+ try
+ {
+ for (int i = 0; i < 500; ++i)
+ {
+ int szblock = getRandomBlocksize();
+ nextID = manager.reserveIDBlock("testCounter", szblock);
+ // simulate failure
+ if (getFailure(i))
+ {
+ manager = new NullPersistenceManager();
+ manager.setPeerID(255);
+ manager.setTimeMark(System.currentTimeMillis());
+ ftimes++;
+ }
+ checkData(data, nextID, szblock);
+ }
+ }
+ catch (Exception e)
+ {
+ fail("Exception calling reserveIDBlock()");
+ log.error(e);
+ }
+ log.debug("failure times: " + ftimes);
+ System.out.println("data in set: " + data.size());
+ System.out.println("failure times: " + ftimes);
+ System.gc();
+
+ }
+
+ private void checkData(Hashtable<String, Integer> data, long nID, int szblock)
+ {
+ long id = nID;
+ for (int i = 0; i < szblock; ++i)
+ {
+ // every ID should be unique
+ String key = Long.toHexString(id);
+ assertNull(data.get(key));
+ data.put(key, szblock);
+ id++;
+ }
+
+ }
+
+ // generate a number big enough to cause a wrap to happen.
+ private int getRandomBlocksize()
+ {
+ int base = 1024;
+ Random var = new Random();
+ base += var.nextInt(50);
+ return base;
+ }
+
+ // check if i can be divided by 53 plus a random
+ private boolean getFailure(int i)
+ {
+ Random var = new Random();
+ int num = 43 + var.nextInt(9);
+ boolean failure = ((i + 1) % num == 0);
+ if (var.nextBoolean())
+ {
+ try
+ {
+ Thread.sleep(var.nextInt(9) * 5);
+ }
+ catch (InterruptedException e)
+ {
+ }
+ }
+ return failure;
+ }
+
+}
Added: branches/Branch_JBossMessaging_1_4_0_SP3_CP/tests/src/org/jboss/test/messaging/jms/clustering/NullPersistenceClusterTest.java
===================================================================
--- branches/Branch_JBossMessaging_1_4_0_SP3_CP/tests/src/org/jboss/test/messaging/jms/clustering/NullPersistenceClusterTest.java (rev 0)
+++ branches/Branch_JBossMessaging_1_4_0_SP3_CP/tests/src/org/jboss/test/messaging/jms/clustering/NullPersistenceClusterTest.java 2008-09-18 02:08:06 UTC (rev 4976)
@@ -0,0 +1,106 @@
+/*
+ * JBoss, Home of Professional Open Source Copyright 2005, JBoss Inc., and
+ * individual contributors as indicated by the @authors tag. See the
+ * copyright.txt in the distribution for a full listing of individual
+ * contributors.
+ *
+ * This is free software; you can redistribute it and/or modify it under the
+ * terms of the GNU Lesser General Public License as published by the Free
+ * Software Foundation; either version 2.1 of the License, or (at your option)
+ * any later version.
+ *
+ * This software is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
+ * FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more
+ * details.
+ *
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with this software; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA, or see the FSF
+ * site: http://www.fsf.org.
+ */
+package org.jboss.test.messaging.jms.clustering;
+
+import java.util.HashSet;
+import java.util.Set;
+
+import javax.jms.Connection;
+import javax.jms.DeliveryMode;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+import javax.jms.TextMessage;
+
+public class NullPersistenceClusterTest extends ClusteringTestBase
+{
+
+ public NullPersistenceClusterTest(String name)
+ {
+ super(name);
+ }
+
+ protected void setUp() throws Exception
+ {
+ System.setProperty("test.database", "null");
+ System.setProperty("test.clustered", "true");
+ System.setProperty("remote", "true");
+ super.setUp();
+ }
+
+ protected void tearDown() throws Exception
+ {
+ super.tearDown();
+ }
+
+ public void testSimpleMessaging() throws Exception
+ {
+ Connection conn = null;
+
+ try
+ {
+ conn = cf.createConnection();
+
+ Session sess = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
+
+ MessageConsumer cons = sess.createConsumer(queue[0]);
+
+ conn.start();
+
+ MessageProducer prod = sess.createProducer(queue[0]);
+
+ final int NUM_MESSAGES = 100;
+
+ for (int i = 0; i < NUM_MESSAGES; i++)
+ {
+ TextMessage tm = sess.createTextMessage("message-" + i);
+
+ prod.send(tm);
+ }
+
+ for (int i = 0; i < NUM_MESSAGES; i++)
+ {
+ TextMessage tm = (TextMessage)cons.receive(1000);
+
+ assertNotNull(tm);
+
+ assertEquals("message-" + i, tm.getText());
+ }
+
+ Message m = cons.receive(2000);
+
+ assertNull(m);
+
+ cons.close();
+ sess.close();
+ }
+ finally
+ {
+ if (conn != null)
+ {
+ conn.close();
+ }
+ }
+ }
+
+}
More information about the jboss-cvs-commits
mailing list