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.ParserPlugin;
import org.embulk.spi.FileInput;
import org.embulk.spi.PageOutput;
import org.embulk.spi.Schema;
import org.embulk.spi.SchemaConfig;

public class JoniParserPlugin
        implements ParserPlugin
{
    public interface PluginTask
            extends Task
    {
        // configuration option 1 (required integer)
        @Config("option1")
        public int getOption1();

        // configuration option 2 (optional string, null is not allowed)
        @Config("option2")
        @ConfigDefault("\"myvalue\"")
        public String getOption2();

        // configuration option 3 (optional string, null is allowed)
        @Config("option3")
        @ConfigDefault("null")
        public Optional<String> getOption3();

        // if you get schema from config or data source
        @Config("columns")
        public SchemaConfig getColumns();
    }

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

        // Write your code here :)
        throw new UnsupportedOperationException("JoniParserPlugin.run method is not implemented yet");
    }
}

@@ -456,724 +456,658 @@
 ig;%0A
-%0Apublic class JoniParserPlugin%0A        implements ParserPlugin%0A%7B%0A    public interface PluginTask%0A            extends Task%0A    %7B%0A        // configuration option 1 (required integer)%0A        @Config(%22option1%22)%0A        public int getOption1();%0A%0A        // configuration option 2 (optional string, null is not allowed)%0A        @Config(%22option2%22)%0A        @ConfigDefault(%22%5C%22myvalue%5C%22%22)%0A        public String getOption2();%0A%0A        // configuration option 3 (optional string, null is allowed)%0A        @Config(%22option3%22)%0A        @ConfigDefault(%22null%22)%0A        public Optional%3CString%3E getOption3();%0A%0A        // if you get schema from config or data source%0A        @Config(%22columns%22)%0A        public SchemaConfig getColumns
+import org.embulk.spi.PageBuilder;%0A%0Aimport org.embulk.spi.time.TimestampParser;%0Aimport org.embulk.spi.Exec;%0Aimport org.embulk.spi.util.Timestamps;%0A%0Aimport org.embulk.spi.util.LineDecoder;%0Aimport org.slf4j.Logger;%0A%0A%0Apublic class JoniParserPlugin%0A        implements ParserPlugin%0A%7B%0A    private static final Logger logger = Exec.getLogger(JoniParserPlugin.class);%0A%0A    public interface PluginTask%0A            extends Task,LineDecoder.DecoderTask, TimestampParser.Task%0A    %7B%0A        @Config(%22columns%22)%0A        SchemaConfig getColumns();%0A%0A        @Config(%22stop_on_invalid_record%22)%0A        @ConfigDefault(%22false%22)%0A        boolean getStopOnInvalidRecord
 ();%0A
+%0A
     
@@ -1567,25 +1567,24 @@
 class);%0A
-%0A
         
 // Write
@@ -1579,137 +1579,486 @@
     
-// Write your code here :)%0A        throw new UnsupportedOperationException(%22JoniParserPlugin.run method is not implemented yet%22);
+LineDecoder lineDecoder = new LineDecoder(input, task);%0A        PageBuilder pageBuilder = new PageBuilder(Exec.getBufferAllocator(), schema, output);%0A        TimestampParser%5B%5D timestampParsers = Timestamps.newTimestampColumnParsers(task, task.getColumns());%0A%0A        while (input.nextFile()) %7B%0A            while (true) %7B%0A                String line = lineDecoder.poll();%0A%0A                if (line == null) %7B%0A                    break;%0A                %7D%0A%0A            %7D%0A        %7D%0A
 %0A   

Write plugin skelton