package org.embulk.parser.joni;

import com.google.common.base.Optional;
import org.embulk.config.Config;
import org.embulk.config.ConfigDefault;
import org.embulk.config.ConfigDiff;
import org.embulk.config.ConfigSource;
import org.embulk.config.Task;
import org.embulk.config.TaskSource;
import org.embulk.spi.*;

import org.embulk.spi.json.JsonParser;
import org.embulk.spi.time.TimestampParser;
import org.embulk.spi.util.Timestamps;

import org.embulk.spi.util.LineDecoder;
import org.joni.*;
import org.msgpack.value.ValueFactory;
import org.slf4j.Logger;

import org.jcodings.specific.UTF8Encoding;

import java.nio.charset.StandardCharsets;
import java.util.Iterator;
import java.util.Locale;
import org.msgpack.value.Value;


public class JoniParserPlugin
        implements ParserPlugin
{
    private static final Logger logger = Exec.getLogger(JoniParserPlugin.class);

    public interface TypecastColumnOption
            extends Task
    {
        @Config("typecast")
        @ConfigDefault("null")
        public Optional<Boolean> getTypecast();
    }

    public interface PluginTask
            extends Task,LineDecoder.DecoderTask, TimestampParser.Task
    {
        @Config("columns")
        SchemaConfig getColumns();

        @Config("stop_on_invalid_record")
        @ConfigDefault("false")
        boolean getStopOnInvalidRecord();

        @Config("format")
        String getFormat();

        @Config("default_typecast")
        @ConfigDefault("true")
        Boolean getDefaultTypecast();

    }

    @Override
    public void transaction(ConfigSource config, ParserPlugin.Control control)
    {
        PluginTask task = config.loadConfig(PluginTask.class);

        Schema schema = task.getColumns().toSchema();

        validateSchema(task,schema);

        control.run(task.dump(), schema);
    }

    @Override
    public void run(TaskSource taskSource, Schema schema,
            FileInput input, PageOutput output)
    {
        PluginTask task = taskSource.loadTask(PluginTask.class);
        LineDecoder lineDecoder = new LineDecoder(input, task);
        PageBuilder pageBuilder = new PageBuilder(Exec.getBufferAllocator(), schema, output);
        TimestampParser[] timestampParsers = Timestamps.newTimestampColumnParsers(task, task.getColumns());

        ColumnVisitorImpl visitor = new ColumnVisitorImpl(task, schema, pageBuilder, timestampParsers);

        Regex regex = buildRegex(task);

        while (input.nextFile()) {
            while (true) {
                String line = lineDecoder.poll();
//                logger.debug(String.format(Locale.ENGLISH,"line = %s",line));
                if (line == null) {
                    break;
                }
                Matcher matcher = regex.matcher(line.getBytes());
                int result = matcher.search(0,line.getBytes().length,Option.DEFAULT);

                if( result != -1 ){
                     Region region = matcher.getEagerRegion();
                     for(Iterator<NameEntry> entry = regex.namedBackrefIterator(); entry.hasNext(); ){
                         NameEntry e = entry.next();
                         String name = captureName(e);

                         int number = e.getBackRefs()[0];
                         int begin = region.beg[number];
                         int end = region.end[number];
                         String strValue = new String(line.getBytes(StandardCharsets.UTF_8), begin, end - begin, StandardCharsets.UTF_8);

                         logger.debug(String.format(Locale.ENGLISH,"<%s> = %s",name,strValue));
                         setValue(schema,visitor,name,strValue);
                     }
                    pageBuilder.addRecord();

                } else {
                    logger.warn(String.format(Locale.ENGLISH,"unmatched line = %s",line));
                }

            }
        }
        pageBuilder.finish();

    }
    private void setValue(Schema schema,ColumnVisitorImpl visitor,String name,String strValue){
        try {
            Value value = ValueFactory.newString(strValue);
            Column column = schema.lookupColumn(name);
            visitor.setValue(value);
            column.visit(visitor);
        } catch (Exception ex) {
            // TODO Error check
            throw new SchemaConfigException(String.format(Locale.ENGLISH, "TODO Error"));

        }
    }

    private Regex buildRegex(PluginTask task)
    {
        String format = task.getFormat();
        byte[] pattern = format.getBytes();
        return new Regex(pattern,0,pattern.length,Option.NONE,UTF8Encoding.INSTANCE);
    }

    private String captureName(NameEntry e){
        return new String(e.name, e.nameP, e.nameEnd - e.nameP);
    }

    private void validateSchema(PluginTask task,Schema schema)
    {
        Regex regex = buildRegex(task);
        // TODO error check.

        for(Iterator<NameEntry> entry = regex.namedBackrefIterator(); entry.hasNext(); ){
            NameEntry e = entry.next();
            String captureName = captureName(e);
            schema.lookupColumn(captureName); // throw Exception;
        }

    }
}

@@ -4503,24 +4503,97 @@
 getBytes();%0A
+        // throw org.joni.exception.SyntaxException if regex is invalid.%0A
         retu
@@ -4666,16 +4666,16 @@
 TANCE);%0A
-
     %7D%0A%0A 
@@ -4902,37 +4902,8 @@
 sk);
-%0A        // TODO error check.
 %0A%0A  
@@ -5075,24 +5075,24 @@
 ureName(e);%0A
-
             
@@ -5134,16 +5134,28 @@
 / throw 
+SchemaConfig
 Exceptio

Add comments about Exception