[jboss-cvs] JBossAS SVN: r66418 - in branches/Branch_4_2/messaging/src/main/org/jboss/mq/server: jmx and 1 other directory.
jboss-cvs-commits at lists.jboss.org
jboss-cvs-commits at lists.jboss.org
Thu Oct 25 05:57:10 EDT 2007
Author: adrian at jboss.org
Date: 2007-10-25 05:57:10 -0400 (Thu, 25 Oct 2007)
New Revision: 66418
Modified:
branches/Branch_4_2/messaging/src/main/org/jboss/mq/server/BasicQueue.java
branches/Branch_4_2/messaging/src/main/org/jboss/mq/server/JMSDestinationManager.java
branches/Branch_4_2/messaging/src/main/org/jboss/mq/server/MessageReference.java
branches/Branch_4_2/messaging/src/main/org/jboss/mq/server/jmx/DestinationManager.java
branches/Branch_4_2/messaging/src/main/org/jboss/mq/server/jmx/DestinationManagerMBean.java
Log:
[JBAS-4870] - Update the redelivery flags in the database before delivering a persistent message such that they survive a crash
Modified: branches/Branch_4_2/messaging/src/main/org/jboss/mq/server/BasicQueue.java
===================================================================
--- branches/Branch_4_2/messaging/src/main/org/jboss/mq/server/BasicQueue.java 2007-10-25 05:10:59 UTC (rev 66417)
+++ branches/Branch_4_2/messaging/src/main/org/jboss/mq/server/BasicQueue.java 2007-10-25 09:57:10 UTC (rev 66418)
@@ -473,13 +473,17 @@
try
{
+ // Set redelivered, vendor-specific flags
message.redelivered();
- // Set redelivered, vendor-specific flags
- message.invalidate();
- // Update the persistent message outside the transaction
- // We want to know the message might have been delivered regardless
- if (message.isPersistent())
- server.getPersistenceManager().update(message, null);
+ // Old style update - doesn't survive a crash
+ if (server.isLazyRedeliveryUpdate())
+ {
+ message.invalidate();
+ // Update the persistent message outside the transaction
+ // We want to know the message might have been delivered regardless
+ if (message.isPersistent())
+ server.getPersistenceManager().update(message, null);
+ }
}
catch (JMSException e)
{
@@ -1160,6 +1164,19 @@
nack.messageID = message.getJMSMessageID();
nack.subscriberId = sub.subscriptionId;
+ // Need to update the message so it recovers from crash with redelivery=true
+ if (server.isLazyRedeliveryUpdate() == false && messageRef.isPersistent())
+ {
+ // Temporarily set the flags
+ messageRef.redelivered();
+ // Update the persistent message outside the transaction
+ // We want to know the message might have been delivered regardless
+ messageRef.invalidate();
+ server.getPersistenceManager().update(messageRef, null);
+ // Now revert the flags back again (but only in memory)
+ messageRef.revertRedelivered();
+ }
+
synchronized (messages)
{
UnackedMessageInfo unacked = new UnackedMessageInfo(messageRef, sub);
Modified: branches/Branch_4_2/messaging/src/main/org/jboss/mq/server/JMSDestinationManager.java
===================================================================
--- branches/Branch_4_2/messaging/src/main/org/jboss/mq/server/JMSDestinationManager.java 2007-10-25 05:10:59 UTC (rev 66417)
+++ branches/Branch_4_2/messaging/src/main/org/jboss/mq/server/JMSDestinationManager.java 2007-10-25 09:57:10 UTC (rev 66418)
@@ -102,7 +102,7 @@
private int lastTemporaryQueue = 1;
private Object lastTemporaryQueueLock = new Object();
-
+
/** The security manager */
private StateManager stateManager;
@@ -112,6 +112,9 @@
/** The Cache Used to hold messages */
private MessageCache messageCache;
+ /** Whether to do the redelivery update lazily */
+ private boolean lazyRedeliveryUpdate;
+
private Object stateLock = new Object();
private Object idLock = new Object();
@@ -168,6 +171,26 @@
}
/**
+ * Get the lazyRedeliveryUpdate.
+ *
+ * @return the lazyRedeliveryUpdate.
+ */
+ public boolean isLazyRedeliveryUpdate()
+ {
+ return lazyRedeliveryUpdate;
+ }
+
+ /**
+ * Set the lazyRedeliveryUpdate.
+ *
+ * @param lazyRedeliveryUpdate the lazyRedeliveryUpdate.
+ */
+ public void setLazyRedeliveryUpdate(boolean lazyRedeliveryUpdate)
+ {
+ this.lazyRedeliveryUpdate = lazyRedeliveryUpdate;
+ }
+
+ /**
* Returns <code>false</code> if the JMS server is currently running and
* handling requests, <code>true</code> otherwise.
*
Modified: branches/Branch_4_2/messaging/src/main/org/jboss/mq/server/MessageReference.java
===================================================================
--- branches/Branch_4_2/messaging/src/main/org/jboss/mq/server/MessageReference.java 2007-10-25 05:10:59 UTC (rev 66417)
+++ branches/Branch_4_2/messaging/src/main/org/jboss/mq/server/MessageReference.java 2007-10-25 09:57:10 UTC (rev 66418)
@@ -180,7 +180,7 @@
/**
* The message is being redelivered
*/
- public void redelivered() throws JMSException
+ public synchronized void redelivered() throws JMSException
{
this.redelivered = true;
@@ -202,6 +202,28 @@
}
/**
+ * Revert the redelivered status
+ */
+ public synchronized void revertRedelivered() throws JMSException
+ {
+ this.redelivered = true;
+
+ messageScheduledDelivery = 0;
+
+ --redeliveryCount;
+ if (redeliveryCount == 0)
+ redelivered = false;
+
+ if (isLateClone() == false)
+ {
+ SpyMessage message = getMessage();
+ message.setJMSRedelivered(redelivered);
+
+ message.header.jmsProperties.put(SpyMessage.PROPERTY_REDELIVERY_COUNT, new Integer(redeliveryCount));
+ }
+ }
+
+ /**
* Returns true if this message reference has expired.
*/
public boolean isExpired()
Modified: branches/Branch_4_2/messaging/src/main/org/jboss/mq/server/jmx/DestinationManager.java
===================================================================
--- branches/Branch_4_2/messaging/src/main/org/jboss/mq/server/jmx/DestinationManager.java 2007-10-25 05:10:59 UTC (rev 66417)
+++ branches/Branch_4_2/messaging/src/main/org/jboss/mq/server/jmx/DestinationManager.java 2007-10-25 09:57:10 UTC (rev 66418)
@@ -81,6 +81,8 @@
private ObjectName threadPool;
/** Default expiry destination */
private ObjectName expiryDestination;
+ /** Whether to do the redelivery update lazily */
+ private boolean lazyRedeliveryUpdate = false;
/**
* @jmx:managed-attribute
@@ -264,6 +266,26 @@
}
/**
+ * Get the lazyRedeliveryUpdate.
+ *
+ * @return the lazyRedeliveryUpdate.
+ */
+ public boolean isLazyRedeliveryUpdate()
+ {
+ return lazyRedeliveryUpdate;
+ }
+
+ /**
+ * Set the lazyRedeliveryUpdate.
+ *
+ * @param lazyRedeliveryUpdate the lazyRedeliveryUpdate.
+ */
+ public void setLazyRedeliveryUpdate(boolean lazyRedeliveryUpdate)
+ {
+ this.lazyRedeliveryUpdate = lazyRedeliveryUpdate;
+ }
+
+ /**
* @jmx:managed-operation
*/
public void createQueue(String name) throws Exception
@@ -384,6 +406,7 @@
{
super.createService();
jmsServer = new JMSDestinationManager(tempParameters);
+ jmsServer.setLazyRedeliveryUpdate(lazyRedeliveryUpdate);
}
protected void startService() throws Exception
@@ -394,7 +417,7 @@
ServiceControllerMBean.class,
ServiceControllerMBean.OBJECT_NAME,
server);
-
+
PersistenceManager pm = (PersistenceManager) server.getAttribute(persistenceManager, "Instance");
jmsServer.setPersistenceManager(pm);
Modified: branches/Branch_4_2/messaging/src/main/org/jboss/mq/server/jmx/DestinationManagerMBean.java
===================================================================
--- branches/Branch_4_2/messaging/src/main/org/jboss/mq/server/jmx/DestinationManagerMBean.java 2007-10-25 05:10:59 UTC (rev 66417)
+++ branches/Branch_4_2/messaging/src/main/org/jboss/mq/server/jmx/DestinationManagerMBean.java 2007-10-25 09:57:10 UTC (rev 66418)
@@ -131,6 +131,15 @@
*/
public void setExpiryDestination(ObjectName destination);
+ /**
+ * Get the lazyRedeliveryUpdate.
+ *
+ * @return the lazyRedeliveryUpdate.
+ */
+ boolean isLazyRedeliveryUpdate();
+
+ void setLazyRedeliveryUpdate(boolean lazyRedeliveryUpdate);
+
void createQueue(java.lang.String name) throws java.lang.Exception;
void createTopic(java.lang.String name) throws java.lang.Exception;
More information about the jboss-cvs-commits
mailing list