* @return <code>true</code> if message processing to continue, <code>false</code> if to be dropped
* @throws RMException if message had already been acknowledged
*/
boolean applyDeliveryAssurance(long mn, Message message) throws RMException {
Continuation cont = getContinuation(message);
DeliveryAssuranceType da = destination.getManager().getDeliveryAssurance();
boolean canSkip = !da.isSetAtLeastOnce() && !da.isSetExactlyOnce();
boolean robust = false;
boolean robustDelivering = false;
if (message != null) {
robust = MessageUtils.isTrue(message.getContextualProperty(Message.ROBUST_ONEWAY));
if (robust) {
robustDelivering =
MessageUtils.isTrue(message.get(RMMessageConstants.DELIVERING_ROBUST_ONEWAY));
}
}
if (robust && !robustDelivering) {
// no check performed if in robust and not in delivering
removeDeliveringMessageNumber(mn);
return true;
}
if (cont != null && da.isSetInOrder() && !cont.isNew()) {
return waitInQueue(mn, canSkip, message, cont);
}
if ((da.isSetExactlyOnce() || da.isSetAtMostOnce())
&& (isAcknowledged(mn)
|| (robustDelivering && deliveringMessageNumbers.contains(mn)))) {
// acknowledge at first opportunity following duplicate message
scheduleImmediateAcknowledgement();
org.apache.cxf.common.i18n.Message msg = new org.apache.cxf.common.i18n.Message(
"MESSAGE_ALREADY_DELIVERED_EXC", LOG, mn, getIdentifier().getValue());
LOG.log(Level.INFO, msg.toString());
throw new RMException(msg);
}
if (robustDelivering) {
deliveringMessageNumbers.add(mn);
}
if (da.isSetInOrder()) {
return waitInQueue(mn, canSkip, message, cont);
}
return true;
}