public void handleMessage ( MessageItem msgItem ) throws SIResourceException { if ( TraceComponent . isAnyTracingEnabled ( ) && tc . isEntryEnabled ( ) ) SibTr . entry ( this , tc , "handleMessage" , new Object [ ] { msgItem } ) ; JsMessage jsMsg = msgItem . getMessage ( ) ; int priority = jsMsg . getPriority ( ) . intValue ( ) ; Reliability reliability = jsMsg . getReliability ( ) ; StreamSet streamSet = getStreamSetForMessage ( jsMsg ) ; if ( streamSet == null ) { handleNewStreamID ( msgItem ) ; } else { TargetStream targetStream = null ; synchronized ( streamSet ) { targetStream = ( TargetStream ) streamSet . getStream ( priority , reliability ) ; if ( targetStream == null ) { targetStream = createStream ( streamSet , priority , reliability , streamSet . getPersistentData ( priority , reliability ) ) ; } } // Update the stateStream with this message // The stateStream itself will do Gap detection // Note that the message cannot be written to the MsgStore or // delivered to the ConsumerDispatcher until the stream is // in order. The TargetStream calls back into this class // using the deliverOrderedMessages() method to do this once // it has an ordered stream targetStream . writeValue ( msgItem ) ; } if ( TraceComponent . isAnyTracingEnabled ( ) && tc . isEntryEnabled ( ) ) SibTr . exit ( tc , "handleMessage" ) ; }
Handle a new message by inserting it in to the appropriate target stream . If the stream ID in the message is a new one a flush query will be sent to the source . If the ID has been seen before but the specific stream is not found a new one will be created and added to the stream set .