Author: rgodfrey
Date: Wed Apr 16 02:34:22 2008
New Revision: 648645
URL: http://svn.apache.org/viewvc?rev=648645&view=rev
Log:
QPID-925 : Only begin store transactions when there is a persistent message to
be committed
Modified:
incubator/qpid/branches/M2.1/java/broker/src/main/java/org/apache/qpid/server/ack/TxAck.java
incubator/qpid/branches/M2.1/java/broker/src/main/java/org/apache/qpid/server/txn/LocalTransactionalContext.java
Modified:
incubator/qpid/branches/M2.1/java/broker/src/main/java/org/apache/qpid/server/ack/TxAck.java
URL:
http://svn.apache.org/viewvc/incubator/qpid/branches/M2.1/java/broker/src/main/java/org/apache/qpid/server/ack/TxAck.java?rev=648645&r1=648644&r2=648645&view=diff
==============================================================================
---
incubator/qpid/branches/M2.1/java/broker/src/main/java/org/apache/qpid/server/ack/TxAck.java
(original)
+++
incubator/qpid/branches/M2.1/java/broker/src/main/java/org/apache/qpid/server/ack/TxAck.java
Wed Apr 16 02:34:22 2008
@@ -22,6 +22,7 @@
import java.util.LinkedList;
import java.util.List;
+import java.util.ArrayList;
import org.apache.qpid.AMQException;
import org.apache.qpid.server.store.StoreContext;
@@ -33,8 +34,8 @@
public class TxAck implements TxnOp
{
private final UnacknowledgedMessageMap _map;
- private final List <UnacknowledgedMessage> _unacked = new
LinkedList<UnacknowledgedMessage>();
- private final List<Long> _individual = new LinkedList<Long>();
+ private final List <UnacknowledgedMessage> _unacked = new
ArrayList<UnacknowledgedMessage>();
+ private List<Long> _individual;
private long _deliveryTag;
private boolean _multiple;
@@ -47,6 +48,10 @@
{
if (!multiple)
{
+ if(_individual == null)
+ {
+ _individual = new ArrayList<Long>();
+ }
//have acked a single message that is not part of
//the previously acked region so record
//individually
@@ -59,30 +64,46 @@
_deliveryTag = deliveryTag;
_multiple = true;
}
+ _unacked.clear();
}
public void consolidate()
{
+ if(_unacked.isEmpty())
+ {
+ consolidate(_unacked);
+ }
+
+ }
+
+ private void consolidate(List<UnacknowledgedMessage> unacked)
+ {
//lookup all the unacked messages that have been acked in this
transaction
if (_multiple)
{
//get all the unacked messages for the accumulated
//multiple acks
- _map.collect(_deliveryTag, true, _unacked);
+ _map.collect(_deliveryTag, true, unacked);
}
//get any unacked messages for individual acks outside the
//range covered by multiple acks
- for (long tag : _individual)
+ if(_individual != null)
{
- if(_deliveryTag < tag)
+ for (Long tag : _individual)
{
- _map.collect(tag, false, _unacked);
+ if(_deliveryTag < tag)
+ {
+ _map.collect(tag, false, unacked);
+ }
}
}
}
public boolean checkPersistent() throws AMQException
{
+
+
+ consolidate();
//if any of the messages in unacked are persistent the txn
//buffer must be marked as persistent:
for (UnacknowledgedMessage msg : _unacked)
Modified:
incubator/qpid/branches/M2.1/java/broker/src/main/java/org/apache/qpid/server/txn/LocalTransactionalContext.java
URL:
http://svn.apache.org/viewvc/incubator/qpid/branches/M2.1/java/broker/src/main/java/org/apache/qpid/server/txn/LocalTransactionalContext.java?rev=648645&r1=648644&r2=648645&view=diff
==============================================================================
---
incubator/qpid/branches/M2.1/java/broker/src/main/java/org/apache/qpid/server/txn/LocalTransactionalContext.java
(original)
+++
incubator/qpid/branches/M2.1/java/broker/src/main/java/org/apache/qpid/server/txn/LocalTransactionalContext.java
Wed Apr 16 02:34:22 2008
@@ -148,8 +148,9 @@
// we will need to create and enlist the op.
if (_ackOp == null)
{
- beginTranIfNecessary();
+
_ackOp = new TxAck(unacknowledgedMessageMap);
+
_txnBuffer.enlist(_ackOp);
}
// update the op to include this ack request
@@ -162,6 +163,10 @@
else
{
_ackOp.update(deliveryTag, multiple);
+ }
+ if(!_inTran && _ackOp.checkPersistent())
+ {
+ beginTranIfNecessary();
}
}
