You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by gt...@apache.org on 2013/02/05 19:30:33 UTC
svn commit: r1442688 -
/activemq/trunk/activemq-stomp/src/main/java/org/apache/activemq/transport/stomp/StompSubscription.java
Author: gtully
Date: Tue Feb 5 18:30:33 2013
New Revision: 1442688
URL: http://svn.apache.org/viewvc?rev=1442688&view=rev
Log:
https://issues.apache.org/jira/browse/AMQ-4297 -resolve intermittent hang/fail of stomp tests
Modified:
activemq/trunk/activemq-stomp/src/main/java/org/apache/activemq/transport/stomp/StompSubscription.java
Modified: activemq/trunk/activemq-stomp/src/main/java/org/apache/activemq/transport/stomp/StompSubscription.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-stomp/src/main/java/org/apache/activemq/transport/stomp/StompSubscription.java?rev=1442688&r1=1442687&r2=1442688&view=diff
==============================================================================
--- activemq/trunk/activemq-stomp/src/main/java/org/apache/activemq/transport/stomp/StompSubscription.java (original)
+++ activemq/trunk/activemq-stomp/src/main/java/org/apache/activemq/transport/stomp/StompSubscription.java Tue Feb 5 18:30:33 2013
@@ -107,20 +107,26 @@ public class StompSubscription {
unconsumedMessage.clear();
}
- synchronized void onStompCommit(TransactionId transactionId) {
- for (Iterator<?> iter = dispatchedMessage.entrySet().iterator(); iter.hasNext();) {
- @SuppressWarnings("rawtypes")
- Map.Entry entry = (Entry)iter.next();
- MessageDispatch msg = (MessageDispatch)entry.getValue();
- if (unconsumedMessage.contains(msg)) {
- iter.remove();
+ void onStompCommit(TransactionId transactionId) {
+ MessageAck ack = null;
+ synchronized (this) {
+ for (Iterator<?> iter = dispatchedMessage.entrySet().iterator(); iter.hasNext();) {
+ @SuppressWarnings("rawtypes")
+ Map.Entry entry = (Entry)iter.next();
+ MessageDispatch msg = (MessageDispatch)entry.getValue();
+ if (unconsumedMessage.contains(msg)) {
+ iter.remove();
+ }
}
- }
- if (!unconsumedMessage.isEmpty()) {
- MessageAck ack = new MessageAck(unconsumedMessage.getLast(), MessageAck.STANDARD_ACK_TYPE, unconsumedMessage.size());
+ if (!unconsumedMessage.isEmpty()) {
+ ack = new MessageAck(unconsumedMessage.getLast(), MessageAck.STANDARD_ACK_TYPE, unconsumedMessage.size());
+ unconsumedMessage.clear();
+ }
+ }
+ // avoid contention with onMessageDispatch
+ if (ack != null) {
protocolConverter.getStompTransport().sendToActiveMQ(ack);
- unconsumedMessage.clear();
}
}