package continuum.core.slice;

import continuum.atom.Atom;
import continuum.atom.AtomID;
import continuum.slab.Iterator;
import continuum.slice.*;

import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.util.stream.Stream;

/**
 * Base scanner using filters and collectors
 */
public class AScanner implements Scanner {

    private Iterator<Atom> iterator;

    private byte[] previousScan = null;

    /**
     * {@inheritDoc}
     */
    @Override
    public Slice slice(Scan scan) throws Exception {
        Collector collector = Collectors.forScan(scan);
        Filter filter = Filters.forScan(scan); // TODO: filter fields and values
        boolean decode = decodeBody(collector);
        byte[] start = previousScan == null ? scan.ID().bytes() : previousScan;
        iterator.seek(start);
        Atom atom = iterator.valid() ? iterator.get() : null;
        boolean stop = false;
        while (!stop && atom != null) {
            switch (filter.filter(atom)) {
                case CONTINUE:          if (decode) collect(collector);
                                        else collectID(collector);
                                        break;
                case SKIP:              break;
                case STOP: default:     stop = true;
                                        break;
            }
            atom = iterate(iterator);
        }

        previousScan = iterator.ID().bytes();

        return collector.slice();
    }

    @Override
    public Stream<Slice> stream(Scan scan) throws Exception {
        return null;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void iterator(Iterator<Atom> iterator) {
        this.iterator = iterator;
    }

    private Atom iterate(Iterator<Atom> iterator) {
        iterator.next();
        if (!iterator.valid()) return null;
        return iterator.get();
    }

    private boolean decodeBody(Collector<?> collector) {
        Class<?> clazz = collector.getClass();
        Type[] types = clazz.getGenericInterfaces();
        Type type = ((ParameterizedType)types[0]).getActualTypeArguments()[0];

        if (Atom.class.getTypeName().equals(type.getTypeName()))
            return true;
        else if (AtomID.class.getTypeName().equals(type.getTypeName()))
            return false;
        throw new Error("Can not collect: " + type);
    }

    private void collect(Collector<Atom> collector) {
        collector.collect(iterator.get());
    }

    private void collectID(Collector<AtomID> collector) {
        collector.collect(iterator.ID());
    }
}

@@ -212,24 +212,98 @@
 flect.Type;%0A
+import java.util.concurrent.Callable;%0Aimport java.util.function.Supplier;%0A
 import java.
@@ -633,25 +633,30 @@
 ector = 
-C
+scan.c
 ollector
 s.forSca
@@ -647,30 +647,17 @@
 ollector
-s.forScan(scan
+(
 );%0A     
@@ -1636,28 +1636,425 @@
      return 
-null
+Stream.generate(AScanner.wrap(() -%3E slice(scan)));%0A    %7D%0A%0A%0A    public static %3CT%3E Supplier%3CT%3E wrap(Callable%3CT%3E callable) %7B%0A        return () -%3E %7B%0A            try %7B%0A                return callable.call();%0A            %7D%0A            catch (RuntimeException e) %7B%0A                throw e;%0A            %7D%0A            catch (Exception e) %7B%0A                throw new RuntimeException(e);%0A            %7D%0A        %7D
 ;%0A    %7D%0A%0A   

prepare for streaming