diff options
| author | Robert Godfrey <rgodfrey@apache.org> | 2012-05-11 12:46:43 +0000 |
|---|---|---|
| committer | Robert Godfrey <rgodfrey@apache.org> | 2012-05-11 12:46:43 +0000 |
| commit | 627984e1ae185bfeae12ad8c1d5f5e2024f7f414 (patch) | |
| tree | b038075c3726c9ce9b392362ceb60c83f02287c9 | |
| parent | bece1e05e7264e85168b0e564623792904d322d6 (diff) | |
| download | qpid-python-627984e1ae185bfeae12ad8c1d5f5e2024f7f414.tar.gz | |
QPID-3993 : NPE on continuation transfers with no deliveryId
git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1337130 13f79535-47bb-0310-9956-ffa450edef68
2 files changed, 34 insertions, 4 deletions
diff --git a/qpid/java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/transport/ReceivingLinkEndpoint.java b/qpid/java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/transport/ReceivingLinkEndpoint.java index bd9244c858..cf86fc2471 100644 --- a/qpid/java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/transport/ReceivingLinkEndpoint.java +++ b/qpid/java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/transport/ReceivingLinkEndpoint.java @@ -32,6 +32,8 @@ public class ReceivingLinkEndpoint extends LinkEndpoint<ReceivingLinkListener> {
+ private UnsignedInteger _lastDeliveryId;
+
private static class TransientState
{
@@ -435,4 +437,16 @@ public class ReceivingLinkEndpoint extends LinkEndpoint<ReceivingLinkListener> {
return _drainLimit;
}
+
+ UnsignedInteger getLastDeliveryId()
+ {
+ return _lastDeliveryId;
+ }
+
+ void setLastDeliveryId(UnsignedInteger lastDeliveryId)
+ {
+ _lastDeliveryId = lastDeliveryId;
+ }
+
+
}
diff --git a/qpid/java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/transport/SessionEndpoint.java b/qpid/java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/transport/SessionEndpoint.java index 4b8bf82a4a..17f90fef59 100644 --- a/qpid/java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/transport/SessionEndpoint.java +++ b/qpid/java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/transport/SessionEndpoint.java @@ -321,21 +321,32 @@ public class SessionEndpoint }
- Delivery delivery = _incomingUnsettled.get(transfer.getDeliveryId());
+ UnsignedInteger deliveryId = transfer.getDeliveryId();
+ if(deliveryId == null)
+ {
+ deliveryId = ((ReceivingLinkEndpoint)endpoint).getLastDeliveryId();
+ }
+
+ Delivery delivery = _incomingUnsettled.get(deliveryId);
if(delivery == null)
{
delivery = new Delivery(transfer, endpoint);
- _incomingUnsettled.put(transfer.getDeliveryId(),delivery);
+ _incomingUnsettled.put(deliveryId,delivery);
if(delivery.isSettled() || Boolean.TRUE.equals(transfer.getAborted()))
{
/*
_availableIncomingCredit++;
*/
}
+
+ if(Boolean.TRUE.equals(transfer.getMore()))
+ {
+ ((ReceivingLinkEndpoint)endpoint).setLastDeliveryId(transfer.getDeliveryId());
+ }
}
else
{
- if(delivery.getDeliveryId().equals(transfer.getDeliveryId()))
+ if(delivery.getDeliveryId().equals(deliveryId))
{
delivery.addTransfer(transfer);
if(delivery.isSettled())
@@ -350,6 +361,11 @@ public class SessionEndpoint _availableIncomingCredit += delivery.getTransfers().size();
*/
}
+
+ if(!Boolean.TRUE.equals(transfer.getMore()))
+ {
+ ((ReceivingLinkEndpoint)endpoint).setLastDeliveryId(null);
+ }
}
else
{
@@ -365,7 +381,7 @@ public class SessionEndpoint if((delivery.isComplete() && delivery.isSettled() || Boolean.TRUE.equals(transfer.getAborted())))
{
- _incomingUnsettled.remove(transfer.getDeliveryId());
+ _incomingUnsettled.remove(deliveryId);
}
}
}
|