[jboss-cvs] JBoss Messaging SVN: r8153 - in branches/RIM_patch: src/main/org/jboss/jms/client/remoting and 5 other directories.
jboss-cvs-commits at lists.jboss.org
jboss-cvs-commits at lists.jboss.org
Mon Dec 20 06:19:07 EST 2010
Author: gaohoward
Date: 2010-12-20 06:19:06 -0500 (Mon, 20 Dec 2010)
New Revision: 8153
Modified:
branches/RIM_patch/integration/EAP4/etc/server/default/deploy/db2-persistence-service.xml
branches/RIM_patch/integration/EAP4/etc/server/default/deploy/mssql-persistence-service.xml
branches/RIM_patch/integration/EAP4/etc/server/default/deploy/mysql-persistence-service.xml
branches/RIM_patch/integration/EAP4/etc/server/default/deploy/ndb-persistence-service.xml
branches/RIM_patch/integration/EAP4/etc/server/default/deploy/oracle-persistence-service.xml
branches/RIM_patch/integration/EAP4/etc/server/default/deploy/postgresql-persistence-service.xml
branches/RIM_patch/integration/EAP4/etc/server/default/deploy/sybase-persistence-service.xml
branches/RIM_patch/src/main/org/jboss/jms/client/remoting/ClientSocketWrapper.java
branches/RIM_patch/src/main/org/jboss/jms/server/connectionmanager/SimpleConnectionManager.java
branches/RIM_patch/src/main/org/jboss/jms/server/endpoint/ServerConnectionEndpoint.java
branches/RIM_patch/src/main/org/jboss/jms/server/endpoint/ServerConsumerEndpoint.java
branches/RIM_patch/src/main/org/jboss/jms/server/endpoint/ServerSessionEndpoint.java
branches/RIM_patch/src/main/org/jboss/messaging/core/contract/Delivery.java
branches/RIM_patch/src/main/org/jboss/messaging/core/contract/PersistenceManager.java
branches/RIM_patch/src/main/org/jboss/messaging/core/impl/ChannelSupport.java
branches/RIM_patch/src/main/org/jboss/messaging/core/impl/ClusterRoundRobinDistributor.java
branches/RIM_patch/src/main/org/jboss/messaging/core/impl/JDBCPersistenceManager.java
branches/RIM_patch/src/main/org/jboss/messaging/core/impl/MessagingQueue.java
branches/RIM_patch/src/main/org/jboss/messaging/core/impl/NullPersistenceManager.java
branches/RIM_patch/src/main/org/jboss/messaging/core/impl/PagingChannelSupport.java
branches/RIM_patch/src/main/org/jboss/messaging/core/impl/SimpleDelivery.java
branches/RIM_patch/src/main/org/jboss/messaging/core/impl/clusterconnection/MessageSucker.java
Log:
added patch
Modified: branches/RIM_patch/integration/EAP4/etc/server/default/deploy/db2-persistence-service.xml
===================================================================
--- branches/RIM_patch/integration/EAP4/etc/server/default/deploy/db2-persistence-service.xml 2010-12-20 07:59:43 UTC (rev 8152)
+++ branches/RIM_patch/integration/EAP4/etc/server/default/deploy/db2-persistence-service.xml 2010-12-20 11:19:06 UTC (rev 8153)
@@ -66,7 +66,7 @@
SELECT_MIN_MAX_PAGE_ORD=SELECT MIN(PAGE_ORD), MAX(PAGE_ORD) FROM JBM_MSG_REF WHERE CHANNEL_ID = ?
UPDATE_DELIVERY_COUNT=UPDATE JBM_MSG_REF SET DELIVERY_COUNT = ? WHERE MESSAGE_ID = ? AND CHANNEL_ID = ?
UPDATE_CHANNEL_ID=UPDATE JBM_MSG_REF SET CHANNEL_ID = ? WHERE CHANNEL_ID = ?
- MOVE_REFERENCE=UPDATE JBM_MSG_REF SET CHANNEL_ID = ? WHERE MESSAGE_ID = ? AND CHANNEL_ID = ?
+ MOVE_REFERENCE=UPDATE JBM_MSG_REF SET CHANNEL_ID = ?, STATE='C' WHERE MESSAGE_ID = ? AND CHANNEL_ID = ? AND STATE='S'
LOAD_MESSAGES=SELECT MESSAGE_ID, RELIABLE, EXPIRATION, TIMESTAMP, PRIORITY, HEADERS, PAYLOAD, TYPE FROM JBM_MSG
INSERT_MESSAGE=INSERT INTO JBM_MSG (MESSAGE_ID, RELIABLE, EXPIRATION, TIMESTAMP, PRIORITY, TYPE, HEADERS, PAYLOAD) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
INSERT_MESSAGE_CONDITIONAL=INSERT INTO JBM_MSG (MESSAGE_ID, RELIABLE, EXPIRATION, TIMESTAMP, PRIORITY, TYPE) SELECT (CAST(? AS DECIMAL(19,0))), (CAST(? AS CHAR)), (CAST(? AS DECIMAL(19,0))), (CAST(? AS DECIMAL(19,0))), (CAST(? AS DECIMAL(19,0))), (CAST(? AS DECIMAL(19,0))) FROM JBM_DUAL WHERE NOT EXISTS (SELECT MESSAGE_ID FROM JBM_MSG WHERE MESSAGE_ID = ?)
@@ -93,6 +93,9 @@
SELECT_TRANSACTION_START_TIME_EXTRA=SELECT TRANSACTION_ID, START_TIME FROM JBM_TX_EX
INSERT_TRANSACTION_EXTRA=INSERT INTO JBM_TX_EX (TRANSACTION_ID, START_TIME) VALUES(?, ?)
DELETE_TRANSACTION_EXTRA=DELETE FROM JBM_TX_EX WHERE TRANSACTION_ID=?
+ UPDATE_MESSAGE_STATE=UPDATE JBM_MSG_REF SET STATE = ? WHERE MESSAGE_ID = ? AND CHANNEL_ID = ?
+ CLAIM_MESSAGE_IN_SUCK=UPDATE JBM_MSG_REF SET STATE='C' WHERE CHANNEL_ID = ? AND STATE='S'
+ LOAD_REFS_IN_SUCK=SELECT MESSAGE_ID, DELIVERY_COUNT, SCHED_DELIVERY FROM JBM_MSG_REF WHERE STATE = 'S' AND CHANNEL_ID = ? ORDER BY ORD
]]></attribute>
<!-- The maximum number of parameters to include in a prepared statement -->
Modified: branches/RIM_patch/integration/EAP4/etc/server/default/deploy/mssql-persistence-service.xml
===================================================================
--- branches/RIM_patch/integration/EAP4/etc/server/default/deploy/mssql-persistence-service.xml 2010-12-20 07:59:43 UTC (rev 8152)
+++ branches/RIM_patch/integration/EAP4/etc/server/default/deploy/mssql-persistence-service.xml 2010-12-20 11:19:06 UTC (rev 8153)
@@ -62,7 +62,7 @@
SELECT_MIN_MAX_PAGE_ORD=SELECT MIN(PAGE_ORD), MAX(PAGE_ORD) FROM JBM_MSG_REF WHERE CHANNEL_ID = ?
UPDATE_DELIVERY_COUNT=UPDATE JBM_MSG_REF SET DELIVERY_COUNT = ? WHERE MESSAGE_ID = ? AND CHANNEL_ID = ?
UPDATE_CHANNEL_ID=UPDATE JBM_MSG_REF SET CHANNEL_ID = ? WHERE CHANNEL_ID = ?
- MOVE_REFERENCE=UPDATE JBM_MSG_REF SET CHANNEL_ID = ? WHERE MESSAGE_ID = ? AND CHANNEL_ID = ?
+ MOVE_REFERENCE=UPDATE JBM_MSG_REF SET CHANNEL_ID = ?, STATE='C' WHERE MESSAGE_ID = ? AND CHANNEL_ID = ? AND STATE='S'
LOAD_MESSAGES=SELECT MESSAGE_ID, RELIABLE, EXPIRATION, TIMESTAMP, PRIORITY, HEADERS, PAYLOAD, TYPE FROM JBM_MSG
INSERT_MESSAGE=INSERT INTO JBM_MSG (MESSAGE_ID, RELIABLE, EXPIRATION, TIMESTAMP, PRIORITY, TYPE, HEADERS, PAYLOAD) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
INSERT_MESSAGE_CONDITIONAL=INSERT INTO JBM_MSG (MESSAGE_ID, RELIABLE, EXPIRATION, TIMESTAMP, PRIORITY, TYPE) SELECT ?, ?, ?, ?, ?, ? FROM JBM_DUAL WHERE NOT EXISTS (SELECT MESSAGE_ID FROM JBM_MSG WHERE MESSAGE_ID = ?)
@@ -89,6 +89,9 @@
SELECT_TRANSACTION_START_TIME_EXTRA=SELECT TRANSACTION_ID, START_TIME FROM JBM_TX_EX
INSERT_TRANSACTION_EXTRA=INSERT INTO JBM_TX_EX (TRANSACTION_ID, START_TIME) VALUES(?, ?)
DELETE_TRANSACTION_EXTRA=DELETE FROM JBM_TX_EX WHERE TRANSACTION_ID=?
+ UPDATE_MESSAGE_STATE=UPDATE JBM_MSG_REF SET STATE = ? WHERE MESSAGE_ID = ? AND CHANNEL_ID = ?
+ CLAIM_MESSAGE_IN_SUCK=UPDATE JBM_MSG_REF SET STATE='C' WHERE CHANNEL_ID = ? AND STATE='S'
+ LOAD_REFS_IN_SUCK=SELECT MESSAGE_ID, DELIVERY_COUNT, SCHED_DELIVERY FROM JBM_MSG_REF WHERE STATE = 'S' AND CHANNEL_ID = ? ORDER BY ORD
]]></attribute>
<!-- The maximum number of parameters to include in a prepared statement -->
Modified: branches/RIM_patch/integration/EAP4/etc/server/default/deploy/mysql-persistence-service.xml
===================================================================
--- branches/RIM_patch/integration/EAP4/etc/server/default/deploy/mysql-persistence-service.xml 2010-12-20 07:59:43 UTC (rev 8152)
+++ branches/RIM_patch/integration/EAP4/etc/server/default/deploy/mysql-persistence-service.xml 2010-12-20 11:19:06 UTC (rev 8153)
@@ -62,7 +62,7 @@
SELECT_MIN_MAX_PAGE_ORD=SELECT MIN(PAGE_ORD), MAX(PAGE_ORD) FROM JBM_MSG_REF WHERE CHANNEL_ID = ?
UPDATE_DELIVERY_COUNT=UPDATE JBM_MSG_REF SET DELIVERY_COUNT = ? WHERE MESSAGE_ID = ? AND CHANNEL_ID = ?
UPDATE_CHANNEL_ID=UPDATE JBM_MSG_REF SET CHANNEL_ID = ? WHERE CHANNEL_ID = ?
- MOVE_REFERENCE=UPDATE JBM_MSG_REF SET CHANNEL_ID = ? WHERE MESSAGE_ID = ? AND CHANNEL_ID = ?
+ MOVE_REFERENCE=UPDATE JBM_MSG_REF SET CHANNEL_ID = ?, STATE='C' WHERE MESSAGE_ID = ? AND CHANNEL_ID = ? AND STATE='S'
LOAD_MESSAGES=SELECT MESSAGE_ID, RELIABLE, EXPIRATION, TIMESTAMP, PRIORITY, HEADERS, PAYLOAD, TYPE FROM JBM_MSG
INSERT_MESSAGE=INSERT INTO JBM_MSG (MESSAGE_ID, RELIABLE, EXPIRATION, TIMESTAMP, PRIORITY, TYPE, HEADERS, PAYLOAD) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
INSERT_MESSAGE_CONDITIONAL=INSERT INTO JBM_MSG (MESSAGE_ID, RELIABLE, EXPIRATION, TIMESTAMP, PRIORITY, TYPE) SELECT ?, ?, ?, ?, ?, ? FROM JBM_DUAL WHERE NOT EXISTS (SELECT MESSAGE_ID FROM JBM_MSG WHERE MESSAGE_ID = ?)
@@ -89,6 +89,9 @@
SELECT_TRANSACTION_START_TIME_EXTRA=SELECT TRANSACTION_ID, START_TIME FROM JBM_TX_EX
INSERT_TRANSACTION_EXTRA=INSERT INTO JBM_TX_EX (TRANSACTION_ID, START_TIME) VALUES(?,?)
DELETE_TRANSACTION_EXTRA=DELETE FROM JBM_TX_EX WHERE TRANSACTION_ID=?
+ UPDATE_MESSAGE_STATE=UPDATE JBM_MSG_REF SET STATE = ? WHERE MESSAGE_ID = ? AND CHANNEL_ID = ?
+ CLAIM_MESSAGE_IN_SUCK=UPDATE JBM_MSG_REF SET STATE='C' WHERE CHANNEL_ID = ? AND STATE='S'
+ LOAD_REFS_IN_SUCK=SELECT MESSAGE_ID, DELIVERY_COUNT, SCHED_DELIVERY FROM JBM_MSG_REF WHERE STATE = 'S' AND CHANNEL_ID = ? ORDER BY ORD
]]></attribute>
<!-- The maximum number of parameters to include in a prepared statement -->
Modified: branches/RIM_patch/integration/EAP4/etc/server/default/deploy/ndb-persistence-service.xml
===================================================================
--- branches/RIM_patch/integration/EAP4/etc/server/default/deploy/ndb-persistence-service.xml 2010-12-20 07:59:43 UTC (rev 8152)
+++ branches/RIM_patch/integration/EAP4/etc/server/default/deploy/ndb-persistence-service.xml 2010-12-20 11:19:06 UTC (rev 8153)
@@ -62,7 +62,7 @@
SELECT_MIN_MAX_PAGE_ORD=SELECT MIN(PAGE_ORD), MAX(PAGE_ORD) FROM JBM_MSG_REF WHERE CHANNEL_ID = ?
UPDATE_DELIVERY_COUNT=UPDATE JBM_MSG_REF SET DELIVERY_COUNT = ? WHERE MESSAGE_ID = ? AND CHANNEL_ID = ?
UPDATE_CHANNEL_ID=UPDATE JBM_MSG_REF SET CHANNEL_ID = ? WHERE CHANNEL_ID = ?
- MOVE_REFERENCE=UPDATE JBM_MSG_REF SET CHANNEL_ID = ? WHERE MESSAGE_ID = ? AND CHANNEL_ID = ?
+ MOVE_REFERENCE=UPDATE JBM_MSG_REF SET CHANNEL_ID = ?, STATE='C' WHERE MESSAGE_ID = ? AND CHANNEL_ID = ? AND STATE='S'
LOAD_MESSAGES=SELECT MESSAGE_ID, RELIABLE, EXPIRATION, TIMESTAMP, PRIORITY, HEADERS, PAYLOAD, TYPE FROM JBM_MSG
INSERT_MESSAGE=INSERT INTO JBM_MSG (MESSAGE_ID, RELIABLE, EXPIRATION, TIMESTAMP, PRIORITY, TYPE, HEADERS, PAYLOAD) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
INSERT_MESSAGE_CONDITIONAL=INSERT INTO JBM_MSG (MESSAGE_ID, RELIABLE, EXPIRATION, TIMESTAMP, PRIORITY, TYPE) SELECT ?, ?, ?, ?, ?, ? FROM JBM_DUAL WHERE NOT EXISTS (SELECT MESSAGE_ID FROM JBM_MSG WHERE MESSAGE_ID = ?)
@@ -89,6 +89,9 @@
SELECT_TRANSACTION_START_TIME_EXTRA=SELECT TRANSACTION_ID, START_TIME FROM JBM_TX_EX
INSERT_TRANSACTION_EXTRA=INSERT INTO JBM_TX_EX (TRANSACTION_ID, START_TIME) VALUES(?, ?)
DELETE_TRANSACTION_EXTRA=DELETE FROM JBM_TX_EX WHERE TRANSACTION_ID=?
+ UPDATE_MESSAGE_STATE=UPDATE JBM_MSG_REF SET STATE = ? WHERE MESSAGE_ID = ? AND CHANNEL_ID = ?
+ CLAIM_MESSAGE_IN_SUCK=UPDATE JBM_MSG_REF SET STATE='C' WHERE CHANNEL_ID = ? AND STATE='S'
+ LOAD_REFS_IN_SUCK=SELECT MESSAGE_ID, DELIVERY_COUNT, SCHED_DELIVERY FROM JBM_MSG_REF WHERE STATE = 'S' AND CHANNEL_ID = ? ORDER BY ORD
]]></attribute>
<!-- The maximum number of parameters to include in a prepared statement -->
Modified: branches/RIM_patch/integration/EAP4/etc/server/default/deploy/oracle-persistence-service.xml
===================================================================
--- branches/RIM_patch/integration/EAP4/etc/server/default/deploy/oracle-persistence-service.xml 2010-12-20 07:59:43 UTC (rev 8152)
+++ branches/RIM_patch/integration/EAP4/etc/server/default/deploy/oracle-persistence-service.xml 2010-12-20 11:19:06 UTC (rev 8153)
@@ -66,7 +66,7 @@
SELECT_MIN_MAX_PAGE_ORD=SELECT MIN(PAGE_ORD), MAX(PAGE_ORD) FROM JBM_MSG_REF WHERE CHANNEL_ID = ?
UPDATE_DELIVERY_COUNT=UPDATE JBM_MSG_REF SET DELIVERY_COUNT = ? WHERE MESSAGE_ID = ? AND CHANNEL_ID = ?
UPDATE_CHANNEL_ID=UPDATE JBM_MSG_REF SET CHANNEL_ID = ? WHERE CHANNEL_ID = ?
- MOVE_REFERENCE=UPDATE JBM_MSG_REF SET CHANNEL_ID = ? WHERE MESSAGE_ID = ? AND CHANNEL_ID = ?
+ MOVE_REFERENCE=UPDATE JBM_MSG_REF SET CHANNEL_ID = ?, STATE='C' WHERE MESSAGE_ID = ? AND CHANNEL_ID = ? AND STATE='S'
LOAD_MESSAGES=SELECT MESSAGE_ID, RELIABLE, EXPIRATION, TIMESTAMP, PRIORITY, HEADERS, PAYLOAD, TYPE FROM JBM_MSG
INSERT_MESSAGE=INSERT INTO JBM_MSG (MESSAGE_ID, RELIABLE, EXPIRATION, TIMESTAMP, PRIORITY, TYPE, HEADERS, PAYLOAD) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
INSERT_MESSAGE_CONDITIONAL=INSERT INTO JBM_MSG (MESSAGE_ID, RELIABLE, EXPIRATION, TIMESTAMP, PRIORITY, TYPE) SELECT ?, ?, ?, ?, ?, ? FROM JBM_DUAL WHERE NOT EXISTS (SELECT MESSAGE_ID FROM JBM_MSG WHERE MESSAGE_ID = ?)
@@ -93,6 +93,9 @@
SELECT_TRANSACTION_START_TIME_EXTRA=SELECT TRANSACTION_ID, START_TIME FROM JBM_TX_EX
INSERT_TRANSACTION_EXTRA=INSERT INTO JBM_TX_EX (TRANSACTION_ID, START_TIME) VALUES(?, ?)
DELETE_TRANSACTION_EXTRA=DELETE FROM JBM_TX_EX WHERE TRANSACTION_ID=?
+ UPDATE_MESSAGE_STATE=UPDATE JBM_MSG_REF SET STATE = ? WHERE MESSAGE_ID = ? AND CHANNEL_ID = ?
+ CLAIM_MESSAGE_IN_SUCK=UPDATE JBM_MSG_REF SET STATE='C' WHERE CHANNEL_ID = ? AND STATE='S'
+ LOAD_REFS_IN_SUCK=SELECT MESSAGE_ID, DELIVERY_COUNT, SCHED_DELIVERY FROM JBM_MSG_REF WHERE STATE = 'S' AND CHANNEL_ID = ? ORDER BY ORD
]]></attribute>
<!-- The maximum number of parameters to include in a prepared statement -->
Modified: branches/RIM_patch/integration/EAP4/etc/server/default/deploy/postgresql-persistence-service.xml
===================================================================
--- branches/RIM_patch/integration/EAP4/etc/server/default/deploy/postgresql-persistence-service.xml 2010-12-20 07:59:43 UTC (rev 8152)
+++ branches/RIM_patch/integration/EAP4/etc/server/default/deploy/postgresql-persistence-service.xml 2010-12-20 11:19:06 UTC (rev 8153)
@@ -62,7 +62,7 @@
SELECT_MIN_MAX_PAGE_ORD=SELECT MIN(PAGE_ORD), MAX(PAGE_ORD) FROM JBM_MSG_REF WHERE CHANNEL_ID = ?
UPDATE_DELIVERY_COUNT=UPDATE JBM_MSG_REF SET DELIVERY_COUNT = ? WHERE MESSAGE_ID = ? AND CHANNEL_ID = ?
UPDATE_CHANNEL_ID=UPDATE JBM_MSG_REF SET CHANNEL_ID = ? WHERE CHANNEL_ID = ?
- MOVE_REFERENCE=UPDATE JBM_MSG_REF SET CHANNEL_ID = ? WHERE MESSAGE_ID = ? AND CHANNEL_ID = ?
+ MOVE_REFERENCE=UPDATE JBM_MSG_REF SET CHANNEL_ID = ?, STATE='C' WHERE MESSAGE_ID = ? AND CHANNEL_ID = ? AND STATE='S'
LOAD_MESSAGES=SELECT MESSAGE_ID, RELIABLE, EXPIRATION, TIMESTAMP, PRIORITY, HEADERS, PAYLOAD, TYPE FROM JBM_MSG
INSERT_MESSAGE=INSERT INTO JBM_MSG (MESSAGE_ID, RELIABLE, EXPIRATION, TIMESTAMP, PRIORITY, TYPE, HEADERS, PAYLOAD) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
INSERT_MESSAGE_CONDITIONAL=INSERT INTO JBM_MSG (MESSAGE_ID, RELIABLE, EXPIRATION, TIMESTAMP, PRIORITY, TYPE) SELECT ?, ?, ?, ?, ?, ? FROM JBM_DUAL WHERE NOT EXISTS (SELECT MESSAGE_ID FROM JBM_MSG WHERE MESSAGE_ID = ?)
@@ -89,6 +89,9 @@
SELECT_TRANSACTION_START_TIME_EXTRA=SELECT TRANSACTION_ID, START_TIME FROM JBM_TX_EX
INSERT_TRANSACTION_EXTRA=INSERT INTO JBM_TX_EX (TRANSACTION_ID, START_TIME) VALUES(?, ?)
DELETE_TRANSACTION_EXTRA=DELETE FROM JBM_TX_EX WHERE TRANSACTION_ID=?
+ UPDATE_MESSAGE_STATE=UPDATE JBM_MSG_REF SET STATE = ? WHERE MESSAGE_ID = ? AND CHANNEL_ID = ?
+ CLAIM_MESSAGE_IN_SUCK=UPDATE JBM_MSG_REF SET STATE='C' WHERE CHANNEL_ID = ? AND STATE='S'
+ LOAD_REFS_IN_SUCK=SELECT MESSAGE_ID, DELIVERY_COUNT, SCHED_DELIVERY FROM JBM_MSG_REF WHERE STATE = 'S' AND CHANNEL_ID = ? ORDER BY ORD
]]></attribute>
<!-- The maximum number of parameters to include in a prepared statement -->
Modified: branches/RIM_patch/integration/EAP4/etc/server/default/deploy/sybase-persistence-service.xml
===================================================================
--- branches/RIM_patch/integration/EAP4/etc/server/default/deploy/sybase-persistence-service.xml 2010-12-20 07:59:43 UTC (rev 8152)
+++ branches/RIM_patch/integration/EAP4/etc/server/default/deploy/sybase-persistence-service.xml 2010-12-20 11:19:06 UTC (rev 8153)
@@ -67,7 +67,7 @@
SELECT_MIN_MAX_PAGE_ORD=SELECT MIN(PAGE_ORD), MAX(PAGE_ORD) FROM JBM_MSG_REF WHERE CHANNEL_ID = ?
UPDATE_DELIVERY_COUNT=UPDATE JBM_MSG_REF SET DELIVERY_COUNT = ? WHERE MESSAGE_ID = ? AND CHANNEL_ID = ?
UPDATE_CHANNEL_ID=UPDATE JBM_MSG_REF SET CHANNEL_ID = ? WHERE CHANNEL_ID = ?
- MOVE_REFERENCE=UPDATE JBM_MSG_REF SET CHANNEL_ID = ? WHERE MESSAGE_ID = ? AND CHANNEL_ID = ?
+ MOVE_REFERENCE=UPDATE JBM_MSG_REF SET CHANNEL_ID = ?, STATE='C' WHERE MESSAGE_ID = ? AND CHANNEL_ID = ? AND STATE='S'
LOAD_MESSAGES=SELECT MESSAGE_ID, RELIABLE, EXPIRATION, TIMESTAMP, PRIORITY, HEADERS, PAYLOAD, TYPE FROM JBM_MSG
INSERT_MESSAGE=INSERT INTO JBM_MSG (MESSAGE_ID, RELIABLE, EXPIRATION, TIMESTAMP, PRIORITY, TYPE, HEADERS, PAYLOAD) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
INSERT_MESSAGE_CONDITIONAL=INSERT INTO JBM_MSG (MESSAGE_ID, RELIABLE, EXPIRATION, TIMESTAMP, PRIORITY, TYPE) SELECT ?, ?, ?, ?, ?, ? FROM JBM_DUAL WHERE NOT EXISTS (SELECT MESSAGE_ID FROM JBM_MSG WHERE MESSAGE_ID = ?)
@@ -94,6 +94,9 @@
SELECT_TRANSACTION_START_TIME_EXTRA=SELECT TRANSACTION_ID, START_TIME FROM JBM_TX_EX
INSERT_TRANSACTION_EXTRA=INSERT INTO JBM_TX_EX (TRANSACTION_ID, START_TIME) VALUES(?, ?)
DELETE_TRANSACTION_EXTRA=DELETE FROM JBM_TX_EX WHERE TRANSACTION_ID=?
+ UPDATE_MESSAGE_STATE=UPDATE JBM_MSG_REF SET STATE = ? WHERE MESSAGE_ID = ? AND CHANNEL_ID = ?
+ CLAIM_MESSAGE_IN_SUCK=UPDATE JBM_MSG_REF SET STATE='C' WHERE CHANNEL_ID = ? AND STATE='S'
+ LOAD_REFS_IN_SUCK=SELECT MESSAGE_ID, DELIVERY_COUNT, SCHED_DELIVERY FROM JBM_MSG_REF WHERE STATE = 'S' AND CHANNEL_ID = ? ORDER BY ORD
]]></attribute>
<!-- The maximum number of parameters to include in a prepared statement -->
Modified: branches/RIM_patch/src/main/org/jboss/jms/client/remoting/ClientSocketWrapper.java
===================================================================
--- branches/RIM_patch/src/main/org/jboss/jms/client/remoting/ClientSocketWrapper.java 2010-12-20 07:59:43 UTC (rev 8152)
+++ branches/RIM_patch/src/main/org/jboss/jms/client/remoting/ClientSocketWrapper.java 2010-12-20 11:19:06 UTC (rev 8153)
@@ -48,7 +48,7 @@
{
// Constants ------------------------------------------------------------------------------------
final static private Logger log = Logger.getLogger(ClientSocketWrapper.class);
- final static protected int CLOSING = 1;
+ final static protected int CLOSING = 254;
// Static ---------------------------------------------------------------------------------------
@@ -90,6 +90,10 @@
{
int b = ((DataInputStream)getInputStream()).readByte();
log.debug(this + ".checkConnection read " + b);
+ if (b != ACK)
+ {
+ throw new IOException("got " + b + " instead of " + ACK);
+ }
}
catch (IOException e)
{
@@ -104,7 +108,7 @@
{
if (log.isTraceEnabled()) log.trace("checking open connection");
- if (((DataInputStream)getInputStream()).available() > 1)
+ if (((DataInputStream)getInputStream()).available() > 0)
{
log.trace("remote endpoint has closed");
throw new IOException("remote endpoint has closed");
Modified: branches/RIM_patch/src/main/org/jboss/jms/server/connectionmanager/SimpleConnectionManager.java
===================================================================
--- branches/RIM_patch/src/main/org/jboss/jms/server/connectionmanager/SimpleConnectionManager.java 2010-12-20 07:59:43 UTC (rev 8152)
+++ branches/RIM_patch/src/main/org/jboss/jms/server/connectionmanager/SimpleConnectionManager.java 2010-12-20 11:19:06 UTC (rev 8153)
@@ -485,7 +485,14 @@
}
try
{
- conn.close();
+ if (conn instanceof ServerConnectionEndpoint)
+ {
+ ((ServerConnectionEndpoint)conn).close(true);
+ }
+ else
+ {
+ conn.close();
+ }
}
catch (Throwable ignore)
{
Modified: branches/RIM_patch/src/main/org/jboss/jms/server/endpoint/ServerConnectionEndpoint.java
===================================================================
--- branches/RIM_patch/src/main/org/jboss/jms/server/endpoint/ServerConnectionEndpoint.java 2010-12-20 07:59:43 UTC (rev 8152)
+++ branches/RIM_patch/src/main/org/jboss/jms/server/endpoint/ServerConnectionEndpoint.java 2010-12-20 11:19:06 UTC (rev 8153)
@@ -364,6 +364,11 @@
public void close() throws JMSException
{
+ close(false);
+ }
+
+ public void close(boolean isFromFailure) throws JMSException
+ {
try
{
//reason for synchronization
@@ -392,7 +397,7 @@
{
ServerSessionEndpoint sess = (ServerSessionEndpoint)i.next();
- sess.localClose();
+ sess.localClose(isFromFailure);
}
sessions.clear();
Modified: branches/RIM_patch/src/main/org/jboss/jms/server/endpoint/ServerConsumerEndpoint.java
===================================================================
--- branches/RIM_patch/src/main/org/jboss/jms/server/endpoint/ServerConsumerEndpoint.java 2010-12-20 07:59:43 UTC (rev 8152)
+++ branches/RIM_patch/src/main/org/jboss/jms/server/endpoint/ServerConsumerEndpoint.java 2010-12-20 11:19:06 UTC (rev 8153)
@@ -35,10 +35,12 @@
import org.jboss.jms.server.selector.Selector;
import org.jboss.jms.wireformat.Dispatcher;
import org.jboss.logging.Logger;
+import org.jboss.messaging.core.contract.Channel;
import org.jboss.messaging.core.contract.Delivery;
import org.jboss.messaging.core.contract.DeliveryObserver;
import org.jboss.messaging.core.contract.Message;
import org.jboss.messaging.core.contract.MessageReference;
+import org.jboss.messaging.core.contract.PersistenceManager;
import org.jboss.messaging.core.contract.PostOffice;
import org.jboss.messaging.core.contract.Queue;
import org.jboss.messaging.core.contract.Receiver;
@@ -320,8 +322,28 @@
clientAccepting = false;
firstTime = false;
- }
-
+ }
+
+ // now for a remote sucker, we need to update the messages status
+ if (remote)
+ {
+ PersistenceManager pm = sessionEndpoint.getPersistenceManager();
+ if (ref.getMessage().isReliable() && messageQueue.isRecoverable())
+ {
+ try
+ {
+ pm.updateMessageState(messageQueue.getChannelID(), ref, "S");
+ }
+ catch (Exception e)
+ {
+ // we need to stop the sucking process. the message should be re-delivered.
+ log.error("Failed to update state for message: " + ref, e);
+ return null;
+ }
+ }
+ delivery.setSucked(true);
+ }
+
try
{
sessionEndpoint.handleDelivery(delivery, this);
@@ -647,6 +669,16 @@
sessionEndpoint.promptDelivery(messageQueue);
}
+ public long getChannelID()
+ {
+ return messageQueue.getChannelID();
+ }
+
+ public Channel getChannel()
+ {
+ return messageQueue;
+ }
+
// Inner classes --------------------------------------------------------------------------------
}
Modified: branches/RIM_patch/src/main/org/jboss/jms/server/endpoint/ServerSessionEndpoint.java
===================================================================
--- branches/RIM_patch/src/main/org/jboss/jms/server/endpoint/ServerSessionEndpoint.java 2010-12-20 07:59:43 UTC (rev 8152)
+++ branches/RIM_patch/src/main/org/jboss/jms/server/endpoint/ServerSessionEndpoint.java 2010-12-20 11:19:06 UTC (rev 8153)
@@ -33,6 +33,7 @@
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicBoolean;
import javax.jms.IllegalStateException;
import javax.jms.InvalidDestinationException;
@@ -74,11 +75,13 @@
import org.jboss.messaging.core.contract.MessageReference;
import org.jboss.messaging.core.contract.MessageStore;
import org.jboss.messaging.core.contract.PersistenceManager;
+import org.jboss.messaging.core.contract.PersistenceManager.ReferenceInfo;
import org.jboss.messaging.core.contract.PostOffice;
import org.jboss.messaging.core.contract.Queue;
import org.jboss.messaging.core.contract.Replicator;
import org.jboss.messaging.core.impl.IDManager;
import org.jboss.messaging.core.impl.MessagingQueue;
+import org.jboss.messaging.core.impl.SimpleDelivery;
import org.jboss.messaging.core.impl.tx.Transaction;
import org.jboss.messaging.core.impl.tx.TransactionException;
import org.jboss.messaging.core.impl.tx.TransactionRepository;
@@ -186,8 +189,11 @@
private Object waitLock = new Object();
private long lastSequence = -1;
+
+ private Map<Long, Long> failureCanceledDels;
+
+ private AtomicBoolean isSuckerSession = new AtomicBoolean(false);
-
// Constructors ---------------------------------------------------------------------------------
ServerSessionEndpoint(String sessionID, ServerConnectionEndpoint connectionEndpoint,
@@ -236,6 +242,8 @@
defaultRedeliveryDelay = sp.getDefaultRedeliveryDelay();
deliveries = new ConcurrentHashMap();
+
+ failureCanceledDels = new HashMap<Long, Long>();
}
// SessionDelegate implementation ---------------------------------------------------------------
@@ -1136,7 +1144,12 @@
void localClose() throws Throwable
{
-
+ localClose(false);
+ }
+
+ void localClose(boolean isFromFailure) throws Throwable
+ {
+
if (closed)
{
//don't throw the exception as it maybe called twice
@@ -1153,10 +1166,14 @@
{
consumersClone = new HashMap(consumers);
}
+
+ List<Channel> curChannels = new ArrayList<Channel>();
for( Iterator i = consumersClone.values().iterator(); i.hasNext(); )
{
- ((ServerConsumerEndpoint)i.next()).localClose();
+ ServerConsumerEndpoint consumer = (ServerConsumerEndpoint)i.next();
+ curChannels.add(consumer.getChannel());
+ consumer.localClose();
}
consumers.clear();
@@ -1213,6 +1230,27 @@
DeliveryRecord rec = (DeliveryRecord)entry.getValue();
+ // for a suck delivery, we need to update the state back to 'C'
+ if (rec.del.isSucked())
+ {
+ // we need to reverse the message (if still there). If reverse failed, we don't do
+ // cancel.
+ if (rec.del.getReference().getMessage().isReliable() && rec.getConsumer().getChannel().isRecoverable())
+ {
+ try
+ {
+ // now ask pm to do it.
+ pm.updateMessageState(rec.getConsumer().getChannelID(), rec.del.getReference(), "C");
+ }
+ catch (Exception e)
+ {
+ // if update failed, it must be a DB failure, we log the error and let others be canceled
+ log.error("Failed to update message " + rec.del.getReference() + " to state C", e);
+ continue;
+ }
+ }
+ }
+
/*
* https://jira.jboss.org/jira/browse/JBMESSAGING-1440
*/
@@ -1222,8 +1260,54 @@
}
channels.add(rec.del.getObserver());
+
+ if (isFromFailure)
+ {
+ failureCanceledDels.put(rec.deliveryID, rec.deliveryID);
+ }
}
+ if (isSuckerSession.get())
+ {
+ // here we handle rare cases where a sucker acked a message but then crashed.
+ // so the message won't be updated to target channel and also the session already
+ // forgets it. We take this chance here
+ // to load those messages into channels and redeliver
+ for (Channel ch : curChannels)
+ {
+ List<ReferenceInfo> refs = pm.claimMessagesInSuck(ch.getChannelID());
+
+ if (refs.size() > 0)
+ {
+ List<Long> mids = new ArrayList<Long>();
+ Map<Long, ReferenceInfo> refInfoMap = new HashMap<Long, ReferenceInfo>();
+
+ for (ReferenceInfo refInfo : refs)
+ {
+ mids.add(refInfo.getMessageId());
+ refInfoMap.put(refInfo.getMessageId(), refInfo);
+ }
+
+ List messages = pm.getMessages(mids);
+
+ iter = messages.iterator();
+
+ while (iter.hasNext())
+ {
+ Message m = (Message)iter.next();
+ MessageReference mref = ms.reference(m);
+
+ ReferenceInfo mInfo = refInfoMap.get(m.getMessageID());
+ mref.setDeliveryCount(mInfo.getDeliveryCount());
+ mref.setScheduledDeliveryTime(mInfo.getScheduledDelivery());
+
+ Delivery del = new SimpleDelivery(ch, mref, true, true);
+ del.cancel();
+ }
+ }
+ }
+ }
+
promptDelivery(channels);
//Close down the executor
@@ -1600,19 +1684,40 @@
synchronized (deliveries)
{
rec = (DeliveryRecord)deliveries.remove(new Long(cancel.getDeliveryId()));
+ if (rec == null)
+ {
+ // The delivery might not be found, if the session is not replicated (i.e. auto_ack or dups_ok)
+ // and has failed over since recoverDeliveries won't have been called
+ if (trace)
+ {
+ log.trace("Cannot find delivery to cancel, session probably failed over and is not replicated");
+ }
+ return null;
+ }
+ // now we check for suckers
+ if (rec.del.isSucked())
+ {
+ // we need to reverse the message (if still there). If reverse failed, we don't do
+ // cancel.
+ if (rec.del.getReference().getMessage().isReliable() && rec.getConsumer().getChannel().isRecoverable())
+ {
+ try
+ {
+ // now ask pm to do it.
+ pm.updateMessageState(rec.getConsumer().getChannelID(), rec.del.getReference(), "C");
+ }
+ catch (Exception e)
+ {
+ if (trace)
+ {
+ log.trace("Failed to update message " + rec.del.getReference() + " to state C");
+ }
+ return null;
+ }
+ }
+ }
}
- if (rec == null)
- {
- //The delivery might not be found, if the session is not replicated (i.e. auto_ack or dups_ok)
- //and has failed over since recoverDeliveries won't have been called
- if (trace)
- {
- log.trace("Cannot find delivery to cancel, session probably failed over and is not replicated");
- }
- return null;
- }
-
//Note we check the flag *and* evaluate again, this is because the server and client clocks may
//be out of synch and don't want to send back to the client a message it thought it has sent to
//the expiry queue
@@ -1762,15 +1867,31 @@
synchronized (deliveries)
{
rec = (DeliveryRecord)deliveries.remove(new Long(ack.getDeliveryID()));
+ if (rec == null)
+ {
+ // This can happen in one of the two cases:
+ //
+ // 1. If an ack comes in after failover, or
+ // 2. The session is closed due to server side connection failure notification processing.
+ // When a connection failure is detected at the server end, it will close all related server side
+ // sessions. As part of closing, any un-acked message will be canceled.
+ // if a normal client side ack comes in just after the session is thus being closed, and this ack
+ // has just been canceled, the client side ack will end up here.
+ //
+ // We treat the cases differently. For case 1, we can safely ignore it.
+ // For case 2, we must throw an exception to indicating that the ack failed and the message will be
+ // re-delivered.
+ if (failureCanceledDels.remove(ack.getDeliveryID()) != null)
+ {
+ // ack should fail
+ throw new JMSException("Message already canceled before this ack " + ack +
+ " and the message will be redelivered.");
+ }
+ log.debug("Cannot find " + ack + " to acknowledge, it was probably acknowledged before");
+ return false;
+ }
}
- if (rec == null)
- {
- //This can happen if an ack comes in after failover
- log.debug("Cannot find " + ack + " to acknowledge, it was probably acknowledged before");
- return false;
- }
-
ServerConsumerEndpoint consumer = rec.getConsumer();
if (consumer != null && consumer.isRemote())
@@ -1857,6 +1978,8 @@
log.trace(this + " created and registered " + ep);
+ isSuckerSession.set(true);
+
return stub;
}
@@ -2409,4 +2532,9 @@
}
}
+ public PersistenceManager getPersistenceManager()
+ {
+ return pm;
+ }
+
}
Modified: branches/RIM_patch/src/main/org/jboss/messaging/core/contract/Delivery.java
===================================================================
--- branches/RIM_patch/src/main/org/jboss/messaging/core/contract/Delivery.java 2010-12-20 07:59:43 UTC (rev 8152)
+++ branches/RIM_patch/src/main/org/jboss/messaging/core/contract/Delivery.java 2010-12-20 11:19:06 UTC (rev 8153)
@@ -53,4 +53,8 @@
* Mark if this delivery is with a prepared XA transaction.
*/
boolean isXAPrepared();
+
+ boolean isSucked();
+
+ void setSucked(boolean isSucked);
}
Modified: branches/RIM_patch/src/main/org/jboss/messaging/core/contract/PersistenceManager.java
===================================================================
--- branches/RIM_patch/src/main/org/jboss/messaging/core/contract/PersistenceManager.java 2010-12-20 07:59:43 UTC (rev 8152)
+++ branches/RIM_patch/src/main/org/jboss/messaging/core/contract/PersistenceManager.java 2010-12-20 11:19:06 UTC (rev 8153)
@@ -99,6 +99,12 @@
//if supports transaction creation time
boolean supportsTxAge();
+ // update the status of the message
+ void updateMessageState(long channelID, MessageReference ref, String state) throws Exception;
+
+ // update messages state to 'C' of the channel whose state is 'S', and return their messages ids
+ List<ReferenceInfo> claimMessagesInSuck(long channelID) throws Exception;
+
// Interface value classes ----------------------------------------------------------------------
class MessageChannelPair
Modified: branches/RIM_patch/src/main/org/jboss/messaging/core/impl/ChannelSupport.java
===================================================================
--- branches/RIM_patch/src/main/org/jboss/messaging/core/impl/ChannelSupport.java 2010-12-20 07:59:43 UTC (rev 8152)
+++ branches/RIM_patch/src/main/org/jboss/messaging/core/impl/ChannelSupport.java 2010-12-20 11:19:06 UTC (rev 8153)
@@ -598,7 +598,7 @@
protected void cancelInternal(MessageReference ref) throws Exception
{
if (trace) { log.trace(this + " cancelling " + ref + " in memory"); }
-
+
synchronized (lock)
{
monitor.unmarkSending(ref);
Modified: branches/RIM_patch/src/main/org/jboss/messaging/core/impl/ClusterRoundRobinDistributor.java
===================================================================
--- branches/RIM_patch/src/main/org/jboss/messaging/core/impl/ClusterRoundRobinDistributor.java 2010-12-20 07:59:43 UTC (rev 8152)
+++ branches/RIM_patch/src/main/org/jboss/messaging/core/impl/ClusterRoundRobinDistributor.java 2010-12-20 11:19:06 UTC (rev 8153)
@@ -73,7 +73,6 @@
public Delivery handle(DeliveryObserver observer, MessageReference ref, Transaction tx)
{
//First try the local distributor
-
if (trace) { log.trace(this + " first trying with local distributor"); }
Delivery del = localDistributor.handle(observer, ref, tx);
@@ -85,7 +84,7 @@
//If no local distributor takes the ref then we try the remote distributor
if (trace) { log.trace(this + " trying with remote distributor"); }
-
+
del = remoteDistributor.handle(observer, ref, tx);
if (trace) { log.trace(this + " remote distributor returned " + del); }
Modified: branches/RIM_patch/src/main/org/jboss/messaging/core/impl/JDBCPersistenceManager.java
===================================================================
--- branches/RIM_patch/src/main/org/jboss/messaging/core/impl/JDBCPersistenceManager.java 2010-12-20 07:59:43 UTC (rev 8152)
+++ branches/RIM_patch/src/main/org/jboss/messaging/core/impl/JDBCPersistenceManager.java 2010-12-20 11:19:06 UTC (rev 8153)
@@ -36,6 +36,7 @@
import org.jboss.messaging.util.StreamUtils;
import org.jboss.messaging.util.Util;
+import javax.jms.JMSException;
import javax.sql.DataSource;
import javax.transaction.TransactionManager;
import javax.transaction.xa.Xid;
@@ -1637,6 +1638,12 @@
log.trace("Updated " + rows + " rows");
}
+ if (rows == 0)
+ {
+ // no message updated, should be canceled back already
+ throw new JMSException("Failed to move message " + ref.getMessage().getMessageID());
+ }
+
return null;
}
finally
@@ -2947,6 +2954,12 @@
map.put("INSERT_TRANSACTION_EXTRA", "INSERT INTO JBM_TX_EX (TRANSACTION_ID, START_TIME) VALUES(?,?)");
map.put("DELETE_TRANSACTION_EXTRA", "DELETE FROM JBM_TX_EX WHERE TRANSACTION_ID=?");
+ // sucker use
+ map.put("UPDATE_MESSAGE_STATE", "UPDATE JBM_MSG_REF SET STATE = ? WHERE MESSAGE_ID = ? AND CHANNEL_ID = ?");
+ map.put("CLAIM_MESSAGE_IN_SUCK", "UPDATE JBM_MSG_REF SET STATE='C' WHERE CHANNEL_ID = ? AND STATE='S'");
+ map.put("LOAD_REFS_IN_SUCK",
+ "SELECT MESSAGE_ID, DELIVERY_COUNT, SCHED_DELIVERY FROM JBM_MSG_REF WHERE STATE = 'S' AND CHANNEL_ID = ? ORDER BY ORD");
+
return map;
}
@@ -3308,4 +3321,96 @@
return supportsTxAge;
}
+ public void updateMessageState(final long channelID, final MessageReference ref, final String c) throws Exception
+ {
+ class UpdateMessageStateRunner extends JDBCTxRunner2
+ {
+ public Object doTransaction() throws Exception
+ {
+ PreparedStatement psReference = null;
+
+ try
+ {
+ psReference = conn.prepareStatement(getSQLStatement("UPDATE_MESSAGE_STATE"));
+
+ psReference.setString(1, c);
+
+ psReference.setLong(2, ref.getMessage().getMessageID());
+
+ psReference.setLong(3, channelID);
+
+ int rows = psReference.executeUpdate();
+
+ if (trace)
+ {
+ log.trace("Updated " + rows + " rows");
+ }
+
+ if (rows != 1)
+ {
+ throw new JMSException("Failed to update message " + ref.getMessage().getMessageID() +
+ " to state " +
+ c);
+ }
+
+ return null;
+ }
+ finally
+ {
+ closeStatement(psReference);
+ }
+ }
+ }
+
+ new UpdateMessageStateRunner().executeWithRetry();
+ }
+
+ public List<ReferenceInfo> claimMessagesInSuck(final long channelID) throws Exception
+ {
+ final List<ReferenceInfo> msgIDs = new ArrayList<ReferenceInfo>();
+
+ class MessageClaimRunner extends JDBCTxRunner2
+ {
+ public Object doTransaction() throws Exception
+ {
+ PreparedStatement ps = null;
+ PreparedStatement ps2 = null;
+ ResultSet rs = null;
+
+ try
+ {
+ ps = conn.prepareStatement(getSQLStatement("LOAD_REFS_IN_SUCK"));
+ ps.setLong(1, channelID);
+ rs = ps.executeQuery();
+
+ while (rs.next())
+ {
+ long msgId = rs.getLong(1);
+ int deliveryCount = rs.getInt(2);
+ long sched = rs.getLong(3);
+
+ msgIDs.add(new ReferenceInfo(msgId, deliveryCount, sched));
+ }
+
+ ps2 = conn.prepareStatement(getSQLStatement("CLAIM_MESSAGE_IN_SUCK"));
+ ps2.setLong(1, channelID);
+ int rows = ps2.executeUpdate();
+
+ if (trace)
+ {
+ log.trace("Message in suck claimed " + rows + " rows");
+ }
+ }
+ finally
+ {
+ closeResultSet(rs);
+ closeStatement(ps);
+ closeStatement(ps2);
+ }
+ return msgIDs;
+ }
+ }
+ new MessageClaimRunner().executeWithRetry();
+ return msgIDs;
+ }
}
Modified: branches/RIM_patch/src/main/org/jboss/messaging/core/impl/MessagingQueue.java
===================================================================
--- branches/RIM_patch/src/main/org/jboss/messaging/core/impl/MessagingQueue.java 2010-12-20 07:59:43 UTC (rev 8152)
+++ branches/RIM_patch/src/main/org/jboss/messaging/core/impl/MessagingQueue.java 2010-12-20 11:19:06 UTC (rev 8153)
@@ -221,6 +221,8 @@
synchronized (lock)
{
flushDownCache();
+
+ pm.claimMessagesInSuck(theChannelID);
PersistenceManager.InitialLoadInfo ili =
pm.mergeAndLoad(theChannelID, this.channelID, fullSize - messageRefs.size(),
@@ -491,7 +493,12 @@
{
return recoverDeliveriesTimeout;
}
-
+
+ public void load() throws Exception
+ {
+ super.load(clustered);
+ }
+
//testing only
public Map getRecoveryArea()
Modified: branches/RIM_patch/src/main/org/jboss/messaging/core/impl/NullPersistenceManager.java
===================================================================
--- branches/RIM_patch/src/main/org/jboss/messaging/core/impl/NullPersistenceManager.java 2010-12-20 07:59:43 UTC (rev 8152)
+++ branches/RIM_patch/src/main/org/jboss/messaging/core/impl/NullPersistenceManager.java 2010-12-20 11:19:06 UTC (rev 8153)
@@ -228,6 +228,15 @@
{
}
+ public void updateMessageState(long channelID, MessageReference ref, String state) throws Exception
+ {
+ }
+
+ public List<ReferenceInfo> claimMessagesInSuck(long channelID) throws Exception
+ {
+ return Collections.emptyList();
+ }
+
}
class IDCounter
Modified: branches/RIM_patch/src/main/org/jboss/messaging/core/impl/PagingChannelSupport.java
===================================================================
--- branches/RIM_patch/src/main/org/jboss/messaging/core/impl/PagingChannelSupport.java 2010-12-20 07:59:43 UTC (rev 8152)
+++ branches/RIM_patch/src/main/org/jboss/messaging/core/impl/PagingChannelSupport.java 2010-12-20 11:19:06 UTC (rev 8153)
@@ -194,7 +194,7 @@
}
}
- public void load() throws Exception
+ public void load(boolean isClustered) throws Exception
{
synchronized (lock)
{
@@ -207,6 +207,12 @@
unload();
+ if (isClustered)
+ {
+ // claim possible sucked messages
+ pm.claimMessagesInSuck(channelID);
+ }
+
//Load the unpaged references
InitialLoadInfo ili = pm.loadFromStart(channelID, fullSize);
Modified: branches/RIM_patch/src/main/org/jboss/messaging/core/impl/SimpleDelivery.java
===================================================================
--- branches/RIM_patch/src/main/org/jboss/messaging/core/impl/SimpleDelivery.java 2010-12-20 07:59:43 UTC (rev 8152)
+++ branches/RIM_patch/src/main/org/jboss/messaging/core/impl/SimpleDelivery.java 2010-12-20 11:19:06 UTC (rev 8153)
@@ -51,7 +51,7 @@
private MessageReference reference;
private boolean recovered;
private Transaction tx;
-
+ private boolean isSucked;
private boolean trace = log.isTraceEnabled();
// Constructors ---------------------------------------------------------------------------------
@@ -136,6 +136,16 @@
return false;
}
+ public void setSucked(boolean sucked)
+ {
+ isSucked = sucked;
+ }
+
+ public boolean isSucked()
+ {
+ return isSucked;
+ }
+
// Package protected ----------------------------------------------------------------------------
// Protected ------------------------------------------------------------------------------------
Modified: branches/RIM_patch/src/main/org/jboss/messaging/core/impl/clusterconnection/MessageSucker.java
===================================================================
--- branches/RIM_patch/src/main/org/jboss/messaging/core/impl/clusterconnection/MessageSucker.java 2010-12-20 07:59:43 UTC (rev 8152)
+++ branches/RIM_patch/src/main/org/jboss/messaging/core/impl/clusterconnection/MessageSucker.java 2010-12-20 11:19:06 UTC (rev 8153)
@@ -27,7 +27,6 @@
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.Session;
-import javax.transaction.Transaction;
import javax.transaction.TransactionManager;
import org.jboss.jms.client.JBossSession;
@@ -212,28 +211,13 @@
suspended = true;
- try
- {
- consumer.closing(-1);
- }
- catch (Throwable t)
- {
- // Ignore
- }
- try
- {
- consumer.close();
- }
- catch (Throwable t)
- {
- // Ignore
- }
-
sourceSession = null;
consumer = null;
clientConsumer = null;
+
+ if (trace) { log.trace(this + " suspended"); }
}
@@ -327,8 +311,6 @@
public void onMessage(Message msg)
{
- Transaction tx = null;
-
try
{
if (trace) { log.trace(this + " sucked message " + msg + " JMSDestination - " + msg.getJMSDestination()); }
@@ -375,15 +357,6 @@
catch (Exception e)
{
log.error("Failed to forward message", e);
-
- try
- {
- if (tx != null) tm.rollback();
- }
- catch (Throwable t)
- {
- if (trace) { log.trace("Failed to rollback tx", t); }
- }
}
}
}
More information about the jboss-cvs-commits
mailing list