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 2009/10/25 19:57:17 UTC
svn commit: r829628 -
/qpid/branches/java-broker-0-10/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java
Author: rgodfrey
Date: Sun Oct 25 18:57:16 2009
New Revision: 829628
URL: http://svn.apache.org/viewvc?rev=829628&view=rev
Log:
Fixed immediate messages to dequeue
Modified:
qpid/branches/java-broker-0-10/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java
Modified: qpid/branches/java-broker-0-10/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-0-10/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java?rev=829628&r1=829627&r2=829628&view=diff
==============================================================================
--- qpid/branches/java-broker-0-10/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java (original)
+++ qpid/branches/java-broker-0-10/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java Sun Oct 25 18:57:16 2009
@@ -55,6 +55,7 @@
import org.apache.qpid.server.logging.subjects.ChannelLogSubject;
import org.apache.qpid.server.logging.actors.AMQPChannelActor;
import org.apache.qpid.server.logging.actors.CurrentActor;
+import org.apache.qpid.server.output.ProtocolOutputConverter;
public class AMQChannel
{
@@ -119,6 +120,8 @@
private static final Runnable NULL_TASK = new Runnable() { public void run() {} };
private List<QueueEntry> _resendList = new ArrayList<QueueEntry>();
+ private static final
+ AMQShortString IMMEDIATE_DELIVERY_REPLY_TEXT = new AMQShortString("Immediate delivery is not possible.");
public AMQChannel(AMQProtocolSession session, int channelId, MessageStore messageStore)
throws AMQException
@@ -991,19 +994,42 @@
if(immediate && !entry.getDeliveredToConsumer() && entry.acquire())
{
+
+
ServerTransaction txn = new LocalTransaction(_messageStore);
Collection<QueueEntry> entries = new ArrayList<QueueEntry>(1);
entries.add(entry);
- txn.dequeue(queue, entry.getMessage(), new MessageAcknowledgeAction(entries));
+ final AMQMessage message = (AMQMessage) entry.getMessage();
+ txn.dequeue(queue, entry.getMessage(),
+ new MessageAcknowledgeAction(entries)
+ {
+ @Override
+ public void postCommit()
+ {
+ try
+ {
+ final
+ ProtocolOutputConverter outputConverter =
+ _session.getProtocolOutputConverter();
+
+ outputConverter.writeReturn(message.getMessagePublishInfo(),
+ message.getContentHeaderBody(),
+ message,
+ _channelId,
+ AMQConstant.NO_CONSUMERS.getCode(),
+ IMMEDIATE_DELIVERY_REPLY_TEXT);
+ }
+ catch (AMQException e)
+ {
+ throw new RuntimeException(e);
+ }
+ super.postCommit();
+ }
+ }
+ );
txn.commit();
- AMQMessage message = (AMQMessage) entry.getMessage();
- _session.getProtocolOutputConverter().writeReturn(message.getMessagePublishInfo(),
- message.getContentHeaderBody(),
- message,
- _channelId,
- AMQConstant.NO_CONSUMERS.getCode(),
- new AMQShortString("Immediate delivery is not possible."));
+
}
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org