diff options
| author | Robert Godfrey <rgodfrey@apache.org> | 2012-10-20 13:43:04 +0000 |
|---|---|---|
| committer | Robert Godfrey <rgodfrey@apache.org> | 2012-10-20 13:43:04 +0000 |
| commit | 8aabbfd6c4c70df5ace15b54dda8475662029e93 (patch) | |
| tree | 805984b7f078fa7bde57fda03c0f37b59c2ec0b5 /java | |
| parent | 29588efa544851170fae5071f3170a4dcb58510a (diff) | |
| download | qpid-python-8aabbfd6c4c70df5ace15b54dda8475662029e93.tar.gz | |
QPID-4383 : Fix receipt of large messages, client ack
git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk/qpid@1400447 13f79535-47bb-0310-9956-ffa450edef68
Diffstat (limited to 'java')
2 files changed, 17 insertions, 9 deletions
diff --git a/java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/transport/Delivery.java b/java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/transport/Delivery.java index 4135199045..aca781afb9 100644 --- a/java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/transport/Delivery.java +++ b/java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/transport/Delivery.java @@ -71,6 +71,10 @@ public class Delivery { setComplete(true); } + if(Boolean.TRUE.equals(transfer.getSettled())) + { + setSettled(true); + } } public List<Transfer> getTransfers() diff --git a/java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/transport/ReceivingLinkEndpoint.java b/java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/transport/ReceivingLinkEndpoint.java index cf86fc2471..f8c2f115c5 100644 --- a/java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/transport/ReceivingLinkEndpoint.java +++ b/java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/transport/ReceivingLinkEndpoint.java @@ -113,33 +113,37 @@ public class ReceivingLinkEndpoint extends LinkEndpoint<ReceivingLinkListener> synchronized (getLock())
{
TransientState transientState;
- boolean existingState = _unsettledMap.containsKey(transfer.getDeliveryTag());
- _unsettledMap.put(transfer.getDeliveryTag(), transfer.getState());
+ final Binary deliveryTag = delivery.getDeliveryTag();
+ boolean existingState = _unsettledMap.containsKey(deliveryTag);
+ if(!existingState || transfer.getState() != null)
+ {
+ _unsettledMap.put(deliveryTag, transfer.getState());
+ }
if(!existingState)
{
transientState = new TransientState(transfer.getDeliveryId());
- if(Boolean.TRUE.equals(transfer.getSettled()))
+ if(delivery.isSettled())
{
transientState.setSettled(true);
}
- _unsettledIds.put(transfer.getDeliveryTag(), transientState);
+ _unsettledIds.put(deliveryTag, transientState);
setLinkCredit(getLinkCredit().subtract(UnsignedInteger.ONE));
setDeliveryCount(getDeliveryCount().add(UnsignedInteger.ONE));
}
else
{
- transientState = _unsettledIds.get(transfer.getDeliveryTag());
+ transientState = _unsettledIds.get(deliveryTag);
transientState.incrementCredit();
- if(Boolean.TRUE.equals(transfer.getSettled()))
+ if(delivery.isSettled())
{
transientState.setSettled(true);
}
}
- if(transientState.isSettled())
+ if(transientState.isSettled() && delivery.isComplete())
{
- _unsettledMap.remove(transfer.getDeliveryTag());
+ _unsettledMap.remove(deliveryTag);
}
getLinkEventListener().messageTransfer(transfer);
@@ -371,7 +375,7 @@ public class ReceivingLinkEndpoint extends LinkEndpoint<ReceivingLinkListener> tag = iter.next();
tagsToUpdate.add(tag);
- deliveryId = _unsettledIds.get(firstTag).getDeliveryId();
+ deliveryId = _unsettledIds.get(tag).getDeliveryId();
if(deliveryId.equals(last.add(UnsignedInteger.ONE)))
{
|