public void internalConsumeMessages ( final LockedMessageEnumeration lockedMessages , final AsynchDispatchScheduler asynchDispatchScheduler ) { final String methodName = "internalConsumeMessages" ; if ( TraceComponent . isAnyTracingEnabled ( ) && TRACE . isEntryEnabled ( ) ) { SibTr . entry ( this , TRACE , methodName , new Object [ ] { lockedMessages , asynchDispatchScheduler } ) ; } try { /* * Retrieve messages */ final List < DispatchableMessage > messages = new ArrayList < DispatchableMessage > ( ) ; SIBusMessage message ; while ( ( message = lockedMessages . nextLocked ( ) ) != null ) { // chetan liberty change : // RAHandler and WAS.exitpoint is not applicable in liberty.Hence removing it final Map handlerContext = new FastSerializableHashMap ( ) ; DispatchableMessage dispatchableMessage = new DispatchableMessage ( message , handlerContext ) ; messages . add ( dispatchableMessage ) ; /* * If required, delete best-effort nonpersistent messages */ if ( _deleteUnrecoverableMessages && ( _unrecoverableReliability . compareTo ( message . getReliability ( ) ) >= 0 ) ) { lockedMessages . deleteCurrent ( null ) ; } } //Only schedule some work if we have some messages if ( messages . size ( ) != 0 ) { final SibRaWork work = new SibRaWork ( ) ; /* * Schedule messages for dispatch * The asynchDispatchScheduler will either * be a cached version if we are been resumed from a * suspend or a newly obtained version */ work . schedule ( messages , asynchDispatchScheduler , this ) ; } } catch ( final Throwable throwable ) { FFDCFilter . processException ( throwable , CLASS_NAME + "." + methodName , "1:375:1.42" , this ) ; SibTr . error ( TRACE , "RETRIEVE_MESSAGES_CWSIV1100" , new Object [ ] { throwable , lockedMessages } ) ; } if ( TraceComponent . isAnyTracingEnabled ( ) && TRACE . isEntryEnabled ( ) ) { SibTr . exit ( this , TRACE , methodName ) ; } }
The consumeMessages method is invoked by the message processor with an enumeration containing one or more messages locked to this consumer . The consumeMessages method will either call the internalConsumeMessage method directly or if XD wishes to suspend the processing this method will be called then XD decides to resume processing . This method retrieves the messages from the enumeration . It then schedules a piece of work that will create the dispatcher on a new thread .