/*
 * 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