void processAck(ControlAck ackMsg) 
  throws SIRollbackException, 
         SIConnectionLostException, 
         SIResourceException, 
         SIErrorException
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc, "processAck",
        new Object[] { ackMsg });

    List indexList = null;
    SourceStream batchListener = null;
    synchronized(sourceStreamManager)
    {
      indexList = sourceStreamManager.processAck(ackMsg);
      batchListener = sourceStreamManager.getBatchListener(ackMsg);
    }
    long index = 0;
    MessageItem msgItem = null;

    if(indexList != null)
    {         
      TransactionCommon tran = sourceBatchHandler.registerInBatch();
      // A marker to indicate how far through the method we get. 
      boolean messageAddCall = false;      
      
      try
      {      
        TickRange tr = null;
        for (int i = 0; i < indexList.size(); i++)
        {
          tr = (TickRange)indexList.get(i);
          batchListener.addToBatchList(tr);
          
          index = tr.itemStreamIndex;
          try
          {
            msgItem =
              (MessageItem)transmissionItemStream.findById(index);
            
            //it is possible that this is null in the case of a message
            //deleted via the Admin panels 
            if( msgItem != null)
            {  
              Transaction msTran = messageProcessor.resolveAndEnlistMsgStoreTransaction(tran);
              msgItem.remove(msTran, msgItem.getLockID());
            }  
          }
          catch (MessageStoreException e)
          {
            // MessageStoreException shouldn't occur so FFDC.
            FFDCFilter.processException(
              e,
              "com.ibm.ws.sib.processor.impl.PtoPOutputHandler.processAck",
              "1:1191:1.241",
              this);

            SibTr.exception(tc, e);
            SibTr.error(tc, "INTERNAL_MESSAGING_ERROR_CWSIP0002",
              new Object[] {
                "com.ibm.ws.sib.processor.impl.PtoPOutputHandler",
                "1:1198:1.241",
                e });

            if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
              SibTr.exit(tc, "processAck", e);

            throw new SIResourceException(
              nls.getFormattedMessage(
                "INTERNAL_MESSAGING_ERROR_CWSIP0002",
                new Object[] {
                  "com.ibm.ws.sib.processor.impl.PtoPOutputHandler",
                  "1:1209:1.241",
                  e },
                null),
              e);
          }
        }
        
        // Indicate that we are about to unlock using messagesAdded.
        messageAddCall = true;
        sourceBatchHandler.messagesAdded(indexList.size(), batchListener);      
      }
      finally
      {
        // Before exiting this method, need to unlock the batch handler if it was locked.
        if (!messageAddCall)
        try
        {      
          sourceBatchHandler.messagesAdded(0);
        }
        catch (SIResourceException e)
        {
          // No FFDC code needed, This will allow for any exceptions that were thrown to
          // be rethrown instead of overiding with a batch handler error.
          if (TraceComponent.isAnyTracingEnabled() && tc.isEventEnabled())
            SibTr.exception(tc, e);               
        }
        
      }
    }
    
    // Before we exit, let the source stream manager take care of any pending flushes
    sourceStreamManager.attemptFlushIfNecessary();

    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "processAck");
  }
void processAck ( ControlAck ackMsg ) throws SIRollbackException , SIConnectionLostException , SIResourceException , SIErrorException { if ( TraceComponent . isAnyTracingEnabled ( ) && tc . isEntryEnabled ( ) ) SibTr . entry ( tc , "processAck" , new Object [ ] { ackMsg } ) ; List indexList = null ; SourceStream batchListener = null ; synchronized ( sourceStreamManager ) { indexList = sourceStreamManager . processAck ( ackMsg ) ; batchListener = sourceStreamManager . getBatchListener ( ackMsg ) ; } long index = 0 ; MessageItem msgItem = null ; if ( indexList != null ) { TransactionCommon tran = sourceBatchHandler . registerInBatch ( ) ; boolean messageAddCall = false ; try { TickRange tr = null ; for ( int i = 0 ; i < indexList . size ( ) ; i ++ ) { tr = ( TickRange ) indexList . get ( i ) ; batchListener . addToBatchList ( tr ) ; index = tr . itemStreamIndex ; try { msgItem = ( MessageItem ) transmissionItemStream . findById ( index ) ; if ( msgItem != null ) { Transaction msTran = messageProcessor . resolveAndEnlistMsgStoreTransaction ( tran ) ; msgItem . remove ( msTran , msgItem . getLockID ( ) ) ; } } catch ( MessageStoreException e ) { FFDCFilter . processException ( e , "com.ibm.ws.sib.processor.impl.PtoPOutputHandler.processAck" , "1:1191:1.241" , this ) ; SibTr . exception ( tc , e ) ; SibTr . error ( tc , "INTERNAL_MESSAGING_ERROR_CWSIP0002" , new Object [ ] { "com.ibm.ws.sib.processor.impl.PtoPOutputHandler" , "1:1198:1.241" , e } ) ; if ( TraceComponent . isAnyTracingEnabled ( ) && tc . isEntryEnabled ( ) ) SibTr . exit ( tc , "processAck" , e ) ; throw new SIResourceException ( nls . getFormattedMessage ( "INTERNAL_MESSAGING_ERROR_CWSIP0002" , new Object [ ] { "com.ibm.ws.sib.processor.impl.PtoPOutputHandler" , "1:1209:1.241" , e } , null ) , e ) ; } } messageAddCall = true ; sourceBatchHandler . messagesAdded ( indexList . size ( ) , batchListener ) ; } finally { if ( ! messageAddCall ) try { sourceBatchHandler . messagesAdded ( 0 ) ; } catch ( SIResourceException e ) { if ( TraceComponent . isAnyTracingEnabled ( ) && tc . isEventEnabled ( ) ) SibTr . exception ( tc , e ) ; } } } sourceStreamManager . attemptFlushIfNecessary ( ) ; if ( TraceComponent . isAnyTracingEnabled ( ) && tc . isEntryEnabled ( ) ) SibTr . exit ( tc , "processAck" ) ; }
/* The messageAddCall boolean needs to be set after a right before a call to
the batch handler messagesAdded so that in the result of an exception being
thrown, the handler is unlocked.