import java.io.IOException;

import java.util.List;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.apache.hadoop.hbase.Cell;

import org.apache.hadoop.hbase.CellUtil;

import org.apache.hadoop.hbase.filter.FilterBase;

import org.apache.hadoop.hbase.util.Bytes;

import com.vividsolutions.jts.geom.Coordinate;

import com.vividsolutions.jts.geom.Geometry;

import com.vividsolutions.jts.geom.GeometryFactory;

public class WithinFilter extends FilterBase {

    static final byte[] TABLE = "wifi".getBytes();

    static final byte[] FAMILY = "a".getBytes();

    static final byte[] ID = "id".getBytes();

    static final byte[] X_COL = "lon".getBytes();

    static final byte[] Y_COL = "lat".getBytes();

    static final Log LOG = LogFactory.getLog(WithinFilter.class);

    final GeometryFactory factory = new GeometryFactory();

    Geometry query = null;

    boolean exclude = false;

    public WithinFilter() {

    }

    public WithinFilter(Geometry query) {

        this.query = query;

    }

    //遍历每行每个列族的每个KeyValue的方法可能很慢，如果可以，HBase会优化对filterRow的调用

    @Override

    public boolean hasFilterRow(){

        return true;

    }        

    //根据column family, column qualifier 或者 column value进行过滤

    @Override

    public ReturnCode filterKeyValue(Cell cell) throws IOException {

        byte[] qualname = CellUtil.cloneQualifier(cell);

        if(Bytes.equals(qualname, Bytes.toBytes("不需要的qualifier名"))) //例如可以处理密码，并且将密码跳过不反回到客户端

            return ReturnCode.SKIP;

        return ReturnCode.INCLUDE;

    }

    //根据经纬度过滤，符合要求的为在区域内的点

    @Override

    public void filterRowCells(List<Cell> celllist) throws IOException{

        double lon = Double.NaN;

        double lat = Double.NaN;

        for(Cell cell : celllist){

            if(Bytes.equals(CellUtil.cloneQualifier(cell), X_COL)){

                lon = Double.parseDouble(new String(CellUtil.cloneValue(cell)));

            }

            if(Bytes.equals(CellUtil.cloneQualifier(cell), Y_COL)){

                lat = Double.parseDouble(new String(CellUtil.cloneValue(cell)));

            }

        }

        Coordinate coord = new Coordinate(lon,lat);

        Geometry point = factory.createPoint(coord); //创建Point

        if(!query.contains(point)){ //测试是否包含该点

            this.exclude = true;

        }

    }

    //如果某一行没有落在查询边界想要排除它是，需要设置exclude标志。

    @Override

    public boolean filterRow() {

        if (LOG.isDebugEnabled())

            LOG.debug("filter applied. " + (this.exclude ? "rejecting" : "keeping"));

        return this.exclude;

    }

    @Override

    public void reset() {

        this.exclude = false;

    }

}