public List < Stream > match ( Message message ) { final Set < Stream > result = Sets . newHashSet ( ) ; final Set < Stream > blackList = Sets . newHashSet ( ) ; for ( final Rule rule : rulesList ) { if ( blackList . contains ( rule . getStream ( ) ) ) { continue ; } final StreamRule streamRule = rule . getStreamRule ( ) ; final StreamRuleType streamRuleType = streamRule . getType ( ) ; final Stream . MatchingType matchingType = rule . getMatchingType ( ) ; if ( ! ruleTypesNotNeedingFieldPresence . contains ( streamRuleType ) && ! message . hasField ( streamRule . getField ( ) ) ) { if ( matchingType == Stream . MatchingType . AND ) { result . remove ( rule . getStream ( ) ) ; // blacklist stream because it can't match anymore blackList . add ( rule . getStream ( ) ) ; } continue ; } final Stream stream ; if ( streamRuleType != StreamRuleType . REGEX ) { stream = rule . match ( message ) ; } else { stream = rule . matchWithTimeOut ( message , streamProcessingTimeout , TimeUnit . MILLISECONDS ) ; } if ( stream == null ) { if ( matchingType == Stream . MatchingType . AND ) { result . remove ( rule . getStream ( ) ) ; // blacklist stream because it can't match anymore blackList . add ( rule . getStream ( ) ) ; } } else { result . add ( stream ) ; if ( matchingType == Stream . MatchingType . OR ) { // blacklist stream because it is already matched blackList . add ( rule . getStream ( ) ) ; } } } final Stream defaultStream = defaultStreamProvider . get ( ) ; boolean alreadyRemovedDefaultStream = false ; for ( Stream stream : result ) { streamMetrics . markIncomingMeter ( stream . getId ( ) ) ; if ( stream . getRemoveMatchesFromDefaultStream ( ) ) { if ( alreadyRemovedDefaultStream || message . removeStream ( defaultStream ) ) { alreadyRemovedDefaultStream = true ; if ( LOG . isTraceEnabled ( ) ) { LOG . trace ( "Successfully removed default stream <{}> from message <{}>" , defaultStream . getId ( ) , message . getId ( ) ) ; } } else { if ( LOG . isWarnEnabled ( ) ) { LOG . warn ( "Couldn't remove default stream <{}> from message <{}>" , defaultStream . getId ( ) , message . getId ( ) ) ; } } } } // either the message stayed on the default stream, in which case we mark that stream's throughput, // or someone removed it, in which case we don't mark it. if ( ! alreadyRemovedDefaultStream ) { streamMetrics . markIncomingMeter ( defaultStream . getId ( ) ) ; } return ImmutableList . copyOf ( result ) ; }
Returns a list of matching streams for the given message .