Author: rajith
Date: Tue Mar 8 16:06:39 2011
New Revision: 1079408
URL: http://svn.apache.org/viewvc?rev=1079408&view=rev
Log:
QPID-2732
The reliability mode is now used on the producer side to determine replay.
Any messages transfers sent to a destination marked unreliable will not be
added to the replay buffer.
Modified:
qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_10.java
qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/Session.java
Modified:
qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_10.java
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_10.java?rev=1079408&r1=1079407&r2=1079408&view=diff
==============================================================================
---
qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_10.java
(original)
+++
qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_10.java
Tue Mar 8 16:06:39 2011
@@ -19,6 +19,7 @@ package org.apache.qpid.client;
import static org.apache.qpid.transport.Option.NONE;
import static org.apache.qpid.transport.Option.SYNC;
+import static org.apache.qpid.transport.Option.UNRELIABLE;
import java.nio.ByteBuffer;
import java.util.HashMap;
@@ -34,6 +35,7 @@ import org.apache.qpid.client.AMQDestina
import org.apache.qpid.client.AMQDestination.DestSyntax;
import org.apache.qpid.client.message.AMQMessageDelegate_0_10;
import org.apache.qpid.client.message.AbstractJMSMessage;
+import org.apache.qpid.client.messaging.address.Link.Reliability;
import org.apache.qpid.client.messaging.address.Node.QueueNode;
import org.apache.qpid.client.protocol.AMQProtocolHandler;
import org.apache.qpid.transport.DeliveryProperties;
@@ -212,6 +214,9 @@ public class BasicMessageProducer_0_10 e
deliveryMode == DeliveryMode.PERSISTENT)
);
+ boolean unreliable = (destination.getDestSyntax() ==
DestSyntax.ADDR) &&
+ (destination.getLink().getReliability() ==
Reliability.UNRELIABLE);
+
org.apache.mina.common.ByteBuffer data = message.getData();
ByteBuffer buffer = data == null ? ByteBuffer.allocate(0) :
data.buf().slice();
@@ -219,7 +224,7 @@ public class BasicMessageProducer_0_10 e
MessageAcceptMode.NONE,
MessageAcquireMode.PRE_ACQUIRED,
new Header(deliveryProp, messageProps),
- buffer, sync ? SYNC : NONE);
+ buffer, sync ? SYNC : NONE, unreliable ? UNRELIABLE :
NONE);
if (sync)
{
ssn.sync();
Modified:
qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/Session.java
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/Session.java?rev=1079408&r1=1079407&r2=1079408&view=diff
==============================================================================
---
qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/Session.java
(original)
+++
qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/Session.java
Tue Mar 8 16:06:39 2011
@@ -664,7 +664,12 @@ public class Session extends SessionInvo
{
sessionCommandPoint(0, 0);
}
- if ((!closing && !transacted && m instanceof MessageTransfer)
|| m.hasCompletionListener())
+
+ boolean replayTransfer = !closing && !transacted &&
+ m instanceof MessageTransfer &&
+ ! m.isUnreliable();
+
+ if ((replayTransfer) || m.hasCompletionListener())
{
commands[mod(next, commands.length)] = m;
commandBytes += m.getBodySize();
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:commits-subscr...@qpid.apache.org
