/*
    * Licensed to the Apache Software Foundation (ASF) under one
    * or more contributor license agreements.  See the NOTICE file
    * distributed with this work for additional information
    * regarding copyright ownership.  The ASF licenses this file
    * to you under the Apache License, Version 2.0 (the
    * "License"); you may not use this file except in compliance
    * with the License.  You may obtain a copy of the License at
    *
   *     http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  package org.apache.hadoop.chukwa.util;
  
  import java.nio.charset.Charset;
  import java.util.ArrayList;
  import java.util.List;
  import java.util.regex.Pattern;
  import java.util.regex.PatternSyntaxException;
  
  import org.apache.commons.lang.ArrayUtils;
  import org.apache.hadoop.chukwa.Chunk;
  import org.apache.hadoop.chukwa.extraction.engine.RecordUtil;
  import org.apache.hadoop.chukwa.util.RegexUtil.CheckedPatternSyntaxException;
  import org.apache.log4j.Logger;
  
  
  public class Filter {
  
    static Logger log = Logger.getLogger(Filter.class);
  
    private static final String[] SEARCH_TARGS = 
    {"datatype", "name", "host", "cluster", "content"};
    static final String SEPARATOR="&";
    
    private static class SearchRule {
      Pattern p;
      String targ;
      
      SearchRule(Pattern p, String t) {
        this.p = p;
        this.targ = t;
      }
      
      boolean matches(Chunk chunk) {
        if(targ.equals("datatype")) {
          return p.matcher(chunk.getDataType()).matches();
        } else if(targ.equals("name")) {
          return p.matcher(chunk.getStreamName()).matches();
        } else if(targ.equals("host")) {
          return p.matcher(chunk.getSource()).matches();
        } else if(targ.equals("cluster")) {
          String cluster = RecordUtil.getClusterName(chunk);
          return p.matcher(cluster).matches();
        } else if(targ.equals("content")) {
          String content = new String(chunk.getData(), Charset.forName("UTF-8"));
          return p.matcher(content).matches();
        } else if(targ.startsWith("tags.")) {
          String tagName = targ.substring("tags.".length());
          if (!RegexUtil.isRegex(tagName)) {
            log.warn("Error parsing 'tagName' as a regex: "
                + RegexUtil.regexError(tagName));
            return false;
          }
          String tagVal = chunk.getTag(tagName);
          if(tagVal == null)
            return false;
          return p.matcher(tagVal).matches();
        } else { 
          assert false: "unknown target: " +targ;
          return false;
        }
      }
      
      public String toString() {
        return targ + "=" +p.toString();
      } 
    }
  
    List<SearchRule> compiledPatterns;
      
    public Filter(String listOfPatterns) throws CheckedPatternSyntaxException {
      compiledPatterns = new ArrayList<SearchRule>();
      //FIXME: could escape these
      String[] patterns = listOfPatterns.split(SEPARATOR);
      for(String p: patterns) {
        int equalsPos = p.indexOf('=');
        
        if(equalsPos < 0 || equalsPos > (p.length() -2)) {
          throw new CheckedPatternSyntaxException(
              "pattern must be of form targ=pattern", p, -1);
        }
        
        String targ = p.substring(0, equalsPos);
       if(!targ.startsWith("tags.") && !ArrayUtils.contains(SEARCH_TARGS, targ)) {
         throw new CheckedPatternSyntaxException(
             "pattern doesn't start with recognized search target", p, -1);
       }
       
       String regex = p.substring(equalsPos+1);
       if (!RegexUtil.isRegex(regex)) {
           throw new CheckedPatternSyntaxException(RegexUtil.regexException(regex));
       }
 
       Pattern pat = Pattern.compile(regex, Pattern.DOTALL);
       compiledPatterns.add(new SearchRule(pat, targ));
     }
   }
 
   public boolean matches(Chunk chunk) {
     for(SearchRule r: compiledPatterns) {
       if(!r.matches(chunk))
         return false;
     }
     return true;
   }
   
   int size() {
     return compiledPatterns.size();
   }
   
   public String toString() {
     StringBuilder sb = new StringBuilder();
     sb.append(compiledPatterns.get(0));
     for(int i=1; i < compiledPatterns.size(); ++i) {
       sb.append(" & ");
       sb.append(compiledPatterns.get(i));
     }
     return sb.toString();
   }
   
   private static final class MatchAll extends Filter {
     public MatchAll() throws CheckedPatternSyntaxException {
       super("datatype=.*");
     }
     
     public boolean matches(Chunk c) {
       return true;
     }
     
     public String toString() {
       return "ALL";
     }
   } 
 
   public static final Filter ALL = newMatchAll();
   private static Filter newMatchAll() {
     try {
       return new MatchAll();
     } catch (CheckedPatternSyntaxException e) {
       throw new RuntimeException("Illegal MatchAll regular expression.", e);
     }
   }
   
 }//end class