You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by rg...@apache.org on 2013/04/01 18:10:58 UTC
svn commit: r1463179 -
/qpid/branches/0.22/qpid/java/broker/src/main/java/org/apache/qpid/server/subscription/SubscriptionImpl.java
Author: rgodfrey
Date: Mon Apr 1 16:10:58 2013
New Revision: 1463179
URL: http://svn.apache.org/r1463179
Log:
QPID-4680 : [Java Broker] NoAck subscription on persistent queue in 0-8/9/9-1 codepath leaves queue entry records in database (merged from trunk r1462941)
Modified:
qpid/branches/0.22/qpid/java/broker/src/main/java/org/apache/qpid/server/subscription/SubscriptionImpl.java
Modified: qpid/branches/0.22/qpid/java/broker/src/main/java/org/apache/qpid/server/subscription/SubscriptionImpl.java
URL: http://svn.apache.org/viewvc/qpid/branches/0.22/qpid/java/broker/src/main/java/org/apache/qpid/server/subscription/SubscriptionImpl.java?rev=1463179&r1=1463178&r2=1463179&view=diff
==============================================================================
--- qpid/branches/0.22/qpid/java/broker/src/main/java/org/apache/qpid/server/subscription/SubscriptionImpl.java (original)
+++ qpid/branches/0.22/qpid/java/broker/src/main/java/org/apache/qpid/server/subscription/SubscriptionImpl.java Mon Apr 1 16:10:58 2013
@@ -42,6 +42,8 @@ import org.apache.qpid.server.protocol.A
import org.apache.qpid.server.protocol.AMQSessionModel;
import org.apache.qpid.server.queue.AMQQueue;
import org.apache.qpid.server.queue.QueueEntry;
+import org.apache.qpid.server.txn.AutoCommitTransaction;
+import org.apache.qpid.server.txn.ServerTransaction;
import java.util.Map;
import java.util.UUID;
@@ -92,7 +94,7 @@ public abstract class SubscriptionImpl i
private final AtomicLong _unacknowledgedBytes = new AtomicLong(0);
private long _createTime = System.currentTimeMillis();
-
+
static final class BrowserSubscription extends SubscriptionImpl
{
@@ -146,6 +148,8 @@ public abstract class SubscriptionImpl i
public static class NoAckSubscription extends SubscriptionImpl
{
+ private volatile AutoCommitTransaction _txn;
+
public NoAckSubscription(AMQChannel channel, AMQProtocolSession protocolSession,
AMQShortString consumerTag, FieldTable filters,
boolean noLocal, FlowCreditManager creditManager,
@@ -190,8 +194,13 @@ public abstract class SubscriptionImpl i
// The send may of course still fail, in which case, as
// the message is unacked, it will be lost.
- entry.dequeue();
+ if(_txn == null)
+ {
+ _txn = new AutoCommitTransaction(getQueue().getVirtualHost().getMessageStore());
+ }
+ _txn.dequeue(getQueue(), entry.getMessage(), NOOP);
+ entry.dequeue();
synchronized (getChannel())
{
@@ -212,6 +221,19 @@ public abstract class SubscriptionImpl i
return false;
}
+ private static final ServerTransaction.Action NOOP =
+ new ServerTransaction.Action()
+ {
+ @Override
+ public void postCommit()
+ {
+ }
+
+ @Override
+ public void onRollback()
+ {
+ }
+ };
}
/**
@@ -275,7 +297,7 @@ public abstract class SubscriptionImpl i
public void send(QueueEntry entry, boolean batch) throws AMQException
{
-
+
synchronized (getChannel())
{
getChannel().getProtocolSession().setDeferFlush(batch);
@@ -576,7 +598,7 @@ public abstract class SubscriptionImpl i
{
return _consumerTag == null ? null : _consumerTag.asString();
}
-
+
public long getSubscriptionID()
{
return _subscriptionID;
@@ -821,7 +843,7 @@ public abstract class SubscriptionImpl i
}
});
}
-
+
public long getUnacknowledgedBytes()
{
return _unacknowledgedBytes.longValue();
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org