public boolean addMessage ( SIMPMessage msgItem ) throws SIResourceException { if ( TraceComponent . isAnyTracingEnabled ( ) && tc . isEntryEnabled ( ) ) SibTr . entry ( tc , "addMessage" , new Object [ ] { msgItem } ) ; // Get the JsMessage as we need to update the Guaranteed fields JsMessage jsMsg = msgItem . getMessage ( ) ; StreamSet streamSet = getStreamSet ( ) ; // Stamp the message with the stream ID msgItem . setGuaranteedStreamUuid ( streamSet . getStreamID ( ) ) ; // NOTE: no synchronization with flush necessary here since startFlush // is not supposed to be called until all competing producer threads have // exited. Reliability reliability = msgItem . getReliability ( ) ; if ( reliability == Reliability . BEST_EFFORT_NONPERSISTENT ) { addBestEffortMessage ( msgItem ) ; if ( TraceComponent . isAnyTracingEnabled ( ) && tc . isEntryEnabled ( ) ) SibTr . exit ( tc , "addMessage" , false ) ; return false ; } else { SourceStream sourceStream = null ; int priority = msgItem . getPriority ( ) ; synchronized ( streamSet ) { sourceStream = ( SourceStream ) streamSet . getStream ( priority , reliability ) ; if ( sourceStream == null ) { sourceStream = createStream ( streamSet , priority , reliability , streamSet . getPersistentData ( priority , reliability ) , false ) ; } } synchronized ( sourceStream ) { long tick = this . messageProcessor . nextTick ( ) ; //Set a unique id in the message if explicitly told to or //if one has not already been set if ( msgItem . getRequiresNewId ( ) || jsMsg . getSystemMessageId ( ) == null ) { jsMsg . setSystemMessageSourceUuid ( this . messageProcessor . getMessagingEngineUuid ( ) ) ; jsMsg . setSystemMessageValue ( tick ) ; msgItem . setRequiresNewId ( false ) ; } jsMsg . setGuaranteedValueEndTick ( tick ) ; jsMsg . setGuaranteedValueValueTick ( tick ) ; jsMsg . setGuaranteedValueRequestedOnly ( false ) ; jsMsg . setGuaranteedValueStartTick ( sourceStream . getLastMsgAdded ( ) + 1 ) ; jsMsg . setGuaranteedValueCompletedPrefix ( sourceStream . getCompletedPrefix ( ) ) ; sourceStream . writeUncommitted ( msgItem ) ; } // end sync if ( TraceComponent . isAnyTracingEnabled ( ) && tc . isEntryEnabled ( ) ) SibTr . exit ( tc , "addMessage" , true ) ; return true ; } }
Add a message in to the appropriate source stream . This will create a stream if one does not exist and set any appropriate fields in the message