package iterators;

import org.apache.accumulo.core.data.ByteSequence;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.Range;
import org.apache.accumulo.core.data.Value;
import org.apache.accumulo.core.iterators.IteratorEnvironment;
import org.apache.accumulo.core.iterators.WrappingIterator;

import java.io.IOException;
import java.util.Collection;

public class Counter extends WrappingIterator {
    Key top_key;
    Value top_value;

    @Override
    public boolean hasTop() {
        return top_key != null;
    }

    @Override
    public Key getTopKey() {
        return this.top_key;
    }

    @Override
    public Value getTopValue() {
        return this.top_value;
    }

    @Override
    public void next() throws IOException {
        top_key = null;
        long count = 0L;
        while (this.getSource().hasTop()) {
            top_key = this.getSource().getTopKey();
            Value v = this.getSource().getTopValue();
            //MyPojo pojo = (MyPojo) SerializationUtils.deserialize(v.get());
            //count += pojo.count;
            count++;
            this.getSource().next();
        }
        this.top_value = new Value(Long.toString(count).getBytes());
    }

    @Override
    public void seek(Range range, Collection<ByteSequence> columnFamilies, boolean inclusive) throws IOException {
        this.getSource().seek(range, columnFamilies, inclusive);
        next();
    }

    @Override
    public Counter deepCopy(IteratorEnvironment env) {
        Counter c = new Counter();
        c.setSource(getSource().deepCopy(env));
        return c;
    }
}