diff options
Diffstat (limited to 'java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/transport/ReceivingLinkEndpoint.java')
| -rw-r--r-- | java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/transport/ReceivingLinkEndpoint.java | 24 |
1 files changed, 14 insertions, 10 deletions
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..5fbca0b695 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);
@@ -155,7 +159,7 @@ public class ReceivingLinkEndpoint extends LinkEndpoint<ReceivingLinkListener> super.receiveFlow(flow);
_remoteDrain = Boolean.TRUE.equals((Boolean)flow.getDrain());
setAvailable(flow.getAvailable());
- _remoteTransferCount = flow.getDeliveryCount();
+ setDeliveryCount(flow.getDeliveryCount());
getLock().notifyAll();
}
}
@@ -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)))
{
|