protected void attemptFlush ( ) throws SIRollbackException , SIConnectionLostException , SIResourceException , SIErrorException { if ( TraceComponent . isAnyTracingEnabled ( ) && tc . isEntryEnabled ( ) ) SibTr . entry ( tc , "attemptFlush" ) ; //The FlushComplete callback FlushComplete callback = null ; //The old set of streams which has just been flushed StreamSet oldStreamSet = null ; // Actions for a flushable stream. Note that we synchronize // up to step 4 to avoid races with threads handling control // messages. // 1. Cache and unset flushInProgress. // 2. Create a new stream ID // 3. Remove any data structures associated with the stream. // 4. Invoke the callback on the cached FlushComplete reference. synchronized ( this ) { // Resolve race between control message handlers who both // see a flushable stream. // flushInProgress was set by startFlush if ( flushInProgress == null ) { if ( TraceComponent . isAnyTracingEnabled ( ) && tc . isEntryEnabled ( ) ) SibTr . exit ( tc , "attemptFlush" ) ; return ; } // Also verify that the stream really is flushable in case // a thread handling a control message called us with a // stale source stream reference. if ( ! flushable ( ) ) { if ( TraceComponent . isAnyTracingEnabled ( ) && tc . isEntryEnabled ( ) ) SibTr . exit ( tc , "attemptFlush" ) ; return ; } // Now cache the callback. When we exit the synchronize block, // the flush will have officially ended. callback = flushInProgress ; flushInProgress = null ; // Cleanup the data structures and null out the // existing streamSet so we force creation of a new // set on the next message. oldStreamSet = streamSet ; streamSet = null ; if ( oldStreamSet != null ) { oldStreamSet . remove ( ) ; } } // Send out a flushed message. If this fails, make sure we get // to at least invoke the callback. try { // The Cellule argument is null as it is ignored by our parent handler // this flush message should be broadcast downstream. if ( oldStreamSet != null ) downControl . sendFlushedMessage ( null , oldStreamSet . getStreamID ( ) ) ; } catch ( Exception e ) { // FFDC FFDCFilter . processException ( e , "com.ibm.ws.sib.processor.gd.SourceStreamManager.attemptFlush" , "1:1158:1.102" , this ) ; // Note that it doesn't make much sense to throw an exception here since // this is a callback from stream array map. if ( TraceComponent . isAnyTracingEnabled ( ) && tc . isEntryEnabled ( ) ) SibTr . exit ( tc , "attemptFlush" , e ) ; return ; } finally { // Invoke the calback in the finally clause so we don't holdup // whoever's waiting for the flush to complete. callback . flushComplete ( destinationHandler ) ; } if ( TraceComponent . isAnyTracingEnabled ( ) && tc . isEntryEnabled ( ) ) SibTr . exit ( tc , "attemptFlush" ) ; }
This method will be called periodically to check if a stream is flushable and if so will actually flush it .