[jboss-cvs] JBoss Messaging SVN: r1369 - in trunk: . src/etc/server/default/deploy src/main/org/jboss/jms/server/destination src/main/org/jboss/messaging/core/plugin src/main/org/jboss/messaging/core/plugin/postoffice src/main/org/jboss/messaging/core/plugin/postoffice/cluster src/main/org/jboss/messaging/core/tx tests/src/org/jboss/test/messaging/jms tests/src/org/jboss/test/messaging/jms/crash
jboss-cvs-commits at lists.jboss.org
jboss-cvs-commits at lists.jboss.org
Wed Sep 27 06:11:26 EDT 2006
Author: timfox
Date: 2006-09-27 06:11:15 -0400 (Wed, 27 Sep 2006)
New Revision: 1369
Added:
trunk/tests/src/org/jboss/test/messaging/jms/crash/CreateTwoClientOnServerCommand.java
Removed:
trunk/tests/src/org/jboss/test/messaging/jms/CreateTwoClientOnServerCommand.java
Modified:
trunk/.classpath
trunk/build-thirdparty.xml
trunk/src/etc/server/default/deploy/clustered-mysql-persistence-service.xml
trunk/src/main/org/jboss/jms/server/destination/QueueService.java
trunk/src/main/org/jboss/messaging/core/plugin/ClusteredPostOfficeService.java
trunk/src/main/org/jboss/messaging/core/plugin/postoffice/DefaultPostOffice.java
trunk/src/main/org/jboss/messaging/core/plugin/postoffice/cluster/DefaultClusteredPostOffice.java
trunk/src/main/org/jboss/messaging/core/plugin/postoffice/cluster/DefaultRouter.java
trunk/src/main/org/jboss/messaging/core/tx/Transaction.java
trunk/src/main/org/jboss/messaging/core/tx/TransactionRepository.java
trunk/tests/src/org/jboss/test/messaging/jms/crash/ClientCrashTwoConnectionsTest.java
Log:
Fixed build, merged last changes
Modified: trunk/.classpath
===================================================================
--- trunk/.classpath 2006-09-27 07:55:32 UTC (rev 1368)
+++ trunk/.classpath 2006-09-27 10:11:15 UTC (rev 1369)
@@ -24,16 +24,11 @@
<classpathentry kind="lib" path="perf/resources/jcommon-1.0.0-rc1.jar"/>
<classpathentry kind="lib" path="perf/resources/jfreechart-1.0.0-rc1.jar"/>
<classpathentry kind="lib" path="thirdparty/apache-log4j/lib/log4j.jar"/>
- <classpathentry sourcepath="/home/clebert/workspaces/jboss-head/aop/src/main" kind="lib" path="thirdparty/jboss/aop/lib/jboss-aop.jar"/>
<classpathentry kind="lib" path="thirdparty/junit/lib/junit.jar"/>
<classpathentry kind="lib" path="thirdparty/jboss/profiler/jvmti/lib/jboss-profiler-jvmti.jar"/>
<classpathentry kind="lib" path="thirdparty/hsqldb/lib/hsqldb.jar"/>
<classpathentry kind="lib" path="thirdparty/apache-logging/lib/commons-logging.jar"/>
- <classpathentry kind="lib" path="thirdparty/jboss/remoting/lib/jboss-remoting.jar"/>
<classpathentry kind="lib" path="thirdparty/jboss/serialization/lib/jboss-serialization.jar"/>
- <classpathentry kind="var" path="ANT_HOME/ant-junit.jar"/>
- <classpathentry sourcepath="/JBossRemoting" kind="lib" path="thirdparty/jboss/remoting/lib/jboss-remoting.jar"/>
- <classpathentry kind="lib" path="thirdparty/jboss/serialization/lib/jboss-serialization.jar"/>
<classpathentry kind="con" path="org.eclipse.jdt.launching.JRE_CONTAINER"/>
<classpathentry kind="lib" path="thirdparty/sun-javacc/lib/javacc.jar"/>
<classpathentry kind="lib" path="thirdparty/apache-log4j/lib/snmpTrapAppender.jar"/>
@@ -57,5 +52,7 @@
<classpathentry kind="lib" path="thirdparty/retrotranslator/lib/retrotranslator-runtime.jar"/>
<classpathentry kind="lib" path="thirdparty/retrotranslator/lib/retrotranslator-transformer.jar"/>
<classpathentry kind="lib" path="thirdparty/trove/lib/trove.jar"/>
+ <classpathentry kind="lib" path="thirdparty/jboss/remoting/lib/jboss-remoting.jar"/>
+ <classpathentry kind="var" path="ANT_HOME/lib/ant-junit.jar"/>
<classpathentry kind="output" path="bin"/>
</classpath>
Modified: trunk/build-thirdparty.xml
===================================================================
--- trunk/build-thirdparty.xml 2006-09-27 07:55:32 UTC (rev 1368)
+++ trunk/build-thirdparty.xml 2006-09-27 10:11:15 UTC (rev 1369)
@@ -106,7 +106,6 @@
<componentref name="jboss/jbossxb" version="1.0.0.CR6"/>
-
</build>
<synchronizeinfo/>
Modified: trunk/src/etc/server/default/deploy/clustered-mysql-persistence-service.xml
===================================================================
--- trunk/src/etc/server/default/deploy/clustered-mysql-persistence-service.xml 2006-09-27 07:55:32 UTC (rev 1368)
+++ trunk/src/etc/server/default/deploy/clustered-mysql-persistence-service.xml 2006-09-27 10:11:15 UTC (rev 1369)
@@ -79,8 +79,8 @@
<attribute name="AsyncChannelConfig">
<config>
<UDP mcast_recv_buf_size="500000" down_thread="false" ip_mcast="true" mcast_send_buf_size="32000"
- mcast_port="45568" ucast_recv_buf_size="500000" use_incoming_packet_handler="false"
- mcast_addr="228.8.8.8" use_outgoing_packet_handler="true" loopback="false" ucast_send_buf_size="32000" ip_ttl="32"/>
+ mcast_port="45567" ucast_recv_buf_size="500000" use_incoming_packet_handler="false"
+ mcast_addr="228.8.8.8" use_outgoing_packet_handler="true" loopback="true" ucast_send_buf_size="32000" ip_ttl="32" bind_addr="127.0.0.1"/>
<AUTOCONF down_thread="false" up_thread="false"/>
<PING timeout="2000" down_thread="false" num_initial_members="3" up_thread="false"/>
<MERGE2 max_interval="10000" down_thread="false" min_interval="5000" up_thread="false"/>
@@ -99,8 +99,8 @@
<attribute name="SyncChannelConfig">
<config>
<UDP mcast_recv_buf_size="500000" down_thread="false" ip_mcast="true" mcast_send_buf_size="32000"
- mcast_port="45567" ucast_recv_buf_size="500000" use_incoming_packet_handler="false"
- mcast_addr="228.8.8.8" use_outgoing_packet_handler="true" loopback="false" ucast_send_buf_size="32000" ip_ttl="32"/>
+ mcast_port="45568" ucast_recv_buf_size="500000" use_incoming_packet_handler="false"
+ mcast_addr="228.8.8.8" use_outgoing_packet_handler="true" loopback="true" ucast_send_buf_size="32000" ip_ttl="32" bind_addr="127.0.0.1"/>
<AUTOCONF down_thread="false" up_thread="false"/>
<PING timeout="2000" down_thread="false" num_initial_members="3" up_thread="false"/>
<MERGE2 max_interval="10000" down_thread="false" min_interval="5000" up_thread="false"/>
@@ -142,8 +142,8 @@
<attribute name="AsyncChannelConfig">
<config>
<UDP mcast_recv_buf_size="500000" down_thread="false" ip_mcast="true" mcast_send_buf_size="32000"
- mcast_port="45568" ucast_recv_buf_size="500000" use_incoming_packet_handler="false"
- mcast_addr="228.8.8.8" use_outgoing_packet_handler="true" loopback="false" ucast_send_buf_size="32000" ip_ttl="32"/>
+ mcast_port="45569" ucast_recv_buf_size="500000" use_incoming_packet_handler="false"
+ mcast_addr="228.8.8.8" use_outgoing_packet_handler="true" loopback="true" ucast_send_buf_size="32000" ip_ttl="32" bind_addr="127.0.0.1"/>
<AUTOCONF down_thread="false" up_thread="false"/>
<PING timeout="2000" down_thread="false" num_initial_members="3" up_thread="false"/>
<MERGE2 max_interval="10000" down_thread="false" min_interval="5000" up_thread="false"/>
@@ -162,8 +162,8 @@
<attribute name="SyncChannelConfig">
<config>
<UDP mcast_recv_buf_size="500000" down_thread="false" ip_mcast="true" mcast_send_buf_size="32000"
- mcast_port="45569" ucast_recv_buf_size="500000" use_incoming_packet_handler="false"
- mcast_addr="228.8.8.8" use_outgoing_packet_handler="true" loopback="false" ucast_send_buf_size="32000" ip_ttl="32"/>
+ mcast_port="45570" ucast_recv_buf_size="500000" use_incoming_packet_handler="false"
+ mcast_addr="228.8.8.8" use_outgoing_packet_handler="true" loopback="true" ucast_send_buf_size="32000" ip_ttl="32" bind_addr="127.0.0.1"/>
<AUTOCONF down_thread="false" up_thread="false"/>
<PING timeout="2000" down_thread="false" num_initial_members="3" up_thread="false"/>
<MERGE2 max_interval="10000" down_thread="false" min_interval="5000" up_thread="false"/>
Modified: trunk/src/main/org/jboss/jms/server/destination/QueueService.java
===================================================================
--- trunk/src/main/org/jboss/jms/server/destination/QueueService.java 2006-09-27 07:55:32 UTC (rev 1368)
+++ trunk/src/main/org/jboss/jms/server/destination/QueueService.java 2006-09-27 10:11:15 UTC (rev 1369)
@@ -92,6 +92,8 @@
if (binding != null)
{
+ log.info("binding already exists");
+
PagingFilteredQueue queue = (PagingFilteredQueue)binding.getQueue();
queue.setPagingParams(destination.getFullSize(), destination.getPageSize(), destination.getDownCacheSize());
@@ -108,6 +110,8 @@
if (postOffice.isLocal())
{
+ log.info("Binding non clustered");
+
queue = new PagingFilteredQueue(destination.getName(), idm.getId(), ms, pm, true, true,
executor, null,
destination.getFullSize(), destination.getPageSize(), destination.getDownCacheSize());
@@ -116,6 +120,8 @@
}
else
{
+ log.info("binding clustered");
+
queue = new LocalClusteredQueue(postOffice, nodeId, destination.getName(), idm.getId(), ms, pm, true, true,
executor, null, tr,
destination.getFullSize(), destination.getPageSize(), destination.getDownCacheSize());
Modified: trunk/src/main/org/jboss/messaging/core/plugin/ClusteredPostOfficeService.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/plugin/ClusteredPostOfficeService.java 2006-09-27 07:55:32 UTC (rev 1368)
+++ trunk/src/main/org/jboss/messaging/core/plugin/ClusteredPostOfficeService.java 2006-09-27 10:11:15 UTC (rev 1369)
@@ -34,9 +34,9 @@
import org.jboss.messaging.core.plugin.contract.PersistenceManager;
import org.jboss.messaging.core.plugin.postoffice.cluster.ClusterRouterFactory;
import org.jboss.messaging.core.plugin.postoffice.cluster.DefaultClusteredPostOffice;
+import org.jboss.messaging.core.plugin.postoffice.cluster.DefaultMessagePullPolicy;
import org.jboss.messaging.core.plugin.postoffice.cluster.DefaultRouterFactory;
import org.jboss.messaging.core.plugin.postoffice.cluster.MessagePullPolicy;
-import org.jboss.messaging.core.plugin.postoffice.cluster.NullMessagePullPolicy;
import org.jboss.messaging.core.tx.TransactionRepository;
import org.w3c.dom.Element;
@@ -217,7 +217,7 @@
String nodeId = serverPeer.getServerPeerID();
- MessagePullPolicy pullPolicy = new NullMessagePullPolicy();
+ MessagePullPolicy pullPolicy = new DefaultMessagePullPolicy();
FilterFactory ff = new SelectorFactory();
Modified: trunk/src/main/org/jboss/messaging/core/plugin/postoffice/DefaultPostOffice.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/plugin/postoffice/DefaultPostOffice.java 2006-09-27 07:55:32 UTC (rev 1368)
+++ trunk/src/main/org/jboss/messaging/core/plugin/postoffice/DefaultPostOffice.java 2006-09-27 10:11:15 UTC (rev 1369)
@@ -374,11 +374,13 @@
{
lock.writeLock().acquire();
+ log.info("loading bindings");
+
Connection conn = null;
PreparedStatement ps = null;
ResultSet rs = null;
TransactionWrapper wrap = new TransactionWrapper();
-
+
try
{
conn = ds.getConnection();
@@ -454,7 +456,7 @@
return binding;
}
-
+
protected void insertBinding(Binding binding) throws Exception
{
Connection conn = null;
Modified: trunk/src/main/org/jboss/messaging/core/plugin/postoffice/cluster/DefaultClusteredPostOffice.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/plugin/postoffice/cluster/DefaultClusteredPostOffice.java 2006-09-27 07:55:32 UTC (rev 1368)
+++ trunk/src/main/org/jboss/messaging/core/plugin/postoffice/cluster/DefaultClusteredPostOffice.java 2006-09-27 10:11:15 UTC (rev 1369)
@@ -214,7 +214,7 @@
{
super (ds, tm, sqlProperties, createTablesOnStartup, nodeId, officeName, ms, pm, tr, filterFactory,
pool);
-
+
this.pm = pm;
this.groupName = groupName;
@@ -241,6 +241,8 @@
public synchronized void start() throws Exception
{
+ log.info("DefaultClusteredPosttoffic::start");
+
if (syncChannelConfigE != null)
{
this.syncChannel = new JChannel(syncChannelConfigE);
@@ -274,7 +276,7 @@
asyncChannel.connect(groupName);
super.start();
-
+
Address currentAddress = syncChannel.getLocalAddress();
log.info(this.nodeId + " address is " + currentAddress);
@@ -305,6 +307,8 @@
public Binding bindClusteredQueue(String condition, LocalClusteredQueue queue) throws Exception
{
+ log.info(this.nodeId + " binding clustered queue: " + queue + " with condition: " + condition);
+
if (!queue.getNodeId().equals(this.nodeId))
{
throw new IllegalArgumentException("Queue node id does not match office node id");
@@ -357,6 +361,8 @@
lock.readLock().acquire();
+ log.info(this.nodeId + " routing reference " + ref);
+
try
{
ClusteredBindings cb = (ClusteredBindings)conditionMap.get(condition);
@@ -407,7 +413,7 @@
ClusteredQueue queue = (ClusteredQueue)del.getObserver();
- // log.info("Routing message to queue:" + queue.getName() + " on node " + queue.getNodeId());
+ log.info("Routing message to queue:" + queue.getName() + " on node " + queue.getNodeId());
if (router.numberOfReceivers() > 1)
{
@@ -517,6 +523,8 @@
{
lock.writeLock().acquire();
+ log.info(this.nodeId + " adding binding from node: " + nodeId +" queue: " + queueName + " with condition: " + condition);
+
try
{
//Sanity
@@ -591,8 +599,8 @@
public void routeFromCluster(org.jboss.messaging.core.Message message, String routingKey,
Map queueNameNodeIdMap) throws Exception
{
- // log.info(this.nodeId + " received route from cluster, ref = " + message.getMessageID() + " routing key " +
- // routingKey + " map " + queueNameNodeIdMap);
+ log.info(this.nodeId + " received route from cluster, ref = " + message.getMessageID() + " routing key " +
+ routingKey + " map " + queueNameNodeIdMap);
lock.readLock().acquire();
@@ -1037,12 +1045,17 @@
if (!isState)
{
//Must be first member in group or non clustered- we load the state ourself from the database
+
+ log.info("First member - so loading bindings from db");
+
super.loadBindings();
}
else
{
//The state will be set in due course via the MessageListener - we must wait until this happens
+ log.info("Not first member - so loading state from group.. waiting");
+
synchronized (setStateLock)
{
//TODO we should implement a timeout on this
@@ -1051,6 +1064,8 @@
setStateLock.wait();
}
}
+
+ log.info("Got state");
}
}
@@ -1058,16 +1073,22 @@
{
Filter filter = filterFactory.createFilter(filterString);
+ log.info("Created binding");
+
Queue queue;
if (nodeId.equals(this.nodeId))
{
QueuedExecutor executor = (QueuedExecutor)pool.get();
+ log.info("created local clustered queue");
+
queue = new LocalClusteredQueue(this, nodeId, queueName, channelId, ms, pm, true,
durable, executor, filter, tr);
}
else
{
+ log.info("created remote queue stub");
+
queue = new RemoteQueueStub(nodeId, queueName, channelId, durable, pm, filter);
}
@@ -1137,6 +1158,8 @@
if (!binding.getQueue().isRecoverable())
{
+ //We only remove the non durable bindings - we still need to be able to handle
+ //messages for a durable subscription "owned" by a node that is not active any more!
toRemove.add(binding);
}
}
@@ -1196,6 +1219,8 @@
private void processStateBytes(byte[] bytes) throws Exception
{
+ log.info("Receiving state from group...");
+
SharedState state = new SharedState();
StreamUtils.fromBytes(state, bytes);
@@ -1214,7 +1239,16 @@
Binding binding = this.createBinding(info.getNodeId(), info.getCondition(), info.getQueueName(), info.getChannelId(), info.getFilterString(), info.isDurable());
- addBinding(binding);
+ if (binding.getNodeId().equals(this.nodeId))
+ {
+ //We deactivate if this is one of our own bindings - it can only
+ //be one of our own durable bindings - and since state is retrieved before we are fully started
+ //then the sub hasn't been deployed so must be deactivated
+
+ binding.getQueue().deactivate();
+ }
+
+ addBinding(binding);
}
this.nodeIdAddressMap.clear();
@@ -1344,6 +1378,8 @@
public void viewAccepted(View view)
{
+ log.info("Got new view, size=" + view.size());
+
if (currentView != null)
{
Iterator iter = currentView.getMembers().iterator();
Modified: trunk/src/main/org/jboss/messaging/core/plugin/postoffice/cluster/DefaultRouter.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/plugin/postoffice/cluster/DefaultRouter.java 2006-09-27 07:55:32 UTC (rev 1368)
+++ trunk/src/main/org/jboss/messaging/core/plugin/postoffice/cluster/DefaultRouter.java 2006-09-27 10:11:15 UTC (rev 1369)
@@ -144,6 +144,8 @@
public Delivery handle(DeliveryObserver observer, MessageReference reference, Transaction tx)
{
//Favour the local queue
+
+ log.info("router handling ref, it has " + this.queues.size());
if (localQueue != null)
{
Modified: trunk/src/main/org/jboss/messaging/core/tx/Transaction.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/tx/Transaction.java 2006-09-27 07:55:32 UTC (rev 1368)
+++ trunk/src/main/org/jboss/messaging/core/tx/Transaction.java 2006-09-27 10:11:15 UTC (rev 1369)
@@ -32,15 +32,13 @@
import org.jboss.logging.Logger;
/**
- *
+ *
* A JMS Server local transaction
- *
+ *
* @author <a href="mailto:tim.fox at jboss.com">Tim Fox</a>
* @author <a href="mailto:ovidiu at jboss.com">Ovidiu Feodorov</a>
- *
+ *
* @version $Revision 1.1$
- *
- * $Id$
*/
public class Transaction
{
@@ -49,35 +47,38 @@
private static final Logger log = Logger.getLogger(Transaction.class);
// Attributes ----------------------------------------------------
-
+
private boolean trace = log.isTraceEnabled();
-
+
protected long id;
-
+
protected int state;
-
+
protected Xid xid;
-
+
+ protected List callbacks;
+
+ protected Map callbackMap;
+
+
/**
* If this is a XA transaction, when a commit is executed the transaction has to be removed from the transaction repository.
* This reference will guarantee the reference back to the repository where the transaction was created
- * */
- protected TransactionRepository transactionRepository;
-
- protected List callbacks;
-
- protected List keyedCallbacks;
-
- protected Map keyedCallbackMap;
-
+ *
+ */
+ protected TransactionRepository repository;
+
+ //A special first callback that is ensured to be executed first
+ protected TxCallback firstCallback;
+
// Static --------------------------------------------------------
-
+
public static final int STATE_ACTIVE = 0;
-
+
public static final int STATE_PREPARED = 1;
-
+
public static final int STATE_COMMITTED = 2;
-
+
public static final int STATE_ROLLEDBACK = 3;
public static final int STATE_ROLLBACK_ONLY = 4;
@@ -111,52 +112,58 @@
}
// Constructors --------------------------------------------------
-
+
Transaction(long id)
{
this.id = id;
state = STATE_ACTIVE;
callbacks = new ArrayList();
- keyedCallbacks = new ArrayList();
- keyedCallbackMap = new HashMap();
+ callbackMap = new HashMap();
}
-
- Transaction(long id, Xid xid, TransactionRepository repository)
+
+ Transaction(long id, Xid xid, TransactionRepository tr)
{
this(id);
this.xid = xid;
- this.transactionRepository=repository;
+ this.repository = tr;
}
-
+
// Public --------------------------------------------------------
-
+
public int getState()
{
return state;
- }
-
+ }
+
public Xid getXid()
{
return xid;
}
- public void addCallback(TxCallback callback)
- {
+ public void addCallback(TxCallback callback, Object key)
+ {
callbacks.add(callback);
+
+ callbackMap.put(key, callback);
+ }
+
+ public void addFirstCallback(TxCallback callback, Object key)
+ {
+ if (firstCallback != null)
+ {
+ throw new IllegalStateException("There is already a first callback");
+ }
+
+ this.firstCallback = callback;
+
+ callbackMap.put(key, callback);
}
-
- public void addKeyedCallback(TxCallback callback, Object key)
+
+ public TxCallback getCallback(Object key)
{
- keyedCallbacks.add(callback);
-
- keyedCallbackMap.put(key, callback);
+ return (TxCallback)callbackMap.get(key);
}
-
- public TxCallback getKeyedCallback(Object key)
- {
- return (TxCallback)keyedCallbackMap.get(key);
- }
-
+
public synchronized void commit() throws Exception
{
if (state == STATE_ROLLBACK_ONLY)
@@ -173,89 +180,103 @@
}
if (trace) { log.trace("executing before commit hooks " + this); }
-
+
boolean onePhase = state != STATE_PREPARED;
-
- List cb = new ArrayList(callbacks);
- cb.addAll(keyedCallbacks);
-
- Iterator iter = cb.iterator();
-
+
+ if (firstCallback != null)
+ {
+ firstCallback.beforeCommit(onePhase);
+ }
+
+ Iterator iter = callbacks.iterator();
+
while (iter.hasNext())
{
TxCallback callback = (TxCallback)iter.next();
-
+
callback.beforeCommit(onePhase);
}
-
+
state = STATE_COMMITTED;
-
+
if (trace) { log.trace("committed " + this); }
-
- iter = cb.iterator();
-
+
+ iter = callbacks.iterator();
+
if (trace) { log.trace("executing after commit hooks " + this); }
-
+
+ if (firstCallback != null)
+ {
+ firstCallback.afterCommit(onePhase);
+ }
+
while (iter.hasNext())
{
TxCallback callback = (TxCallback)iter.next();
-
+
callback.afterCommit(onePhase);
}
-
+
callbacks = null;
-
- keyedCallbacks = null;
-
- keyedCallbackMap = null;
-
- if (transactionRepository!=null)
+
+ callbackMap = null;
+
+ firstCallback = null;
+
+ if (repository !=null)
{
- transactionRepository.deleteTransaction(this);
+ repository.deleteTransaction(this);
}
-
- if (trace) { log.trace("commit process complete " + this); }
+
+ if (trace) { log.trace("commit process complete " + this); }
}
-
+
public synchronized void prepare() throws Exception
{
if (state != STATE_ACTIVE)
{
throw new TransactionException("Transaction not active, cannot prepare");
}
-
+
if (trace) { log.trace("executing before prepare hooks " + this); }
-
- List cb = new ArrayList(callbacks);
- cb.addAll(keyedCallbacks);
-
- Iterator iter = cb.iterator();
-
+
+ if (firstCallback != null)
+ {
+ firstCallback.beforePrepare();
+ }
+
+ Iterator iter = callbacks.iterator();
+
while (iter.hasNext())
{
TxCallback callback = (TxCallback)iter.next();
-
+
callback.beforePrepare();
}
-
+
state = STATE_PREPARED;
-
+
if (trace) { log.trace("prepared " + this); }
-
- iter = cb.iterator();
-
+
+ if (firstCallback != null)
+ {
+ firstCallback.afterPrepare();
+ }
+
+ iter = callbacks.iterator();
+
if (trace) { log.trace("executing after prepare hooks " + this); }
-
+
while (iter.hasNext())
{
TxCallback callback = (TxCallback)iter.next();
-
+
callback.afterPrepare();
- }
-
+ }
+
if (trace) { log.trace("prepare process complete " + this); }
}
-
+
public synchronized void rollback() throws Exception
{
if (state == STATE_COMMITTED)
@@ -266,41 +287,47 @@
{
throw new TransactionException("Transaction already rolled back, cannot rollback");
}
-
+
if (trace) { log.trace("executing before rollback hooks " + this); }
-
+
boolean onePhase = state != STATE_PREPARED;
+
+ if (firstCallback != null)
+ {
+ firstCallback.beforeRollback(onePhase);
+ }
- List cb = new ArrayList(callbacks);
- cb.addAll(keyedCallbacks);
-
- for(Iterator i = cb.iterator(); i.hasNext(); )
+ for(Iterator i = callbacks.iterator(); i.hasNext(); )
{
TxCallback callback = (TxCallback)i.next();
callback.beforeRollback(onePhase);
}
-
+
state = STATE_ROLLEDBACK;
-
+
if (trace) { log.trace("rolled back " + this); }
if (trace) { log.trace("executing after prepare hooks " + this); }
- for(Iterator i = cb.iterator(); i.hasNext();)
+ if (firstCallback != null)
{
+ firstCallback.afterRollback(onePhase);
+ }
+
+ for(Iterator i = callbacks.iterator(); i.hasNext();)
+ {
TxCallback callback = (TxCallback)i.next();
callback.afterRollback(onePhase);
- }
-
+ }
+
callbacks = null;
- keyedCallbacks = null;
- keyedCallbackMap = null;
-
- if (transactionRepository!=null)
+ callbackMap = null;
+
+ if (repository != null)
{
- transactionRepository.deleteTransaction(this);
+ repository.deleteTransaction(this);
}
-
+
if (trace) { log.trace("rollback process complete " + this); }
}
@@ -310,12 +337,12 @@
state = STATE_ROLLBACK_ONLY;
}
-
+
public long getId()
{
return id;
}
-
+
public String toString()
{
StringBuffer sb = new StringBuffer("TX(");
@@ -326,13 +353,13 @@
}
// Package protected ---------------------------------------------
-
+
// Protected -----------------------------------------------------
-
+
// Private -------------------------------------------------------
-
+
// Inner classes -------------------------------------------------
-
+
}
Modified: trunk/src/main/org/jboss/messaging/core/tx/TransactionRepository.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/tx/TransactionRepository.java 2006-09-27 07:55:32 UTC (rev 1368)
+++ trunk/src/main/org/jboss/messaging/core/tx/TransactionRepository.java 2006-09-27 10:11:15 UTC (rev 1369)
@@ -159,8 +159,6 @@
globalToLocalMap.remove(id);
-
-
}
public Transaction createTransaction(Xid xid) throws Exception
Deleted: trunk/tests/src/org/jboss/test/messaging/jms/CreateTwoClientOnServerCommand.java
===================================================================
--- trunk/tests/src/org/jboss/test/messaging/jms/CreateTwoClientOnServerCommand.java 2006-09-27 07:55:32 UTC (rev 1368)
+++ trunk/tests/src/org/jboss/test/messaging/jms/CreateTwoClientOnServerCommand.java 2006-09-27 10:11:15 UTC (rev 1369)
@@ -1,195 +0,0 @@
-<<<<<<< .working
-/*
- * 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;
-
-import java.util.ArrayList;
-import java.util.List;
-
-import javax.jms.Connection;
-import javax.jms.ConnectionFactory;
-import javax.jms.DeliveryMode;
-import javax.jms.MessageConsumer;
-import javax.jms.MessageProducer;
-import javax.jms.Queue;
-import javax.jms.Session;
-import javax.jms.TextMessage;
-import javax.jms.Topic;
-
-import org.jboss.jms.client.JBossConnection;
-import org.jboss.test.messaging.tools.jmx.rmi.Command;
-
-/**
- *
- * A CreateClientOnServerCommand.
- *
- * @author <a href="tim.fox at jboss.com">Tim Fox</a>
- * @author <a href="clebert.suconic at jboss.com">Clebert Suconic</a>
- * @version 1.1
- *
- * $Id$
- */
-public class CreateTwoClientOnServerCommand implements Command
-{
- private static final long serialVersionUID = -997724797145152821L;
-
- private ConnectionFactory cf;
-
- private boolean retainReference;
-
- private static List commands = new ArrayList();
-
- Connection conn1;
- Connection conn2;
-
- public CreateTwoClientOnServerCommand(ConnectionFactory cf, boolean retainReference)
- {
- this.cf = cf;
-
- this.retainReference = retainReference;
- }
-
- /*
- * Just create a connection, send and receive a message and leave the connection open.
- */
- public Object execute() throws Exception
- {
- if (retainReference)
- {
- commands.add(this);
- }
-
- conn1 = cf.createConnection();
- conn1.setClientID("test1");
- conn1.start();
-
- conn2 = cf.createConnection();
- conn2.setClientID("test2");
- conn2.start();
-
- conn1.close();
-
- String arrays[] = new String[2];
- arrays[0] = ((JBossConnection)conn1).getRemotingClientSessionId();
- arrays[1] = ((JBossConnection)conn2).getRemotingClientSessionId();
-
- //Return the remoting client session id for the connection
- return arrays;
- }
-
- }
-=======
-/*
- * 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;
-
-import java.util.ArrayList;
-import java.util.List;
-
-import javax.jms.Connection;
-import javax.jms.ConnectionFactory;
-import javax.jms.Topic;
-
-import org.jboss.jms.client.JBossConnection;
-import org.jboss.test.messaging.tools.jmx.rmi.Command;
-
-/**
- *
- * A CreateClientOnServerCommand.
- *
- * @author <a href="tim.fox at jboss.com">Tim Fox</a>
- * @author <a href="clebert.suconic at jboss.com">Clebert Suconic</a>
- * @version 1.1
- *
- * $Id$
- */
-public class CreateTwoClientOnServerCommand implements Command
-{
- private static final long serialVersionUID = -997724797145152821L;
-
- private ConnectionFactory cf;
-
- private boolean retainReference;
-
- private static List commands = new ArrayList();
-
- Topic topic;
- Connection conn1;
- Connection conn2;
-
- public CreateTwoClientOnServerCommand(ConnectionFactory cf, Topic topic, boolean retainReference)
- {
- this.cf = cf;
- this.topic = topic;
-
- this.retainReference = retainReference;
- }
-
- /*
- * Just create a connection, send and receive a message and leave the connection open.
- */
- public Object execute() throws Exception
- {
- if (retainReference)
- {
- commands.add(this);
- }
-
- conn1 = cf.createConnection();
- conn1.setClientID("test1");
- conn1.start();
-
- conn2 = cf.createConnection();
- conn2.setClientID("test2");
- conn2.start();
-
- conn1.close();
-
- String arrays[] = new String[2];
- arrays[0] = ((JBossConnection)conn1).getRemotingClientSessionId();
- arrays[1] = ((JBossConnection)conn2).getRemotingClientSessionId();
-
- // Return the remoting client session id for the connection
- return arrays;
- }
-
- }
->>>>>>> .merge-right.r1367
Modified: trunk/tests/src/org/jboss/test/messaging/jms/crash/ClientCrashTwoConnectionsTest.java
===================================================================
--- trunk/tests/src/org/jboss/test/messaging/jms/crash/ClientCrashTwoConnectionsTest.java 2006-09-27 07:55:32 UTC (rev 1368)
+++ trunk/tests/src/org/jboss/test/messaging/jms/crash/ClientCrashTwoConnectionsTest.java 2006-09-27 10:11:15 UTC (rev 1369)
@@ -1,4 +1,3 @@
-<<<<<<< .working
/*
* JBoss, Home of Professional Open Source
* Copyright 2005, JBoss Inc., and individual contributors as indicated
@@ -30,7 +29,6 @@
import org.jboss.jms.server.connectionmanager.SimpleConnectionManager;
import org.jboss.logging.Logger;
import org.jboss.test.messaging.MessagingTestCase;
-import org.jboss.test.messaging.jms.CreateTwoClientOnServerCommand;
import org.jboss.test.messaging.tools.ServerManagement;
import org.jboss.test.messaging.tools.jmx.ServiceContainer;
import org.jboss.test.messaging.tools.jmx.rmi.LocalTestServer;
@@ -42,156 +40,6 @@
* A ClientCrashTest.
*
* @author <a href="tim.fox at jboss.com">Tim Fox</a>
-* @author <a href="clebert.suconic at jboss.com">Clebert Suconic</a>
- * @version 1.1
- *
- * $Id$
- */
-public class ClientCrashTwoConnectionsTest extends MessagingTestCase
-{
- // Constants -----------------------------------------------------
-
- // Static --------------------------------------------------------
-
- private static final Logger log = Logger.getLogger(ClientCrashTwoConnectionsTest.class);
-
- // Attributes ----------------------------------------------------
-
- protected Server localServer;
-
- protected Server remoteServer;
-
- // Constructors --------------------------------------------------
-
- public ClientCrashTwoConnectionsTest(String name)
- {
- super(name);
- }
-
- // Public --------------------------------------------------------
-
- public void setUp() throws Exception
- {
- super.setUp();
-
- // Start the local server
- localServer = new LocalTestServer();
-
- // Start all the services locally
- localServer.start("all");
-
-
- // This crash test is relying on a precise value of LeaseInterval, so we don't rely on
- // the default, whatever that is ...
-
- localServer.setAttribute(ServiceContainer.REMOTING_OBJECT_NAME, "LeasePeriod", "3000");
-
- // Connect to the remote server, but don't start a servicecontainer on it. We are only using
- // the remote server to open a client connection to the local server.
- ServerManagement.create();
-
- remoteServer = ServerManagement.getServer();
-
- log.debug("setup done");
- }
-
- public void tearDown() throws Exception
- {
- localServer.stop();
- }
-
- /**
- * Test that when a remote jms client crashes, server side resources for connections are
- * cleaned-up.
- */
- public void testClientCrashWithTwoConnections() throws Exception
- {
- InitialContext ic = new InitialContext(InVMInitialContextFactory.getJNDIEnvironment());
- ConnectionFactory cf = (ConnectionFactory)ic.lookup("/ConnectionFactory");
-
- CreateTwoClientOnServerCommand command = new CreateTwoClientOnServerCommand(cf,true);
-
- String remotingSessionId[] = (String[])remoteServer.executeCommand(command);
-
- ConnectionManager cm = localServer.getServerPeer().getConnectionManager();
-
- log.info("server(0) = " + remotingSessionId[0]);
- log.info("server(1) = " + remotingSessionId[1]);
- log.info("we have = " + ((SimpleConnectionManager)cm).getClients().size() + " clients registered on SimpleconnectionManager");
-
- // Now we should have a client connection from the remote server to the local server
- remoteServer.exit();
- log.info("killed remote server");
-
- // Wait for connection resources to be cleared up
- Thread.sleep(25000);
-
- // See if we still have a connection with this id
-
- //Connection state shouldn't have been cleared up by now
- assertFalse(cm.containsSession(remotingSessionId[0]));
- assertFalse(cm.containsSession(remotingSessionId[1]));
-
- log.info("Servers = " + ((SimpleConnectionManager)cm).getClients().size());
-
- assertEquals(0,((SimpleConnectionManager)cm).getClients().size());
- }
-
-
- // Package protected ---------------------------------------------
-
- // Protected -----------------------------------------------------
-
- // Private -------------------------------------------------------
-
-
- // Inner classes -------------------------------------------------
-
-}
-=======
-/*
- * 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.crash;
-
-import javax.jms.ConnectionFactory;
-import javax.jms.Topic;
-import javax.naming.InitialContext;
-
-import org.jboss.jms.server.ConnectionManager;
-import org.jboss.jms.server.connectionmanager.SimpleConnectionManager;
-import org.jboss.logging.Logger;
-import org.jboss.test.messaging.MessagingTestCase;
-import org.jboss.test.messaging.jms.CreateTwoClientOnServerCommand;
-import org.jboss.test.messaging.tools.ServerManagement;
-import org.jboss.test.messaging.tools.jmx.ServiceContainer;
-import org.jboss.test.messaging.tools.jmx.rmi.LocalTestServer;
-import org.jboss.test.messaging.tools.jmx.rmi.Server;
-import org.jboss.test.messaging.tools.jndi.InVMInitialContextFactory;
-
-/**
- *
- * A ClientCrashTest.
- *
- * @author <a href="tim.fox at jboss.com">Tim Fox</a>
* @author <a href="clebert.suconic at jboss.com">Clebert Suconic</a>
* @version 1.1
*
@@ -303,4 +151,3 @@
// Inner classes -------------------------------------------------
}
->>>>>>> .merge-right.r1367
Copied: trunk/tests/src/org/jboss/test/messaging/jms/crash/CreateTwoClientOnServerCommand.java (from rev 1368, trunk/tests/src/org/jboss/test/messaging/jms/CreateTwoClientOnServerCommand.java)
===================================================================
--- trunk/tests/src/org/jboss/test/messaging/jms/CreateTwoClientOnServerCommand.java 2006-09-27 07:55:32 UTC (rev 1368)
+++ trunk/tests/src/org/jboss/test/messaging/jms/crash/CreateTwoClientOnServerCommand.java 2006-09-27 10:11:15 UTC (rev 1369)
@@ -0,0 +1,94 @@
+/*
+ * 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.crash;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import javax.jms.Connection;
+import javax.jms.ConnectionFactory;
+import javax.jms.Topic;
+
+import org.jboss.jms.client.JBossConnection;
+import org.jboss.test.messaging.tools.jmx.rmi.Command;
+
+/**
+ *
+ * A CreateClientOnServerCommand.
+ *
+ * @author <a href="tim.fox at jboss.com">Tim Fox</a>
+ * @author <a href="clebert.suconic at jboss.com">Clebert Suconic</a>
+ * @version 1.1
+ *
+ * $Id$
+ */
+public class CreateTwoClientOnServerCommand implements Command
+{
+ private static final long serialVersionUID = -997724797145152821L;
+
+ private ConnectionFactory cf;
+
+ private boolean retainReference;
+
+ private static List commands = new ArrayList();
+
+ Topic topic;
+ Connection conn1;
+ Connection conn2;
+
+ public CreateTwoClientOnServerCommand(ConnectionFactory cf, Topic topic, boolean retainReference)
+ {
+ this.cf = cf;
+ this.topic = topic;
+
+ this.retainReference = retainReference;
+ }
+
+ /*
+ * Just create a connection, send and receive a message and leave the connection open.
+ */
+ public Object execute() throws Exception
+ {
+ if (retainReference)
+ {
+ commands.add(this);
+ }
+
+ conn1 = cf.createConnection();
+ conn1.setClientID("test1");
+ conn1.start();
+
+ conn2 = cf.createConnection();
+ conn2.setClientID("test2");
+ conn2.start();
+
+ conn1.close();
+
+ String arrays[] = new String[2];
+ arrays[0] = ((JBossConnection)conn1).getRemotingClientSessionId();
+ arrays[1] = ((JBossConnection)conn2).getRemotingClientSessionId();
+
+ // Return the remoting client session id for the connection
+ return arrays;
+ }
+
+}
More information about the jboss-cvs-commits
mailing list