package com.tinkerpop.gremlin.elastic.process.graph.traversal.sideEffect;

import com.tinkerpop.gremlin.elastic.ElasticService;
import com.tinkerpop.gremlin.elastic.structure.ElasticGraph;
import com.tinkerpop.gremlin.process.Traversal;
import com.tinkerpop.gremlin.process.graph.step.sideEffect.GraphStep;
import com.tinkerpop.gremlin.process.graph.util.HasContainer;
import com.tinkerpop.gremlin.structure.Edge;
import com.tinkerpop.gremlin.structure.Element;
import com.tinkerpop.gremlin.structure.Vertex;
import org.elasticsearch.common.geo.builders.ShapeBuilder;
import org.elasticsearch.common.xcontent.XContentParser;
import org.elasticsearch.common.xcontent.json.JsonXContent;
import org.elasticsearch.index.query.BoolFilterBuilder;
import org.elasticsearch.index.query.FilterBuilder;
import org.elasticsearch.index.query.FilterBuilders;
import org.elasticsearch.index.query.GeoShapeFilterBuilder;
import sun.reflect.generics.reflectiveObjects.NotImplementedException;

import java.util.*;

public class ElasticGraphStep<E extends Element> extends GraphStep<E> {
    public final List<HasContainer> hasContainers = new ArrayList<HasContainer>();
    ElasticService elasticService;


    public ElasticGraphStep(final Traversal traversal, final ElasticGraph graph, final Class<E> returnClass, Optional<String> label, final Object... ids) {
        super(traversal, graph, returnClass, ids);
        if (label.isPresent()) this.setLabel(label.get());
        this.setIteratorSupplier(() -> (Iterator<E>) (Vertex.class.isAssignableFrom(this.returnClass) ? this.vertices() : this.edges()));
        this.elasticService = graph.elasticService;
    }

    private Iterator<? extends Vertex> vertices() {
        return elasticService.searchVertices(getFilter(), label());
    }

    private Iterator<? extends Edge> edges() {
        return elasticService.searchEdges(getFilter(), label());
    }

    private FilterBuilder getFilter() {
        if (this.hasContainers.size() == 0 && this.getIds().length == 0)
            return null;

        BoolFilterBuilder boolFilterBuilder = FilterBuilders.boolFilter();
        for (HasContainer has : this.hasContainers) {
            if (has.predicate.toString().equals("eq")) {
                if(has.key.equals("~label"))
                    this.setLabel(has.value.toString());
                else
                    boolFilterBuilder = boolFilterBuilder.must(FilterBuilders.termFilter(has.key, has.value));
            }
            else if(has.predicate.toString().equals("without"))
                boolFilterBuilder = boolFilterBuilder.mustNot(FilterBuilders.existsFilter(has.key));
            else if(has.predicate instanceof Geo)
                boolFilterBuilder = boolFilterBuilder.must(new GeoShapeFilterBuilder(has.key, GetShapeBuilder(has.value), ((Geo)has.predicate).getRelation()));
            else throw new NotImplementedException();
        }
        if (this.getIds().length > 0) {
            String[] stringIds = Arrays.copyOf(getIds(), getIds().length, String[].class);
            boolFilterBuilder = boolFilterBuilder.must(FilterBuilders.idsFilter().addIds(stringIds));
        }

        return boolFilterBuilder;
    }

    private ShapeBuilder GetShapeBuilder(Object object)  {
        try {
            String geoJson = (String) object;
            XContentParser parser = JsonXContent.jsonXContent.createParser(geoJson);
            parser.nextToken();

            return ShapeBuilder.parse(parser);
        }
        catch (Exception e){
            return null;
        }
    }

    private String label(){
        return this.getLabel().isPresent() ? this.getLabel().get() : null;
    }
}

@@ -405,107 +405,9 @@
 ure.
-Edge;%0Aimport com.tinkerpop.gremlin.structure.Element;%0Aimport com.tinkerpop.gremlin.structure.Vertex
+*
 ;%0Aim
@@ -2077,17 +2077,16 @@
       if
- 
 (has.pre
@@ -2087,35 +2087,149 @@
 as.predicate
-.toString()
+ instanceof Compare) %7B%0A                String predicateString = has.predicate.toString();%0A                if (predicateString
 .equals(%22eq%22
@@ -2229,24 +2229,28 @@
 ls(%22eq%22)) %7B%0A
+    
             
@@ -2251,24 +2251,25 @@
           if
+ 
 (has.key.equ
@@ -2279,24 +2279,28 @@
 (%22~label%22))%0A
+    
             
@@ -2364,13 +2364,21 @@
     
+    
 else%0A
+    
     
@@ -2500,74 +2500,978 @@
     
+ 
-%7D%0A            else if(has.predicate.toString().equals(%22
+   %7D else if (predicateString.equals(%22neq%22))%0A                    boolFilterBuilder = boolFilterBuilder.mustNot(FilterBuilders.termFilter(has.key, has.value));%0A                else if (predicateString.equals(%22gt%22))%0A                    boolFilterBuilder.must(FilterBuilders.rangeFilter(has.key).gt(has.value));%0A                else if (predicateString.equals(%22gte%22))%0A                    boolFilterBuilder.must(FilterBuilders.rangeFilter(has.key).gte(has.value));%0A                else if (predicateString.equals(%22lt%22))%0A                    boolFilterBuilder.must(FilterBuilders.rangeFilter(has.key).lt(has.value));%0A                else if (predicateString.equals(%22lte%22))%0A                    boolFilterBuilder.must(FilterBuilders.rangeFilter(has.key).lte(has.value));%0A                else%0A                    throw new NotImplementedException();%0A            %7D%0A            else if(has.predicate instanceof Contains)%7B%0A                if(has.predicate == Contains.
 without
-%22)
 )%0A
+    
     
@@ -3563,24 +3563,198 @@
 (has.key));%0A
+                else if(has.predicate == Contains.within)%0A                    boolFilterBuilder = boolFilterBuilder.must(FilterBuilders.existsFilter(has.key));%0A            %7D%0A
             
@@ -3919,16 +3919,17 @@
 , ((Geo)
+ 
 has.pred

support gt,gte,lt,lte and has with one field