blockForPause();
try{
//get all invoker beans for the sequence
InvokerBeanMgr storageMapMgr = storageManager
.getInvokerBeanMgr();
RMDBeanMgr rmdBeanMgr = storageManager.getRMDBeanMgr();
RMDBean rMDBean = rmdBeanMgr.retrieve(sequenceID);
if (rMDBean != null) {
//The outOfOrder window is the set of known sequence messages (including those
//that are missing) at the time the button is pressed.
long firstMessageInOutOfOrderWindow = rMDBean.getNextMsgNoToProcess();
InvokerBean selector = new InvokerBean();
selector.setSequenceID(sequenceID);
Iterator stMapIt = storageMapMgr.find(selector).iterator();
long highestMsgNumberInvoked = 0;
Transaction transaction = null;
//invoke each bean in turn.
//NOTE: here we are breaking ordering
while(stMapIt.hasNext()){
//invoke the app
try{
transaction = storageManager.getTransaction();
InvokerBean invoker = (InvokerBean)stMapIt.next();
// start a new worker thread and let it do the invocation.
String workId = sequenceID + "::" + invoker.getMsgNo(); //creating a workId to uniquely identify the
//piece of work that will be assigned to the Worker.
String messageContextKey = invoker.getMessageContextRefKey();
InvokerWorker worker = new InvokerWorker(context,
messageContextKey,
true); //want to ignore the enxt msg number
worker.setLock(getWorkerLock());
worker.setWorkId(workId);
// Wrap the invoker worker with the correct context, if needed.
Runnable work = worker;
ContextManager contextMgr = SandeshaUtil.getContextManager(context);
if(contextMgr != null) {
work = contextMgr.wrapWithContext(work, invoker.getContext());
}
threadPool.execute(work);
//adding the workId to the lock after assigning it to a thread makes sure
//that all the workIds in the Lock are handled by threads.
getWorkerLock().addWork(workId);
long msgNumber = invoker.getMsgNo();
//if necessary, update the "next message number" bean under this transaction
if(msgNumber>highestMsgNumberInvoked){
highestMsgNumberInvoked = invoker.getMsgNo();
rMDBean.setNextMsgNoToProcess(highestMsgNumberInvoked+1);
if(allowLaterDeliveryOfMissingMessages){
//we also need to update the sequence OUT_OF_ORDER_RANGES property
//so as to include our latest view of this outOfOrder range.
//We do that here (rather than once at the end) so that we reamin
//transactionally consistent
Range r = new Range(firstMessageInOutOfOrderWindow,highestMsgNumberInvoked);
RangeString rangeString = null;
if(rMDBean.getOutOfOrderRanges()==null){
//insert a new blank one one
rangeString = new RangeString();
}
else{
rangeString = rMDBean.getOutOfOrderRanges();
}
//update the range String with the new value
rangeString.addRange(r);
rMDBean.setOutOfOrderRanges(rangeString);
}
rmdBeanMgr.update(rMDBean);
}
if(transaction != null && transaction.isActive()) transaction.commit();
transaction = null;
}