package iterators;

import domain.MyPojo;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.Value;
import org.apache.accumulo.core.iterators.IteratorEnvironment;
import org.apache.accumulo.core.iterators.WrappingIterator;
import org.apache.commons.lang.SerializationUtils;

import java.io.IOException;

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() {
        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;
            try {
                this.getSource().next();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        this.top_value = new Value(Long.toString(count).getBytes());
    }

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