You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by gr...@apache.org on 2010/09/20 13:05:16 UTC
svn commit: r998887 - in /qpid/branches/0.5.x-dev/qpid/java:
broker/src/main/java/org/apache/qpid/server/
broker/src/main/java/org/apache/qpid/server/configuration/
broker/src/main/java/org/apache/qpid/server/connection/
broker/src/main/java/org/apache...
Author: grkvlt
Date: Mon Sep 20 11:05:15 2010
New Revision: 998887
URL: http://svn.apache.org/viewvc?rev=998887&view=rev
Log:
QPID-2864: Add producer configurable transaction timeouts
Adds a configurable open or idle transaction timeout, which either
warns (using operational logging) or closes the connection, depending
on configuration. Default behaviour is no action.
Added:
qpid/branches/0.5.x-dev/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/transacted/TransactionTimeoutConfigurationTest.java
qpid/branches/0.5.x-dev/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/transacted/TransactionTimeoutDisabledTest.java
qpid/branches/0.5.x-dev/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/transacted/TransactionTimeoutTest.java
qpid/branches/0.5.x-dev/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/transacted/TransactionTimeoutTestCase.java
Modified:
qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java
qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/VirtualHostConfiguration.java
qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/connection/ConnectionRegistry.java
qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/connection/IConnectionRegistry.java
qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/messages/LogMessages.properties
qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/subjects/ChannelLogSubject.java
qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQMinaProtocolSession.java
qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSession.java
qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java
qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/txn/LocalTransactionalContext.java
qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/txn/NonTransactionalContext.java
qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/txn/TransactionalContext.java
qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java
Modified: qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java
URL: http://svn.apache.org/viewvc/qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java?rev=998887&r1=998886&r2=998887&view=diff
==============================================================================
--- qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java (original)
+++ qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java Mon Sep 20 11:05:15 2010
@@ -31,9 +31,11 @@ import java.util.concurrent.ConcurrentMa
import java.util.concurrent.ConcurrentHashMap;
import org.apache.log4j.Logger;
+import org.apache.qpid.AMQConnectionException;
import org.apache.qpid.AMQException;
import org.apache.qpid.framing.*;
import org.apache.qpid.framing.abstraction.MessagePublishInfo;
+import org.apache.qpid.protocol.AMQConstant;
import org.apache.qpid.server.ack.UnacknowledgedMessageMap;
import org.apache.qpid.server.ack.UnacknowledgedMessageMapImpl;
import org.apache.qpid.server.exchange.Exchange;
@@ -120,6 +122,7 @@ public class AMQChannel
private LogActor _actor;
private LogSubject _logSubject;
+ private long _txnUpdateTime;
public AMQChannel(AMQProtocolSession session, int channelId, MessageStore messageStore)
throws AMQException
@@ -166,6 +169,8 @@ public class AMQChannel
_currentMessage = new IncomingMessage(_messageStore.getNewMessageId(), info, _txnContext, _session);
_currentMessage.setMessageStore(_messageStore);
_currentMessage.setExchange(e);
+
+ updateTransactionalActivity();
}
public void publishContentHeader(ContentHeaderBody contentHeaderBody)
@@ -770,6 +775,15 @@ public class AMQChannel
public void acknowledgeMessage(long deliveryTag, boolean multiple) throws AMQException
{
_unacknowledgedMessageMap.acknowledgeMessage(deliveryTag, multiple, _txnContext);
+ updateTransactionalActivity();
+ }
+
+ public void updateTransactionalActivity()
+ {
+ if (isTransactional())
+ {
+ _txnUpdateTime = System.currentTimeMillis();
+ }
}
/**
@@ -1016,4 +1030,57 @@ public class AMQChannel
{
return _blocking.get();
}
+
+ /**
+ * This method is called from the housekeeping thread to check the status of
+ * transactions on this channel and react appropriately.
+ *
+ * If a transaction is open for too long or idle for too long then a warning
+ * is logged or the connection is closed, depending on the configuration. An open
+ * transaction is one that has recent activity. The transaction age is counted
+ * from the time the transaction was started. An idle transaction is one that
+ * has had no activity, such as publishing or acknowledgeing messages.
+ *
+ * @param openWarn time in milliseconds before alerting on open transaction
+ * @param openClose time in milliseconds before closing connection with open transaction
+ * @param idleWarn time in milliseconds before alerting on idle transaction
+ * @param idleClose time in milliseconds before closing connection with idle transaction
+ */
+ public void checkTransactionStatus(long openWarn, long openClose, long idleWarn, long idleClose) throws AMQException
+ {
+ if (isTransactional() && _txnContext.inTransaction())
+ {
+ long currentTime = System.currentTimeMillis();
+ long openTime = currentTime - _txnContext.getTransactionStartTime();
+ long idleTime = currentTime - _txnUpdateTime;
+
+ // Log a warning on idle or open transactions
+ if (idleWarn > 0L && idleTime > idleWarn)
+ {
+ _actor.message(ChannelMessages.CHN_IDLE_TXN(idleTime));
+ }
+ else if (openWarn > 0L && openTime > openWarn)
+ {
+ _actor.message(ChannelMessages.CHN_OPEN_TXN(openTime));
+ }
+
+ // Close connection for idle or open transactions that have timed out
+ if (idleClose > 0L && idleTime > idleClose)
+ {
+ _session.closeConnection(_channelId, new AMQConnectionException(AMQConstant.REQUEST_TIMEOUT,
+ "Idle transaction timed out", 0, 0,
+ _session.getProtocolOutputConverter().getProtocolMajorVersion(),
+ _session.getProtocolOutputConverter().getProtocolMinorVersion(),
+ (Throwable) null), true);
+ }
+ else if (openClose > 0L && openTime > openClose)
+ {
+ _session.closeConnection(_channelId, new AMQConnectionException(AMQConstant.REQUEST_TIMEOUT,
+ "Open transaction timed out", 0, 0,
+ _session.getProtocolOutputConverter().getProtocolMajorVersion(),
+ _session.getProtocolOutputConverter().getProtocolMinorVersion(),
+ (Throwable) null), true);
+ }
+ }
+ }
}
Modified: qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/VirtualHostConfiguration.java
URL: http://svn.apache.org/viewvc/qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/VirtualHostConfiguration.java?rev=998887&r1=998886&r2=998887&view=diff
==============================================================================
--- qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/VirtualHostConfiguration.java (original)
+++ qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/VirtualHostConfiguration.java Mon Sep 20 11:05:15 2010
@@ -176,4 +176,23 @@ public class VirtualHostConfiguration
return _config.getLong("queues.flowResumeCapacity", getCapacity());
}
+ public long getTransactionTimeoutOpenWarn()
+ {
+ return _config.getLong("transactionTimeout.openWarn", 0L);
+ }
+
+ public long getTransactionTimeoutOpenClose()
+ {
+ return _config.getLong("transactionTimeout.openClose", 0L);
+ }
+
+ public long getTransactionTimeoutIdleWarn()
+ {
+ return _config.getLong("transactionTimeout.idleWarn", 0L);
+ }
+
+ public long getTransactionTimeoutIdleClose()
+ {
+ return _config.getLong("transactionTimeout.idleClose", 0L);
+ }
}
Modified: qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/connection/ConnectionRegistry.java
URL: http://svn.apache.org/viewvc/qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/connection/ConnectionRegistry.java?rev=998887&r1=998886&r2=998887&view=diff
==============================================================================
--- qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/connection/ConnectionRegistry.java (original)
+++ qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/connection/ConnectionRegistry.java Mon Sep 20 11:05:15 2010
@@ -28,6 +28,8 @@ import org.apache.qpid.AMQConnectionExce
import org.apache.qpid.protocol.AMQConstant;
import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.ArrayList;
+import java.util.Collection;
import java.util.List;
public class ConnectionRegistry implements IConnectionRegistry
@@ -70,4 +72,9 @@ public class ConnectionRegistry implemen
{
_registry.remove(connnection);
}
+
+ public Collection<AMQProtocolSession> getConnections()
+ {
+ return new ArrayList<AMQProtocolSession>(_registry);
+ }
}
Modified: qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/connection/IConnectionRegistry.java
URL: http://svn.apache.org/viewvc/qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/connection/IConnectionRegistry.java?rev=998887&r1=998886&r2=998887&view=diff
==============================================================================
--- qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/connection/IConnectionRegistry.java (original)
+++ qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/connection/IConnectionRegistry.java Mon Sep 20 11:05:15 2010
@@ -20,13 +20,13 @@
*/
package org.apache.qpid.server.connection;
-import org.apache.qpid.server.protocol.AMQMinaProtocolSession;
-import org.apache.qpid.server.protocol.AMQProtocolSession;
+import java.util.Collection;
+
import org.apache.qpid.AMQException;
+import org.apache.qpid.server.protocol.AMQProtocolSession;
public interface IConnectionRegistry
{
-
public void initialise();
public void close() throws AMQException;
@@ -35,4 +35,5 @@ public interface IConnectionRegistry
public void deregisterConnection(AMQProtocolSession connnection);
+ public Collection<AMQProtocolSession> getConnections();
}
Modified: qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/messages/LogMessages.properties
URL: http://svn.apache.org/viewvc/qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/messages/LogMessages.properties?rev=998887&r1=998886&r2=998887&view=diff
==============================================================================
--- qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/messages/LogMessages.properties (original)
+++ qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/messages/LogMessages.properties Mon Sep 20 11:05:15 2010
@@ -274,6 +274,9 @@ CHN_CLOSE = CHN-1003 : Close
CHN_PREFETCH_SIZE = CHN-1004 : Prefetch Size (bytes) {0,number} : Count {1,number}
CHN_FLOW_ENFORCED = CHN-1005 : Flow Control Enforced (Queue {0})
CHN_FLOW_REMOVED = CHN-1006 : Flow Control Removed
+#Channel Transactions
+CHN_OPEN_TXN = CHN-1007 : Open Transaction : {0,number} ms
+CHN_IDLE_TXN = CHN-1008 : Idle Transaction : {0,number} ms
#Queue
# 0 - owner
Modified: qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/subjects/ChannelLogSubject.java
URL: http://svn.apache.org/viewvc/qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/subjects/ChannelLogSubject.java?rev=998887&r1=998886&r2=998887&view=diff
==============================================================================
--- qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/subjects/ChannelLogSubject.java (original)
+++ qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/subjects/ChannelLogSubject.java Mon Sep 20 11:05:15 2010
@@ -37,7 +37,7 @@ public class ChannelLogSubject extends A
* 4 - Channel ID
*/
public static String CHANNEL_FORMAT = ConnectionLogSubject.CONNECTION_FORMAT
- + "/ch:{4}";
+ + "/ch:{4}";
public ChannelLogSubject(AMQChannel channel)
{
Modified: qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQMinaProtocolSession.java
URL: http://svn.apache.org/viewvc/qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQMinaProtocolSession.java?rev=998887&r1=998886&r2=998887&view=diff
==============================================================================
--- qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQMinaProtocolSession.java (original)
+++ qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQMinaProtocolSession.java Mon Sep 20 11:05:15 2010
@@ -20,6 +20,21 @@
*/
package org.apache.qpid.server.protocol;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.security.Principal;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CopyOnWriteArraySet;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+
+import javax.management.JMException;
+import javax.security.sasl.SaslServer;
+
import org.apache.log4j.Logger;
import org.apache.mina.common.CloseFuture;
import org.apache.mina.common.IdleStatus;
@@ -52,12 +67,12 @@ import org.apache.qpid.protocol.AMQMetho
import org.apache.qpid.protocol.AMQMethodListener;
import org.apache.qpid.server.AMQChannel;
import org.apache.qpid.server.handler.ServerMethodDispatcherImpl;
+import org.apache.qpid.server.logging.LogActor;
+import org.apache.qpid.server.logging.LogSubject;
import org.apache.qpid.server.logging.actors.AMQPConnectionActor;
import org.apache.qpid.server.logging.actors.CurrentActor;
-import org.apache.qpid.server.logging.subjects.ConnectionLogSubject;
-import org.apache.qpid.server.logging.LogSubject;
-import org.apache.qpid.server.logging.LogActor;
import org.apache.qpid.server.logging.messages.ConnectionMessages;
+import org.apache.qpid.server.logging.subjects.ConnectionLogSubject;
import org.apache.qpid.server.management.Managable;
import org.apache.qpid.server.management.ManagedObject;
import org.apache.qpid.server.output.ProtocolOutputConverter;
@@ -69,20 +84,6 @@ import org.apache.qpid.server.virtualhos
import org.apache.qpid.server.virtualhost.VirtualHostRegistry;
import org.apache.qpid.transport.Sender;
-import javax.management.JMException;
-import javax.security.sasl.SaslServer;
-import java.net.InetSocketAddress;
-import java.net.SocketAddress;
-import java.security.Principal;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.CopyOnWriteArrayList;
-import java.util.concurrent.CopyOnWriteArraySet;
-import java.util.concurrent.atomic.AtomicLong;
-import java.util.concurrent.atomic.AtomicBoolean;
-
public class AMQMinaProtocolSession implements AMQProtocolSession, Managable
{
private static final Logger _logger = Logger.getLogger(AMQProtocolSession.class);
@@ -103,7 +104,7 @@ public class AMQMinaProtocolSession impl
private VirtualHost _virtualHost;
- private final Map<Integer, AMQChannel> _channelMap = new HashMap<Integer, AMQChannel>();
+ private final Map<Integer, AMQChannel> _channelMap = new ConcurrentHashMap<Integer, AMQChannel>();
private final AMQChannel[] _cachedChannels = new AMQChannel[CHANNEL_CACHE_SIZE + 1];
Modified: qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSession.java
URL: http://svn.apache.org/viewvc/qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSession.java?rev=998887&r1=998886&r2=998887&view=diff
==============================================================================
--- qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSession.java (original)
+++ qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSession.java Mon Sep 20 11:05:15 2010
@@ -34,6 +34,7 @@ import org.apache.qpid.server.output.Pro
import org.apache.qpid.server.virtualhost.VirtualHost;
import java.security.Principal;
+import java.util.Collection;
public interface AMQProtocolSession extends AMQVersionAwareProtocolSession
@@ -113,6 +114,13 @@ public interface AMQProtocolSession exte
* than one session but this is not validated.
*/
void addChannel(AMQChannel channel) throws AMQException;
+
+ /**
+ * Return all channels associated with this session.
+ *
+ * @return a collection containing this sessions's channels
+ */
+ Collection<AMQChannel> getChannels();
/**
* Close a specific channel. This will remove any resources used by the channel, including: <ul><li>any queue
Modified: qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java
URL: http://svn.apache.org/viewvc/qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java?rev=998887&r1=998886&r2=998887&view=diff
==============================================================================
--- qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java (original)
+++ qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java Mon Sep 20 11:05:15 2010
@@ -510,6 +510,7 @@ public class SimpleAMQQueue implements A
throws AMQException
{
_deliveredMessages.incrementAndGet();
+
if (_logger.isDebugEnabled())
{
_logger.debug(sub + ": deliverMessage: " + entry.debugIdentity());
Modified: qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/txn/LocalTransactionalContext.java
URL: http://svn.apache.org/viewvc/qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/txn/LocalTransactionalContext.java?rev=998887&r1=998886&r2=998887&view=diff
==============================================================================
--- qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/txn/LocalTransactionalContext.java (original)
+++ qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/txn/LocalTransactionalContext.java Mon Sep 20 11:05:15 2010
@@ -56,6 +56,7 @@ public class LocalTransactionalContext i
private boolean _messageDelivered = false;
private final AMQChannel _channel;
+ private long _txnStartTime;
private abstract class DeliveryAction
{
@@ -114,6 +115,7 @@ public class LocalTransactionalContext i
public LocalTransactionalContext(final AMQChannel channel)
{
_channel = channel;
+ _txnStartTime = System.currentTimeMillis();
}
public StoreContext getStoreContext()
@@ -135,6 +137,7 @@ public class LocalTransactionalContext i
public void rollback() throws AMQException
{
_txnBuffer.rollback(getStoreContext());
+
// Hack to deal with uncommitted non-transactional writes
if (getMessageStore().inTran(getStoreContext()))
{
@@ -142,6 +145,7 @@ public class LocalTransactionalContext i
_inTran = false;
}
+ _txnStartTime = System.currentTimeMillis();
_postCommitDeliveryList.clear();
}
@@ -215,6 +219,17 @@ public class LocalTransactionalContext i
{
// Not required in this transactional context
}
+
+ public boolean inTransaction()
+ {
+ return _inTran;
+ }
+
+ public long getTransactionStartTime()
+ {
+ return _txnStartTime;
+ }
+
public void beginTranIfNecessary() throws AMQException
{
@@ -226,6 +241,7 @@ public class LocalTransactionalContext i
}
getMessageStore().beginTran(getStoreContext());
+ _txnStartTime = System.currentTimeMillis();
_inTran = true;
}
}
@@ -259,6 +275,7 @@ public class LocalTransactionalContext i
{
_messageDelivered = false;
_inTran = getMessageStore().inTran(getStoreContext());
+ _txnStartTime = System.currentTimeMillis();
}
try
Modified: qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/txn/NonTransactionalContext.java
URL: http://svn.apache.org/viewvc/qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/txn/NonTransactionalContext.java?rev=998887&r1=998886&r2=998887&view=diff
==============================================================================
--- qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/txn/NonTransactionalContext.java (original)
+++ qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/txn/NonTransactionalContext.java Mon Sep 20 11:05:15 2010
@@ -77,6 +77,16 @@ public class NonTransactionalContext imp
_inTran = true;
}
}
+
+ public boolean inTransaction()
+ {
+ return false;
+ }
+
+ public long getTransactionStartTime()
+ {
+ return 0L;
+ }
public void commit() throws AMQException
{
Modified: qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/txn/TransactionalContext.java
URL: http://svn.apache.org/viewvc/qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/txn/TransactionalContext.java?rev=998887&r1=998886&r2=998887&view=diff
==============================================================================
--- qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/txn/TransactionalContext.java (original)
+++ qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/txn/TransactionalContext.java Mon Sep 20 11:05:15 2010
@@ -91,6 +91,20 @@ public interface TransactionalContext
* @throws AMQException If the transaction cannot be started for any reason.
*/
void beginTranIfNecessary() throws AMQException;
+
+ /**
+ * Returns whether or not a transaction is currently in progress.
+ *
+ * @return true if a transaction is currently in progress
+ */
+ boolean inTransaction();
+
+ /**
+ * Return the time the current transaction started.
+ *
+ * @return the time this transaction started or 0 if not in a transaction
+ */
+ long getTransactionStartTime();
/**
* Makes all pending operations on the transaction permanent and visible.
@@ -118,13 +132,13 @@ public interface TransactionalContext
void deliver(final AMQQueue queue, AMQMessage message) throws AMQException;
/**
- * Requeues the specified message entry (message queue pair)
- *
- *
- * @param queueEntry The message,queue pair
- *
- * @throws AMQException If the message cannot be delivered for any reason.
- */
+ * Requeues the specified message entry (message queue pair)
+ *
+ *
+ * @param queueEntry The message,queue pair
+ *
+ * @throws AMQException If the message cannot be delivered for any reason.
+ */
void requeue(QueueEntry queueEntry) throws AMQException;
Modified: qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java
URL: http://svn.apache.org/viewvc/qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java?rev=998887&r1=998886&r2=998887&view=diff
==============================================================================
--- qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java (original)
+++ qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java Mon Sep 20 11:05:15 2010
@@ -27,6 +27,8 @@ import org.apache.qpid.framing.AMQShortS
import org.apache.qpid.framing.FieldTable;
import org.apache.qpid.framing.abstraction.ContentChunk;
import org.apache.qpid.server.AMQBrokerManagerMBean;
+import org.apache.qpid.server.AMQChannel;
+import org.apache.qpid.server.logging.actors.AbstractActor;
import org.apache.qpid.server.logging.actors.CurrentActor;
import org.apache.qpid.server.logging.messages.VirtualHostMessages;
import org.apache.qpid.server.configuration.ExchangeConfiguration;
@@ -41,6 +43,7 @@ import org.apache.qpid.server.exchange.E
import org.apache.qpid.server.exchange.ExchangeRegistry;
import org.apache.qpid.server.management.AMQManagedObject;
import org.apache.qpid.server.management.ManagedObject;
+import org.apache.qpid.server.protocol.AMQProtocolSession;
import org.apache.qpid.server.queue.AMQQueue;
import org.apache.qpid.server.queue.AMQQueueFactory;
import org.apache.qpid.server.queue.DefaultQueueRegistry;
@@ -230,7 +233,7 @@ public class VirtualHost implements Acce
private void initialiseHouseKeeping(long period)
{
/* add a timer task to iterate over queues, cleaning expired messages from queues with no consumers */
- if (period != 0L)
+ if (period > 0L)
{
class HouseKeepingTask extends TimerTask
{
@@ -238,6 +241,12 @@ public class VirtualHost implements Acce
public void run()
{
+ CurrentActor.set(new AbstractActor(ApplicationRegistry.getInstance().getRootMessageLogger()) {
+ public String getLogMessage()
+ {
+ return "[" + Thread.currentThread().getName() + "]";
+ }
+ });
_hkLogger.info("Starting the houseKeeping job");
for (AMQQueue q : _queueRegistry.getQueues())
{
@@ -253,7 +262,27 @@ public class VirtualHost implements Acce
// house keeping task from running.
}
}
+ for (AMQProtocolSession conn : _connectionRegistry.getConnections())
+ {
+ _hkLogger.debug("Checking for long running open transactions on connection " + conn);
+ for (AMQChannel ch : conn.getChannels())
+ {
+ try
+ {
+ ch.checkTransactionStatus(_configuration.getTransactionTimeoutOpenWarn(),
+ _configuration.getTransactionTimeoutOpenClose(),
+ _configuration.getTransactionTimeoutIdleWarn(),
+ _configuration.getTransactionTimeoutIdleClose());
+ }
+ catch (Exception e)
+ {
+ _hkLogger.error("Exception in housekeeping for connection: " + conn.toString(), e);
+ }
+ }
+ }
+
_hkLogger.info("HouseKeeping job completed.");
+ CurrentActor.remove();
}
}
Added: qpid/branches/0.5.x-dev/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/transacted/TransactionTimeoutConfigurationTest.java
URL: http://svn.apache.org/viewvc/qpid/branches/0.5.x-dev/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/transacted/TransactionTimeoutConfigurationTest.java?rev=998887&view=auto
==============================================================================
--- qpid/branches/0.5.x-dev/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/transacted/TransactionTimeoutConfigurationTest.java (added)
+++ qpid/branches/0.5.x-dev/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/transacted/TransactionTimeoutConfigurationTest.java Mon Sep 20 11:05:15 2010
@@ -0,0 +1,86 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.test.unit.transacted;
+
+/**
+ * This verifies that changing the {@code transactionTimeout} configuration will alter
+ * the behaviour of the transaction open and idle logging, and that when the connection
+ * will be closed.
+ */
+public class TransactionTimeoutConfigurationTest extends TransactionTimeoutTestCase
+{
+ @Override
+ protected void configure() throws Exception
+ {
+ // Setup housekeeping every second
+ setConfigurationProperty("virtualhosts.virtualhost." + VIRTUALHOST + ".housekeeping.expiredMessageCheckPeriod", "1000");
+
+ // Set transaction timout properties.
+ setConfigurationProperty("virtualhosts.virtualhost." + VIRTUALHOST + ".transactionTimeout.openWarn", "2000");
+ setConfigurationProperty("virtualhosts.virtualhost." + VIRTUALHOST + ".transactionTimeout.openClose", "10000");
+ setConfigurationProperty("virtualhosts.virtualhost." + VIRTUALHOST + ".transactionTimeout.idleWarn", "1000");
+ setConfigurationProperty("virtualhosts.virtualhost." + VIRTUALHOST + ".transactionTimeout.idleClose", "5000");
+ }
+
+ public void testProducerIdleCommit() throws Exception
+ {
+ try
+ {
+ producer();
+
+ send(5, 0);
+
+ sleep(20);
+
+ _psession.commit();
+ fail("should fail");
+ }
+ catch (Exception e)
+ {
+ _exception = e;
+ }
+
+ monitor(5, 0);
+
+ check(IDLE);
+ }
+
+ public void testProducerOpenCommit() throws Exception
+ {
+ try
+ {
+ producer();
+
+ send(5, 3);
+
+ _psession.commit();
+ fail("should fail");
+ }
+ catch (Exception e)
+ {
+ _exception = e;
+ }
+
+ monitor(6, 3);
+
+ check(OPEN);
+ }
+}
Added: qpid/branches/0.5.x-dev/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/transacted/TransactionTimeoutDisabledTest.java
URL: http://svn.apache.org/viewvc/qpid/branches/0.5.x-dev/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/transacted/TransactionTimeoutDisabledTest.java?rev=998887&view=auto
==============================================================================
--- qpid/branches/0.5.x-dev/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/transacted/TransactionTimeoutDisabledTest.java (added)
+++ qpid/branches/0.5.x-dev/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/transacted/TransactionTimeoutDisabledTest.java Mon Sep 20 11:05:15 2010
@@ -0,0 +1,76 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.test.unit.transacted;
+
+/**
+ * This verifies that the default behaviour is not to time out transactions.
+ */
+public class TransactionTimeoutDisabledTest extends TransactionTimeoutTestCase
+{
+ @Override
+ protected void configure() throws Exception
+ {
+ // Setup housekeeping every second
+ setConfigurationProperty("virtualhosts.virtualhost." + VIRTUALHOST + ".housekeeping.expiredMessageCheckPeriod", "1000");
+ }
+
+ public void testProducerIdleCommit() throws Exception
+ {
+ try
+ {
+ producer();
+
+ send(5, 0);
+
+ sleep(20);
+
+ _psession.commit();
+ }
+ catch (Exception e)
+ {
+ fail("Should have succeeded");
+ }
+
+ assertTrue("Listener should not have received exception", _caught.getCount() == 1);
+
+ monitor(0, 0);
+ }
+
+ public void testProducerOpenCommit() throws Exception
+ {
+ try
+ {
+ producer();
+
+ send(5, 3);
+
+ _psession.commit();
+ }
+ catch (Exception e)
+ {
+ fail("Should have succeeded");
+ }
+
+ assertTrue("Listener should not have received exception", _caught.getCount() == 1);
+
+ monitor(0, 0);
+ }
+}
Added: qpid/branches/0.5.x-dev/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/transacted/TransactionTimeoutTest.java
URL: http://svn.apache.org/viewvc/qpid/branches/0.5.x-dev/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/transacted/TransactionTimeoutTest.java?rev=998887&view=auto
==============================================================================
--- qpid/branches/0.5.x-dev/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/transacted/TransactionTimeoutTest.java (added)
+++ qpid/branches/0.5.x-dev/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/transacted/TransactionTimeoutTest.java Mon Sep 20 11:05:15 2010
@@ -0,0 +1,340 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.test.unit.transacted;
+
+/**
+ * This tests the behaviour of transactional sessions when the {@code transactionTimeout} configuration
+ * is set for a virtual host.
+ *
+ * A producer that is idle for too long or open for too long will have its connection closed and
+ * any further operations will fail with a 408 resource timeout exception. Consumers will not
+ * be affected by the transaction timeout configuration.
+ */
+public class TransactionTimeoutTest extends TransactionTimeoutTestCase
+{
+ public void testProducerIdleCommit() throws Exception
+ {
+ try
+ {
+ producer();
+
+ send(5, 0);
+
+ sleep(20);
+
+ _psession.commit();
+ fail("should fail");
+ }
+ catch (Exception e)
+ {
+ _exception = e;
+ }
+
+ monitor(10, 0);
+
+ check(IDLE);
+ }
+
+ public void testProducerOpenCommit() throws Exception
+ {
+ try
+ {
+ producer();
+
+ send(6, 5);
+
+ _psession.commit();
+ fail("should fail");
+ }
+ catch (Exception e)
+ {
+ _exception = e;
+ }
+
+ monitor(0, 10);
+
+ check(OPEN);
+ }
+
+ public void testProducerIdleCommitTwice() throws Exception
+ {
+ try
+ {
+ producer();
+
+ send(5, 0);
+
+ sleep(10);
+
+ _psession.commit();
+
+ send(5, 0);
+
+ sleep(20);
+
+ _psession.commit();
+ fail("should fail");
+ }
+ catch (Exception e)
+ {
+ _exception = e;
+ }
+
+ monitor(15, 0);
+
+ check(IDLE);
+ }
+
+ public void testProducerOpenCommitTwice() throws Exception
+ {
+ try
+ {
+ producer();
+
+ send(5, 0);
+
+ sleep(10);
+
+ _psession.commit();
+
+ send(6, 5);
+
+ _psession.commit();
+ fail("should fail");
+ }
+ catch (Exception e)
+ {
+ _exception = e;
+ }
+
+ monitor(5, 10);
+
+ check(OPEN);
+ }
+
+ public void testProducerIdleRollback() throws Exception
+ {
+ try
+ {
+ producer();
+
+ send(5, 0);
+
+ sleep(20);
+
+ _psession.rollback();
+ fail("should fail");
+ }
+ catch (Exception e)
+ {
+ _exception = e;
+ }
+
+ monitor(10, 0);
+
+ check(IDLE);
+ }
+
+ public void testProducerIdleRollbackTwice() throws Exception
+ {
+ try
+ {
+ producer();
+
+ send(5, 0);
+
+ sleep(10);
+
+ _psession.rollback();
+
+ send(5, 0);
+
+ sleep(20);
+
+ _psession.rollback();
+ fail("should fail");
+ }
+ catch (Exception e)
+ {
+ _exception = e;
+ }
+
+ monitor(15, 0);
+
+ check(IDLE);
+ }
+
+ public void testConsumerCommitClose() throws Exception
+ {
+ try
+ {
+ producer();
+
+ consumer();
+
+ send(1, 0);
+
+ _psession.commit();
+
+ expect(1, 0);
+
+ _csession.commit();
+
+ sleep(30);
+
+ _csession.close();
+ }
+ catch (Exception e)
+ {
+ fail("should have succeeded: " + e.getMessage());
+ }
+
+ monitor(0, 0);
+ }
+
+ public void testConsumerIdleReceiveCommit() throws Exception
+ {
+ try
+ {
+ producer();
+
+ consumer();
+
+ send(1, 0);
+
+ _psession.commit();
+
+ sleep(20);
+
+ expect(1, 0);
+
+ sleep(20);
+
+ _csession.commit();
+ }
+ catch (Exception e)
+ {
+ fail("Should have succeeded");
+ }
+
+ monitor(0, 0);
+ }
+
+ public void testConsumerIdleCommit() throws Exception
+ {
+ try
+ {
+ producer();
+
+ consumer();
+
+ send(1, 0);
+
+ _psession.commit();
+
+ expect(1, 0);
+
+ sleep(20);
+
+ _csession.commit();
+ }
+ catch (Exception e)
+ {
+ fail("Should have succeeded");
+ }
+
+ monitor(0, 0);
+ }
+
+ public void testConsumerIdleRollback() throws Exception
+ {
+ try
+ {
+ producer();
+
+ consumer();
+
+ send(1, 0);
+
+ _psession.commit();
+
+ expect(1, 0);
+
+ sleep(20);
+
+ _csession.rollback();
+ }
+ catch (Exception e)
+ {
+ fail("Should have succeeded");
+ }
+
+ monitor(0, 0);
+ }
+
+ public void testConsumerOpenCommit() throws Exception
+ {
+ try
+ {
+ producer();
+
+ consumer();
+
+ send(1, 0);
+
+ _psession.commit();
+
+ sleep(30);
+
+ _csession.commit();
+ }
+ catch (Exception e)
+ {
+ fail("Should have succeeded");
+ }
+
+ monitor(0, 0);
+ }
+
+ public void testConsumerOpenRollback() throws Exception
+ {
+ try
+ {
+ producer();
+
+ consumer();
+
+ send(1, 0);
+
+ _psession.commit();
+
+ sleep(30);
+
+ _csession.rollback();
+ }
+ catch (Exception e)
+ {
+ fail("Should have succeeded");
+ }
+
+ monitor(0, 0);
+ }
+}
Added: qpid/branches/0.5.x-dev/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/transacted/TransactionTimeoutTestCase.java
URL: http://svn.apache.org/viewvc/qpid/branches/0.5.x-dev/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/transacted/TransactionTimeoutTestCase.java?rev=998887&view=auto
==============================================================================
--- qpid/branches/0.5.x-dev/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/transacted/TransactionTimeoutTestCase.java (added)
+++ qpid/branches/0.5.x-dev/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/transacted/TransactionTimeoutTestCase.java Mon Sep 20 11:05:15 2010
@@ -0,0 +1,253 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.test.unit.transacted;
+
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import javax.jms.DeliveryMode;
+import javax.jms.ExceptionListener;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.Queue;
+import javax.jms.TextMessage;
+
+import junit.framework.TestCase;
+
+import org.apache.qpid.AMQException;
+import org.apache.qpid.client.AMQConnection;
+import org.apache.qpid.client.AMQConnectionURL;
+import org.apache.qpid.client.AMQQueue;
+import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.jms.ConnectionURL;
+import org.apache.qpid.jms.Session;
+import org.apache.qpid.protocol.AMQConstant;
+import org.apache.qpid.test.utils.QpidTestCase;
+import org.apache.qpid.util.LogMonitor;
+
+/**
+ * The {@link TestCase} for transaction timeout testing.
+ */
+public class TransactionTimeoutTestCase extends QpidTestCase implements ExceptionListener
+{
+ public static final String VIRTUALHOST = "test";
+ public static final String TEXT = "0123456789abcdefghiforgettherest";
+ public static final String CHN_OPEN_TXN = "CHN-1007";
+ public static final String CHN_IDLE_TXN = "CHN-1008";
+ public static final String IDLE = "Idle";
+ public static final String OPEN = "Open";
+
+ protected LogMonitor _monitor;
+ protected AMQConnection _con;
+ protected Session _psession, _csession;
+ protected Queue _queue;
+ protected MessageConsumer _consumer;
+ protected MessageProducer _producer;
+ protected CountDownLatch _caught = new CountDownLatch(1);
+ protected String _message;
+ protected Exception _exception;
+ protected AMQConstant _code;
+
+ protected void configure() throws Exception
+ {
+ // Setup housekeeping every second
+ setConfigurationProperty("virtualhosts.virtualhost." + VIRTUALHOST + ".housekeeping.expiredMessageCheckPeriod", "1000");
+
+ /*
+ * Set transaction timout properties. The XML in the virtualhosts configuration is as follows:
+ *
+ * <transactionTimeout>
+ * <openWarn>10000</openWarn>
+ * <openClose>20000</openClose>
+ * <idleWarn>5000</idleWarn>
+ * <idleClose>15000</idleClose>
+ * </transactionTimeout>
+ */
+ setConfigurationProperty("virtualhosts.virtualhost." + VIRTUALHOST + ".transactionTimeout.openWarn", "10000");
+ setConfigurationProperty("virtualhosts.virtualhost." + VIRTUALHOST + ".transactionTimeout.openClose", "20000");
+ setConfigurationProperty("virtualhosts.virtualhost." + VIRTUALHOST + ".transactionTimeout.idleWarn", "5000");
+ setConfigurationProperty("virtualhosts.virtualhost." + VIRTUALHOST + ".transactionTimeout.idleClose", "15000");
+ }
+
+ protected void setUp() throws Exception
+ {
+ // Configure timeouts
+ configure();
+
+ // Monitor log file
+ _monitor = new LogMonitor(_outputFile);
+
+ // Start broker
+ super.setUp();
+
+ // Connect to broker
+ String broker = _broker.equals(VM) ? ("vm://:" + DEFAULT_VM_PORT) : ("tcp://localhost:" + DEFAULT_PORT);
+ ConnectionURL url = new AMQConnectionURL("amqp://guest:guest@clientid/test?brokerlist='" + broker + "'&maxprefetch='1'");
+ _con = (AMQConnection) getConnection(url);
+ _con.setExceptionListener(this);
+ _con.start();
+
+ // Create queue
+ Session qsession = _con.createSession(true, Session.SESSION_TRANSACTED);
+ AMQShortString queueName = new AMQShortString("test");
+ _queue = new AMQQueue(qsession.getDefaultQueueExchangeName(), queueName, queueName, false, true);
+ qsession.close();
+ }
+
+ protected void tearDown() throws Exception
+ {
+ try
+ {
+ _con.close();
+ }
+ catch (Exception e)
+ {
+ e.printStackTrace();
+ }
+ finally
+ {
+ super.tearDown();
+ }
+ }
+
+ /**
+ * Create a transacted persistent message producer session.
+ */
+ protected void producer() throws Exception
+ {
+ _psession = _con.createSession(true, Session.SESSION_TRANSACTED);
+ _producer = _psession.createProducer(_queue);
+ _producer.setDeliveryMode(DeliveryMode.PERSISTENT);
+ }
+
+ /**
+ * Create a transacted message consumer session.
+ */
+ protected void consumer() throws Exception
+ {
+ _csession = _con.createSession(true, Session.SESSION_TRANSACTED);
+ _consumer = _csession.createConsumer(_queue);
+ }
+
+ /**
+ * Send a number of messages to the queue, optionally pausing after each.
+ */
+ protected void send(int count, int delay) throws Exception
+ {
+ for (int i = 0; i < count; i++)
+ {
+ sleep(delay);
+ Message msg = _psession.createTextMessage(TEXT);
+ msg.setIntProperty("i", i);
+ _producer.send(msg);
+ }
+ }
+
+ /**
+ * Sleep for an integral number of seconds.
+ */
+ protected void sleep(int seconds) throws Exception
+ {
+ try
+ {
+ Thread.sleep(seconds * 1000L);
+ }
+ catch (InterruptedException ie)
+ {
+ throw new RuntimeException("Interrupted");
+ }
+ }
+
+ /**
+ * Check for idle and open messages.
+ *
+ * Either exactly zero messages, or +-2 error accepted around the specified number.
+ */
+ protected void monitor(int idle, int open) throws Exception
+ {
+ List<String> idleMsgs = _monitor.findMatches(CHN_IDLE_TXN);
+ List<String> openMsgs = _monitor.findMatches(CHN_OPEN_TXN);
+
+ String idleErr = "Expected " + idle + " but found " + idleMsgs.size() + " txn idle messages";
+ String openErr = "Expected " + open + " but found " + openMsgs.size() + " txn open messages";
+
+ if (idle == 0)
+ {
+ assertTrue(idleErr, idleMsgs.isEmpty());
+ }
+ else
+ {
+ assertTrue(idleErr, idleMsgs.size() >= idle - 2 && idleMsgs.size() <= idle + 2);
+ }
+
+ if (open == 0)
+ {
+ assertTrue(openErr, openMsgs.isEmpty());
+ }
+ else
+ {
+ assertTrue(openErr, openMsgs.size() >= open - 2 && openMsgs.size() <= open + 2);
+ }
+ }
+
+ /**
+ * Receive a number of messages, optionally pausing after each.
+ */
+ protected void expect(int count, int delay) throws Exception
+ {
+ for (int i = 0; i < count; i++)
+ {
+ sleep(delay);
+ Message msg = _consumer.receive(1000);
+ assertNotNull("Message should not be null", msg);
+ assertTrue("Message should be a text message", msg instanceof TextMessage);
+ assertEquals("Message content does not match expected", TEXT, ((TextMessage) msg).getText());
+ assertEquals("Message order is incorrect", i, msg.getIntProperty("i"));
+ }
+ }
+
+ /**
+ * Checks that the correct exception was thrown and was received
+ * by the listener with a 408 error code.
+ */
+ protected void check(String reason)throws InterruptedException
+ {
+ assertTrue("Should have caught exception in listener", _caught.await(1, TimeUnit.SECONDS));
+ assertNotNull("Should have thrown exception to client", _exception);
+ assertTrue("Exception message should contain '" + reason + "': " + _message, _message.contains(reason + " transaction timed out"));
+ assertNotNull("Exception should have an error code", _code);
+ assertEquals("Error code should be 408", AMQConstant.REQUEST_TIMEOUT, _code);
+ }
+
+ /** @see javax.jms.ExceptionListener#onException(javax.jms.JMSException) */
+ public void onException(JMSException jmse)
+ {
+ _caught.countDown();
+ _message = jmse.getLinkedException().getMessage();
+ if (jmse.getLinkedException() instanceof AMQException)
+ {
+ _code = ((AMQException) jmse.getLinkedException()).getErrorCode();
+ }
+ }
+}
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org