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 | 6cf94bdb3acb2887664e953b17d600d8f3b7386a (patch) | |
| tree | 0166d51f1b060e1eed8cedde215bf86b46795128 /qpid/java | |
| parent | 5086f15e77748fc19c3548a2e318ced1eaaa2e02 (diff) | |
| download | qpid-python-6cf94bdb3acb2887664e953b17d600d8f3b7386a.tar.gz | |
QPID-4383 : Fix receipt of large messages, client ack
git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1400447 13f79535-47bb-0310-9956-ffa450edef68
Diffstat (limited to 'qpid/java')
2 files changed, 17 insertions, 9 deletions
diff --git a/qpid/java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/transport/Delivery.java b/qpid/java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/transport/Delivery.java index 4135199045..aca781afb9 100644 --- a/qpid/java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/transport/Delivery.java +++ b/qpid/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/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 cf86fc2471..f8c2f115c5 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 @@ -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)))
{
|