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();

        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);

        String format = task.getFormat();
        logger.info(String.format(Locale.ENGLISH,"format = %s",format));
        byte[] pattern = format.getBytes();

        Regex regex = new Regex(pattern,0,pattern.length,Option.NONE,UTF8Encoding.INSTANCE);

        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();
                         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);
                         String name = new String(e.name, e.nameP, e.nameEnd - e.nameP);
                         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) {
            throw new SchemaConfigException(String.format(Locale.ENGLISH, "TODO Error"));

        }
    }
}

@@ -1732,24 +1732,62 @@
 oSchema();%0A%0A
+        validateSchema(task,schema);%0A%0A
         cont
@@ -1816,16 +1816,16 @@
 chema);%0A
-
     %7D%0A%0A 
@@ -2390,168 +2390,8 @@
 );%0A%0A
-        String format = task.getFormat();%0A        logger.info(String.format(Locale.ENGLISH,%22format = %25s%22,format));%0A        byte%5B%5D pattern = format.getBytes();%0A%0A
     
@@ -2412,76 +2412,23 @@
 x = 
-new 
+build
 Regex(
-pattern,0,pattern.length,Option.NONE,UTF8Encoding.INSTANCE
+task
 );%0A%0A
@@ -3104,24 +3104,80 @@
 try.next();%0A
+                         String name = captureName(e);%0A%0A
             
@@ -3334,17 +3334,16 @@
 umber%5D;%0A
-%0A
         
@@ -3476,97 +3476,9 @@
 8);%0A
-                         String name = new String(e.name, e.nameP, e.nameEnd - e.nameP);
 %0A
+
     
@@ -4225,24 +4225,56 @@
 ption ex) %7B%0A
+            // TODO Error check%0A
             
@@ -4348,24 +4348,24 @@
  Error%22));%0A%0A
-
         %7D%0A  
@@ -4367,11 +4367,761 @@
  %7D%0A    %7D
+%0A%0A    private Regex buildRegex(PluginTask task)%0A    %7B%0A        String format = task.getFormat();%0A        byte%5B%5D pattern = format.getBytes();%0A        return new Regex(pattern,0,pattern.length,Option.NONE,UTF8Encoding.INSTANCE);%0A    %7D%0A%0A    private String captureName(NameEntry e)%7B%0A        return new String(e.name, e.nameP, e.nameEnd - e.nameP);%0A    %7D%0A%0A    private void validateSchema(PluginTask task,Schema schema)%0A    %7B%0A        Regex regex = buildRegex(task);%0A        // TODO error check.%0A%0A        for(Iterator%3CNameEntry%3E entry = regex.namedBackrefIterator(); entry.hasNext(); )%7B%0A            NameEntry e = entry.next();%0A            String captureName = captureName(e);%0A            schema.lookupColumn(captureName); // throw Exception;%0A        %7D%0A%0A    %7D
 %0A%7D%0A

Add column check method