in broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/StandardReceivingLinkEndpoint.java [157:392]
protected Error receiveDelivery(Delivery delivery)
{
ReceiverSettleMode transferReceiverSettleMode = delivery.getReceiverSettleMode();
if(delivery.getResume())
{
DeliveryState deliveryState = _unsettled.get(delivery.getDeliveryTag());
if (deliveryState instanceof Outcome)
{
boolean settled = shouldReceiverSettleFirst(transferReceiverSettleMode);
updateDisposition(delivery.getDeliveryTag(), deliveryState, settled);
return null;
}
else
{
// TODO: QPID-7845: create message ?
}
}
else
{
ServerMessage<?> serverMessage;
UnsignedInteger messageFormat = delivery.getMessageFormat();
DeliveryState xfrState = delivery.getState();
MessageFormat format = MessageFormatRegistry.getFormat(messageFormat.intValue());
if(format != null)
{
if (delivery.getTotalPayloadSize() == 0)
{
return new Error(AmqpError.NOT_IMPLEMENTED, "Delivery without payload is not supported");
}
try (QpidByteBuffer payload = delivery.getPayload())
{
serverMessage = format.createMessage(payload,
getAddressSpace().getMessageStore(),
getSession().getConnection().getReference());
}
catch (AmqpErrorRuntimeException e)
{
return e.getCause().getError();
}
}
else
{
final Error err = new Error();
err.setCondition(AmqpError.NOT_IMPLEMENTED);
err.setDescription("Unknown message format: " + messageFormat);
return err;
}
MessageReference<?> reference = serverMessage.newReference();
try
{
Binary transactionId = null;
if (xfrState != null)
{
if (xfrState instanceof TransactionalState)
{
transactionId = ((TransactionalState) xfrState).getTxnId();
}
}
final ServerTransaction transaction;
boolean setRollbackOnly = true;
if (transactionId != null)
{
try
{
transaction = getSession().getTransaction(transactionId);
}
catch (UnknownTransactionException e)
{
return new Error(TransactionError.UNKNOWN_ID,
String.format("transaction-id '%s' is unknown.", transactionId));
}
if (!(transaction instanceof AutoCommitTransaction))
{
transaction.addPostTransactionAction(new ServerTransaction.Action()
{
@Override
public void postCommit()
{
updateDisposition(delivery.getDeliveryTag(), null, true);
}
@Override
public void onRollback()
{
updateDisposition(delivery.getDeliveryTag(), null, true);
}
});
}
}
else
{
transaction = new AsyncAutoCommitTransaction(getAddressSpace().getMessageStore(), this);
}
try
{
Session_1_0 session = getSession();
session.getAMQPConnection()
.checkAuthorizedMessagePrincipal(serverMessage.getMessageHeader().getUserId());
Outcome outcome;
if (serverMessage.isPersistent() && !getAddressSpace().getMessageStore().isPersistent())
{
final Error preconditionFailedError = new Error(AmqpError.PRECONDITION_FAILED,
"Non-durable message store cannot accept durable message.");
if (_rejectedOutcomeSupportedBySource)
{
final Rejected rejected = new Rejected();
rejected.setError(preconditionFailedError);
outcome = rejected;
}
else
{
// TODO - disposition not updated for the non-transaction case
return preconditionFailedError;
}
}
else
{
try
{
getReceivingDestination().send(serverMessage,
transaction,
session.getSecurityToken());
outcome = ACCEPTED;
if (_producer != null)
{
_producer.registerMessageDelivered(serverMessage.getSizeIncludingHeader());
if (Producer.DeliveryType.DELAYED_DELIVERY.equals(_producer.getDeliveryType()))
{
final String to = serverMessage.getTo();
if (!Objects.equals(to, _producer.getDestination()))
{
final MessageDestination messageDestination = getAddressSpace()
.getAttainedMessageDestination(serverMessage.getTo(), false);
_producer.setDestinationType(Producer.DestinationType.from(messageDestination));
_producer.setDestinationId(Producer.DestinationType.getId(messageDestination));
_producer.setDestination(to);
}
}
}
}
catch (UnroutableMessageException e)
{
final Error error = new Error();
error.setCondition(e.getErrorCondition());
error.setDescription(e.getMessage());
String targetAddress = getTarget().getAddress();
if (targetAddress == null || "".equals(targetAddress.trim()))
{
error.setInfo(Collections.singletonMap(DELIVERY_TAG, delivery.getDeliveryTag()));
}
if (!_rejectedOutcomeSupportedBySource ||
(delivery.isSettled() && !(transaction instanceof LocalTransaction)))
{
return error;
}
else
{
if (delivery.isSettled() && transaction instanceof LocalTransaction)
{
((LocalTransaction) transaction).setRollbackOnly();
}
Rejected rejected = new Rejected();
rejected.setError(error);
outcome = rejected;
}
}
}
Outcome sourceDefaultOutcome = getSource().getDefaultOutcome();
boolean defaultOutcome = sourceDefaultOutcome != null &&
sourceDefaultOutcome.getSymbol().equals(outcome.getSymbol());
DeliveryState resultantState;
if (transactionId == null)
{
resultantState = defaultOutcome ? null : outcome;
}
else
{
TransactionalState transactionalState = new TransactionalState();
transactionalState.setOutcome(defaultOutcome ? null : outcome);
transactionalState.setTxnId(transactionId);
resultantState = transactionalState;
}
boolean settled = shouldReceiverSettleFirst(transferReceiverSettleMode);
if (transaction instanceof AsyncAutoCommitTransaction)
{
_pendingDispositions.add(new PendingDispositionHolder(delivery.getDeliveryTag(),
resultantState,
settled));
}
else
{
getSession().receivedComplete();
updateDisposition(delivery.getDeliveryTag(), resultantState, settled);
}
getSession().registerMessageReceived(serverMessage.getSize());
if (transactionId != null)
{
getSession().registerTransactedMessageReceived();
}
setRollbackOnly = false;
}
catch (AccessControlException e)
{
final Error err = new Error();
err.setCondition(AmqpError.NOT_ALLOWED);
err.setDescription(e.getMessage());
return err;
}
finally
{
if (setRollbackOnly && transaction instanceof LocalTransaction)
{
((LocalTransaction) transaction).setRollbackOnly();
}
}
}
finally
{
reference.release();
}
}
return null;
}