public List processAck ( ControlAck ackMsg , long ackPrefix ) throws SIRollbackException , SIConnectionLostException , SIResourceException { if ( TraceComponent . isAnyTracingEnabled ( ) && tc . isEntryEnabled ( ) ) SibTr . entry ( tc , "processAck" , new Object [ ] { ackMsg , new Long ( ackPrefix ) } ) ; List indexList = null ; // Short circuit if incoming message has wrong stream ID if ( ! hasStream ( ackMsg . getGuaranteedStreamUUID ( ) ) ) { // Bogus stream, ignore the message // Return a null indexList to ensure no further processing is // attempted by the caller if ( TraceComponent . isAnyTracingEnabled ( ) && tc . isEntryEnabled ( ) ) SibTr . exit ( tc , "processAck" , new Object [ ] { "unknown stream ID - returning null array list(message ignored)" } ) ; return indexList ; } int priority = ackMsg . getPriority ( ) . intValue ( ) ; Reliability reliability = ackMsg . getReliability ( ) ; // There is no stream for BestEffort non persistent messages if ( reliability . compareTo ( Reliability . BEST_EFFORT_NONPERSISTENT ) > 0 ) { StreamSet streamSet = getStreamSet ( ) ; SourceStream sourceStream = ( SourceStream ) streamSet . getStream ( priority , reliability ) ; // If there's no source stream then priority was bogus so ignore the ack if ( sourceStream == null ) { // Return a null indexList to ensure no further processing is // attempted by the caller if ( TraceComponent . isAnyTracingEnabled ( ) && tc . isEntryEnabled ( ) ) SibTr . exit ( tc , "processAck" , new Object [ ] { "unknown priority - returning null array list(message ignored)" } ) ; return indexList ; } // If this increases the finality prefix then // update it and delete the acked messages from the ItemStream long completedPrefix = sourceStream . getAckPrefix ( ) ; if ( ackPrefix > completedPrefix ) { // Update the completedPrefix and the oack value for the stream // returns a lit of the itemStream ids of the newly Acked messages // which we can then remove from the itemStream indexList = sourceStream . writeAckPrefix ( ackPrefix ) ; } } else { // We didn't expect an Ack for a best effort message so write an entry // to the trace log indicating that it happened if ( TraceComponent . isAnyTracingEnabled ( ) && tc . isDebugEnabled ( ) ) SibTr . debug ( tc , "Unexpected Ack message for BEST_EFFORT_NONPERSISTENT message " ) ; } if ( TraceComponent . isAnyTracingEnabled ( ) && tc . isEntryEnabled ( ) ) SibTr . exit ( tc , "processAck" , indexList ) ; return indexList ; }
Process an Ack control message but use the specified ack prefix to decide if the ack prefix can be advanced .