docs: 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);
      }