// Let the message store ack the message.
messagePointer.getMessageStore().removeMessage(ack);
if (TransactionManager.isCurrentTransaction()) {
// Hook in a callback on first acked message
if (!ts.hasAckedMessage()) {
TransactionManager.getContexTransaction().addPostRollbackTask(new TransactionTask() {
public void execute() throws Throwable {
List ackList = ts.listAckedMessages();
HashMap redeliverMap = new HashMap();
//for (int x = ackList.size()-1; x >= 0 ; x--){
for (int x = 0; x < ackList.size(); x++){
DurableMessagePointer messagePointer = (DurableMessagePointer) ackList.get(x);
ActiveMQMessage message = messagePointer.getMessage();
message.setJMSRedelivered(true);
if (message.incrementDeliveryCount() >= redeliveryPolicy.getMaximumRetryCount()) {
if (log.isDebugEnabled()){
log.debug("Message: " + message + " has exceeded its retry count");
}
// TODO: see if we can use the deadLetterPolicy of the container that dispatched the message.
deadLetterPolicy.sendToDeadLetter(message);
}
else if (ack.isExpired()) {
if (log.isDebugEnabled()){
log.debug("Message: " + message + " has expired");
}
// TODO: see if we can use the deadLetterPolicy of the container that dispatched the message.
deadLetterPolicy.sendToDeadLetter(message);
}
else {
Set containers = destinationMap.get(message.getJMSActiveMQDestination());
for (Iterator i = containers.iterator();i.hasNext();) {
DurableQueueBoundedMessageContainer container = (DurableQueueBoundedMessageContainer) i.next();
LinkedList l = (LinkedList) redeliverMap.get(container);
if( l==null ) {
l = new LinkedList();
redeliverMap.put(container, l);
}
l.add(messagePointer);
}
}
}
for (Iterator i = redeliverMap.keySet().iterator();i.hasNext();) {
DurableQueueBoundedMessageContainer container = (DurableQueueBoundedMessageContainer) i.next();
List l = (List) redeliverMap.get(container);
container.redeliver(l);
}
ts.removeAllAckedMessages();
}
});
TransactionManager.getContexTransaction().addPostCommitTask(new TransactionTask() {
public void execute() throws Throwable {
ts.removeAllAckedMessages();
}
});
}