private void remotePut ( MessageItem msg , SIBUuid8 sourceMEUuid ) throws SIResourceException , SIDiscriminatorSyntaxException { if ( TraceComponent . isAnyTracingEnabled ( ) && tc . isEntryEnabled ( ) ) SibTr . entry ( tc , "remotePut" , new Object [ ] { msg , sourceMEUuid } ) ; // Update the origin map if this is a new stream. // This unsynchronized update is ok until multiple paths // are possible (in which case there could be a race // for the same stream on this method). SIBUuid12 stream = msg . getMessage ( ) . getGuaranteedStreamUUID ( ) ; if ( ! _originStreamMap . containsKey ( stream ) ) _originStreamMap . put ( stream , sourceMEUuid ) ; // First get matches // Match Consumers is only called for PubSub targets. MessageProcessorSearchResults searchResults = matchMessage ( msg ) ; String topic = msg . getMessage ( ) . getDiscriminator ( ) ; //First the remote to remote case (forwarding) HashMap allPubSubOutputHandlers = _destination . getAllPubSubOutputHandlers ( ) ; List matchingPubsubOutputHandlers = searchResults . getPubSubOutputHandlers ( topic ) ; // Check to see if we have any Neighbours if ( allPubSubOutputHandlers != null && allPubSubOutputHandlers . size ( ) > 0 ) { remoteToRemotePut ( msg , allPubSubOutputHandlers , matchingPubsubOutputHandlers ) ; } // Calling getAllPubSubOutputHandlers() locks the handlers // so now we have finished we need to unlock _destination . unlockPubsubOutputHandlers ( ) ; // Now handle remote to local case // If there are any ConsumerDispatchers we will need to save the // list in the Targetstream with the message Set consumerDispatchers = searchResults . getConsumerDispatchers ( topic ) ; // Check to see if we have any matching Neighbours if ( consumerDispatchers != null && consumerDispatchers . size ( ) > 0 ) { // Add the list of consumerDispatchers to the MessageItem // as we will need it in DeliverOrdered messages msg . setSearchResults ( searchResults ) ; // This will add the message to the TargetStream remoteToLocalPut ( msg ) ; } else { // This will add Silence to the TargetStream remoteToLocalPutSilence ( msg ) ; } if ( TraceComponent . isAnyTracingEnabled ( ) && tc . isEntryEnabled ( ) ) SibTr . exit ( tc , "remotePut" ) ; }
The remote put method is driven when the producer is remote to this PubSub Input handler .