public void deliverBatch(int batchSize, short sessionId, ConversationHelper convHelper)
   {
      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.entry(this, tc, "deliverBatch",
                                           new Object[]
                                           {
                                             ""+batchSize,
                                             ""+sessionId,
                                             convHelper
                                           });

      QueueData queueData = queue.get(0);
      AsynchConsumerProxyQueue proxyQueue = (AsynchConsumerProxyQueue)queueData.getProxyQueue();
      proxyQueue.setAsynchConsumerThread(Thread.currentThread());
      final ConsumerSessionProxy cs = (ConsumerSessionProxy)proxyQueue.getDestinationSessionProxy();

      cs.resetCallbackThreadState();

      boolean unlockAllMessages = false;
      try
      {
         JsMessage[] msgs = getBatch(batchSize, convHelper);

         LockedMessageEnumerationImpl lockedMsgEnum =                                     // D218666.1
                  new LockedMessageEnumerationImpl(proxyQueue,
                                                   this,
                                                   msgs,
                                                   Thread.currentThread(),
                                                   proxyQueue.getLMEOperationMonitor());

         if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) SibTr.debug(this, tc, " ** About to call user callback **");
         proxyQueue.getAsynchConsumerCallback().consumeMessages(lockedMsgEnum);
         if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) SibTr.debug(this, tc, " ** User callback has returned  **");

         // Check the remaining message count. If it is not 0, then the user did
         // not view all the messages and we need to unlock them.
         int remainingLockedMessages = lockedMsgEnum.getRemainingMessageCount();
         if (remainingLockedMessages != 0)
         {
            if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) SibTr.debug(this, tc, "There are still " + remainingLockedMessages + " locked messages! - Unlocking them");

            lockedMsgEnum.unlockUnseen();                                                 // D218666.1
         }
      }
      catch(Throwable t)
      {
         // Start f200337

         // Naughty, naughty. A user callback threw an exception. At this point, the user has
         // stuffed themselves a little bit as they cannot really be sure what the state of play
         // is - they may have unlocked some messages, they may have deleted some and some may
         // still be locked.
         // The strategy is therefore to FFDC, pass the exception to any registered connection
         // listeners (through their asynchronousException() method) and then call unlock all.
         // Any messages that have been deleted are gone, and any messages that were not consumed
         // will be made available for consumption again. If the user was using express messages
         // then they will loose any messages in the locked message enumeration.
         // That'll learm 'em.

         FFDCFilter.processException(t,
                                     CLASS_NAME + ".deliverBatch",
                                     CommsConstants.ASYNCHPQ_DELIVER_01,
                                     this);

         if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) SibTr.debug(this, tc, "exception thrown");
         if (TraceComponent.isAnyTracingEnabled() && tc.isEventEnabled()) SibTr.event(tc, "exception", t);

         proxyQueue.setAsynchConsumerThread(null);

         // Pass the exception to any registered exception listeners
         cs.deliverAsyncException(t);

         // We need to purge our local proxy queue, and unlock all the
         // messages in the ME to redeliver the messages.
         // However, we cannot do this before we act on any stop
         // that might have been called within the asynch consumer
         // before they threw the exception, otherwise the ME
         // will automatically send us the next batch.
         unlockAllMessages = true;
      }

      proxyQueue.setAsynchConsumerThread(null);

      concurrentAccessLock.notifyAll();

      try
      {
         if (cs.performInCallbackActions())
         {
            // Only request the next batch of messages from the messaging engine if the async consumer session
            // is started and is not in the process of stopping, and we are not
            // going to send and unlockAll (which implicitly requests the next batch)
            if (!unlockAllMessages && proxyQueue.getStarted() && !proxyQueue.isStopping()) {
              convHelper.requestNextMessageBatch();
            } else if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                SibTr.debug(this, tc, "Not requesting next message batch: " +
                            "unlockAllMessages=" + unlockAllMessages +
                            "isStarted=" + proxyQueue.getStarted() +
                            "isStopping=" + proxyQueue.isStopping());
            }
         }
      }
      catch(SIException e)
      {
         FFDCFilter.processException(e,
               CLASS_NAME + ".deliverBatch",
               CommsConstants.ASYNCHPQ_DELIVER_04,
               this);
         cs.deliverAsyncException(e);
      }

      // If we handled an exception and might need to redeliver messages,
      // then flow an unlockAll to the messaging engine to unlock all
      // messages and deliver the next batch. We perform this after
      // handling any in-callback actions, so that the messaging engine
      // knows whether or not to send us the next batch.
      // If we handled a close in-callback action (as well as
      // an exception) the unlockAll call will fail. However, there
      // is no method we can call on the consumer session to check whether
      // it is closed or not (isClosed is not properly implemented) so
      // it's safest as an APAR code change to simply accept that in
      // this unlikely circumstance we will generate an FFDC.
      if (unlockAllMessages)
      {
        try
        {
           cs.unlockAll();
        }
        catch(SIException e)
        {
           FFDCFilter.processException(e,
                 CLASS_NAME + ".deliverBatch",
                 CommsConstants.ASYNCHPQ_DELIVER_03,
                 this);
           if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) SibTr.debug(this, tc, "exception thrown");
           if (TraceComponent.isAnyTracingEnabled() && tc.isEventEnabled()) SibTr.event(tc, "exception", e);
        }
      }

      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.exit(this, tc, "deliverBatch");
   }
public void deliverBatch ( int batchSize , short sessionId , ConversationHelper convHelper ) { if ( TraceComponent . isAnyTracingEnabled ( ) && tc . isEntryEnabled ( ) ) SibTr . entry ( this , tc , "deliverBatch" , new Object [ ] { "" + batchSize , "" + sessionId , convHelper } ) ; QueueData queueData = queue . get ( 0 ) ; AsynchConsumerProxyQueue proxyQueue = ( AsynchConsumerProxyQueue ) queueData . getProxyQueue ( ) ; proxyQueue . setAsynchConsumerThread ( Thread . currentThread ( ) ) ; final ConsumerSessionProxy cs = ( ConsumerSessionProxy ) proxyQueue . getDestinationSessionProxy ( ) ; cs . resetCallbackThreadState ( ) ; boolean unlockAllMessages = false ; try { JsMessage [ ] msgs = getBatch ( batchSize , convHelper ) ; LockedMessageEnumerationImpl lockedMsgEnum = new LockedMessageEnumerationImpl ( proxyQueue , this , msgs , Thread . currentThread ( ) , proxyQueue . getLMEOperationMonitor ( ) ) ; if ( TraceComponent . isAnyTracingEnabled ( ) && tc . isDebugEnabled ( ) ) SibTr . debug ( this , tc , " ** About to call user callback **" ) ; proxyQueue . getAsynchConsumerCallback ( ) . consumeMessages ( lockedMsgEnum ) ; if ( TraceComponent . isAnyTracingEnabled ( ) && tc . isDebugEnabled ( ) ) SibTr . debug ( this , tc , " ** User callback has returned  **" ) ; int remainingLockedMessages = lockedMsgEnum . getRemainingMessageCount ( ) ; if ( remainingLockedMessages != 0 ) { if ( TraceComponent . isAnyTracingEnabled ( ) && tc . isDebugEnabled ( ) ) SibTr . debug ( this , tc , "There are still " + remainingLockedMessages + " locked messages! - Unlocking them" ) ; lockedMsgEnum . unlockUnseen ( ) ; } } catch ( Throwable t ) { FFDCFilter . processException ( t , CLASS_NAME + ".deliverBatch" , CommsConstants . ASYNCHPQ_DELIVER_01 , this ) ; if ( TraceComponent . isAnyTracingEnabled ( ) && tc . isDebugEnabled ( ) ) SibTr . debug ( this , tc , "exception thrown" ) ; if ( TraceComponent . isAnyTracingEnabled ( ) && tc . isEventEnabled ( ) ) SibTr . event ( tc , "exception" , t ) ; proxyQueue . setAsynchConsumerThread ( null ) ; cs . deliverAsyncException ( t ) ; unlockAllMessages = true ; } proxyQueue . setAsynchConsumerThread ( null ) ; concurrentAccessLock . notifyAll ( ) ; try { if ( cs . performInCallbackActions ( ) ) { if ( ! unlockAllMessages && proxyQueue . getStarted ( ) && ! proxyQueue . isStopping ( ) ) { convHelper . requestNextMessageBatch ( ) ; } else if ( TraceComponent . isAnyTracingEnabled ( ) && tc . isDebugEnabled ( ) ) { SibTr . debug ( this , tc , "Not requesting next message batch: " + "unlockAllMessages=" + unlockAllMessages + "isStarted=" + proxyQueue . getStarted ( ) + "isStopping=" + proxyQueue . isStopping ( ) ) ; } } } catch ( SIException e ) { FFDCFilter . processException ( e , CLASS_NAME + ".deliverBatch" , CommsConstants . ASYNCHPQ_DELIVER_04 , this ) ; cs . deliverAsyncException ( e ) ; } if ( unlockAllMessages ) { try { cs . unlockAll ( ) ; } catch ( SIException e ) { FFDCFilter . processException ( e , CLASS_NAME + ".deliverBatch" , CommsConstants . ASYNCHPQ_DELIVER_03 , this ) ; if ( TraceComponent . isAnyTracingEnabled ( ) && tc . isDebugEnabled ( ) ) SibTr . debug ( this , tc , "exception thrown" ) ; if ( TraceComponent . isAnyTracingEnabled ( ) && tc . isEventEnabled ( ) ) SibTr . event ( tc , "exception" , e ) ; } } if ( TraceComponent . isAnyTracingEnabled ( ) && tc . isEntryEnabled ( ) ) SibTr . exit ( this , tc , "deliverBatch" ) ; }
This method will deliver a batch of messages to the async callback.

@param batchSize
@param sessionId
@param convHelper