You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ri...@apache.org on 2007/11/27 19:09:35 UTC
svn commit: r598721 - in /incubator/qpid/branches/M2.1.1/java/client/src:
main/java/org/apache/qpid/client/AMQSession.java
main/java/org/apache/qpid/client/BasicMessageConsumer.java
test/java/org/apache/qpid/test/unit/transacted/CommitRollbackTest.java
Author: ritchiem
Date: Tue Nov 27 10:09:33 2007
New Revision: 598721
URL: http://svn.apache.org/viewvc?rev=598721&view=rev
Log:
QPID-621 : Patch Supplied by Aidan Skinner. Msg Ack after msg consumer is closed.
Modified:
incubator/qpid/branches/M2.1.1/java/client/src/main/java/org/apache/qpid/client/AMQSession.java
incubator/qpid/branches/M2.1.1/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java
incubator/qpid/branches/M2.1.1/java/client/src/test/java/org/apache/qpid/test/unit/transacted/CommitRollbackTest.java
Modified: incubator/qpid/branches/M2.1.1/java/client/src/main/java/org/apache/qpid/client/AMQSession.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/M2.1.1/java/client/src/main/java/org/apache/qpid/client/AMQSession.java?rev=598721&r1=598720&r2=598721&view=diff
==============================================================================
--- incubator/qpid/branches/M2.1.1/java/client/src/main/java/org/apache/qpid/client/AMQSession.java (original)
+++ incubator/qpid/branches/M2.1.1/java/client/src/main/java/org/apache/qpid/client/AMQSession.java Tue Nov 27 10:09:33 2007
@@ -106,8 +106,10 @@
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Iterator;
+import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
@@ -243,6 +245,12 @@
*/
private Map<AMQShortString, BasicMessageConsumer> _consumers =
new ConcurrentHashMap<AMQShortString, BasicMessageConsumer>();
+
+ /**
+ * Contains a list of consumers which have been removed but which might still have
+ * messages to acknowledge, eg in client ack or transacted modes
+ */
+ private CopyOnWriteArrayList<BasicMessageConsumer> _removedConsumers = new CopyOnWriteArrayList<BasicMessageConsumer>();
/** Provides a count of consumers on destinations, in order to be able to know if a destination has consumers. */
private ConcurrentHashMap<Destination, AtomicInteger> _destinationConsumerCount =
@@ -647,6 +655,22 @@
lastTag = next;
}
}
+
+ if (_transacted)
+ {
+ // Do the above, but for consumers which have been de-registered since the
+ // last commit
+ for (int i = 0; i < _removedConsumers.size(); i++)
+ {
+ // Sends acknowledgement to server
+ Long next = _removedConsumers.get(i).getLastDelivered();
+ if (next != null && next > lastTag)
+ {
+ lastTag = next;
+ }
+ _removedConsumers.remove(i);
+ }
+ }
if (lastTag != -1)
{
@@ -1690,6 +1714,14 @@
_destinationConsumerCount.remove(dest);
}
}
+
+
+ // Consumers that are closed in a transaction must be stored
+ // so that messages they have received can be acknowledged on commit
+ if (_transacted)
+ {
+ _removedConsumers.add(consumer);
+ }
}
}
@@ -2652,7 +2684,7 @@
}
// Reject messages on pre-receive queue
- consumer.rollback();
+ consumer.rollbackPendingMessages();
// Reject messages on pre-dispatch queue
rejectMessagesForConsumerTag(consumer.getConsumerTag(), true);
@@ -2694,6 +2726,13 @@
consumer.clearReceiveQueue();
}
+ }
+
+ for (int i = 0; i < _removedConsumers.size(); i++)
+ {
+ // Sends acknowledgement to server
+ _removedConsumers.get(i).rollback();
+ _removedConsumers.remove(i);
}
setConnectionStopped(isStopped);
Modified: incubator/qpid/branches/M2.1.1/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/M2.1.1/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java?rev=598721&r1=598720&r2=598721&view=diff
==============================================================================
--- incubator/qpid/branches/M2.1.1/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java (original)
+++ incubator/qpid/branches/M2.1.1/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java Tue Nov 27 10:09:33 2007
@@ -953,7 +953,11 @@
}
}
- // rollback pending messages
+ rollbackPendingMessages();
+ }
+
+ public void rollbackPendingMessages()
+ {
if (_synchronousQueue.size() > 0)
{
if (_logger.isDebugEnabled())
Modified: incubator/qpid/branches/M2.1.1/java/client/src/test/java/org/apache/qpid/test/unit/transacted/CommitRollbackTest.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/M2.1.1/java/client/src/test/java/org/apache/qpid/test/unit/transacted/CommitRollbackTest.java?rev=598721&r1=598720&r2=598721&view=diff
==============================================================================
--- incubator/qpid/branches/M2.1.1/java/client/src/test/java/org/apache/qpid/test/unit/transacted/CommitRollbackTest.java (original)
+++ incubator/qpid/branches/M2.1.1/java/client/src/test/java/org/apache/qpid/test/unit/transacted/CommitRollbackTest.java Tue Nov 27 10:09:33 2007
@@ -458,6 +458,7 @@
assertTrue("Messasge is marked as redelivered" + result, !result.getJMSRedelivered());
_logger.info("Closing Consumer");
+ _session.rollback();
_consumer.close();
_logger.info("Creating New consumer");