package com.tinkerpop.gremlin.process.graph.step.filter;

import com.tinkerpop.gremlin.process.Traversal;
import com.tinkerpop.gremlin.process.util.FastNoSuchElementException;
import com.tinkerpop.gremlin.process.util.TraversalHelper;

import java.util.concurrent.atomic.AtomicLong;

/**
 * @author Marko A. Rodriguez (http://markorodriguez.com)
 */
public final class RangeStep<S> extends FilterStep<S> {

    private final long low;
    private final long high;
    private final AtomicLong counter = new AtomicLong(-1l);

    public RangeStep(final Traversal traversal, final long low, final long high) {
        super(traversal);
        if (low != -1 && high != -1 && low > high) {
            throw new IllegalArgumentException("Not a legal range: [" + low + ", " + high + "]");
        }
        this.low = low;
        this.high = high;


        this.setPredicate(traverser -> {
            final long tempCounter = this.counter.get() + traverser.getBulk();
            if ((this.low == -1 || tempCounter >= this.low) && (this.high == -1 || tempCounter <= this.high)) {
                final long differenceCounter = this.high - tempCounter;
                if (traverser.getBulk() > differenceCounter) {
                    traverser.asAdmin().setBulk(differenceCounter-1);
                }
                this.counter.set(tempCounter);
                return true;
            } else if (this.high != -1 && tempCounter > this.high) {
                final long differenceCounter = tempCounter - this.high;
                if (traverser.getBulk() > differenceCounter) {
                    traverser.asAdmin().setBulk(differenceCounter);
                    this.counter.set(this.counter.get() + differenceCounter);
                    return true;
                } else
                    throw FastNoSuchElementException.instance();
            } else {
                this.counter.set(tempCounter);
                return false;
            }
        });
    }

    @Override
    public void reset() {
        super.reset();
        this.counter.set(-1l);
    }

    @Override
    public String toString() {
        return TraversalHelper.makeStepString(this, this.low, this.high);
    }

    public long getLowRange() {
        return this.low;
    }

    public long getHighRange() {
        return this.high;
    }
}

@@ -511,18 +511,17 @@
 micLong(
--1
+0
 l);%0A%0A   
@@ -896,32 +896,30 @@
     
-final long tempCounter =
+if (this.high != -1 &&
  thi
@@ -938,213 +938,202 @@
 t() 
-+ traverser.getBulk();%0A            if ((this.low == -1 %7C%7C tempCounter %3E= this.low) && (this.high == -1 %7C%7C tempCounter %3C= this.high)) %7B%0A                final long differenceCounter = this.high - tempCounter
+%3E this.high) %7B%0A                throw FastNoSuchElementException.instance();%0A            %7D%0A%0A            if (traverser.getBulk() == 0) %7B%0A                throw FastNoSuchElementException.instance()
 ;%0A  
@@ -1129,32 +1129,40 @@
 tance();%0A       
+     %7D%0A%0A
          if (tra
@@ -1146,36 +1146,48 @@
     %7D%0A%0A         
-if (
+   long avail = 
 traverser.getBul
@@ -1193,164 +1193,256 @@
 lk()
- %3E differenceCounter) %7B%0A                    traverser.asAdmin().setBulk(differenceCounter-1);%0A                %7D%0A                this.counter.set(tempCounter
+;%0A            if (this.counter.get() + avail %3C= this.low) %7B%0A                // Will not surpass the low w/ this traverser. Skip and filter the whole thing.%0A                this.counter.getAndAdd(avail);%0A                traverser.asAdmin().setBulk(0
 );%0A 
@@ -1455,35 +1455,36 @@
          return 
-tru
+fals
 e;%0A            %7D
@@ -1487,264 +1487,222 @@
    %7D
- else if (this.high != -1 && tempCounter %3E this.high) %7B%0A                final long differenceCounter = tempCounter - this.high;%0A                if (traverser.getBulk() %3E differenceCounter) %7B%0A                    traverser.asAdmin().setBulk(differenceC
+%0A%0A            // Skip for the low and trim for the high. Both can happen at once.%0A%0A            long toSkip = 0;%0A            if (this.counter.get() %3C this.low) %7B%0A                toSkip = this.low - this.c
 ounter
+.get(
 );%0A 
@@ -1700,32 +1700,39 @@
 .get();%0A        
+    %7D%0A%0A
             this
@@ -1731,103 +1731,171 @@
     
-this.counter.set(this.counter.get() + differenceCounter);%0A                    return true;%0A    
+long toTrim = 0;%0A            if (this.high != -1 && this.counter.get() + avail %3E this.high) %7B%0A                toTrim = this.counter.get() + avail - this.high - 1;%0A
     
@@ -1903,21 +1903,17 @@
        %7D
- else
+%0A
 %0A       
@@ -1921,81 +1921,46 @@
     
-        throw FastNoSuchElementException.instance();%0A            %7D else %7B
+long toEmit = avail - toSkip - toTrim;
 %0A   
@@ -1960,36 +1960,32 @@
 im;%0A            
-    
 this.counter.set
@@ -1985,55 +1985,85 @@
 ter.
-set(tempCounter);%0A                return false;
+getAndAdd(toSkip + toEmit);%0A            traverser.asAdmin().setBulk(toEmit);%0A
 %0A   
@@ -2063,33 +2063,44 @@
 );%0A%0A            
-%7D
+return true;
 %0A        %7D);%0A   

Fix RangeStep