package org.embulk.filter.stdout;

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.Column;
import org.embulk.spi.FilterPlugin;
import org.embulk.spi.PageOutput;
import org.embulk.spi.PageBuilder;
import org.embulk.spi.PageReader;
import org.embulk.spi.Page;
import org.embulk.spi.Exec;
import org.embulk.spi.util.PagePrinter;
import org.embulk.spi.Schema;
import org.embulk.spi.time.TimestampFormatter;
import org.embulk.spi.ColumnVisitor;

public class StdoutFilterPlugin
        implements FilterPlugin
{
    public interface PluginTask
            extends Task, TimestampFormatter.FormatterTask
    {
    }

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

        Schema outputSchema = inputSchema;

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

    @Override
    public PageOutput open(final TaskSource taskSource, final Schema inputSchema,
            final Schema outputSchema, final PageOutput output)
    {
        final PluginTask task = taskSource.loadTask(PluginTask.class);

        return new PageOutput() {
            private final PageReader pageReader = new PageReader(inputSchema);
            private final PageBuilder pageBuilder = new PageBuilder(Exec.getBufferAllocator(), outputSchema, output);
            private final PagePrinter pagePrinter = new PagePrinter(inputSchema, task);
            private final ColumnVisitorImpl visitor = new ColumnVisitorImpl(pageBuilder);

            @Override
            public void finish() {
                pageBuilder.finish();
            }

            @Override
            public void close() {
                pageBuilder.close();
            }

            @Override
            public void add(Page page) {
                pageReader.setPage(page);

                while (pageReader.nextRecord()) {
                    System.out.println(pagePrinter.printRecord(pageReader, ",")); // here!
                    outputSchema.visitColumns(visitor);
                    pageBuilder.addRecord();
                }
            }

            class ColumnVisitorImpl implements ColumnVisitor {
                private final PageBuilder pageBuilder;

                ColumnVisitorImpl(PageBuilder pageBuilder) {
                    this.pageBuilder = pageBuilder;
                }

                @Override
                public void booleanColumn(Column outputColumn) {
                    if (pageReader.isNull(outputColumn)) {
                        pageBuilder.setNull(outputColumn);
                    } else {
                        pageBuilder.setBoolean(outputColumn, pageReader.getBoolean(outputColumn));
                    }
                }

                @Override
                public void longColumn(Column outputColumn) {
                    if (pageReader.isNull(outputColumn)) {
                        pageBuilder.setNull(outputColumn);
                    } else {
                        pageBuilder.setLong(outputColumn, pageReader.getLong(outputColumn));
                    }
                }

                @Override
                public void doubleColumn(Column outputColumn) {
                    if (pageReader.isNull(outputColumn)) {
                        pageBuilder.setNull(outputColumn);
                    } else {
                        pageBuilder.setDouble(outputColumn, pageReader.getDouble(outputColumn));
                    }
                }

                @Override
                public void stringColumn(Column outputColumn) {
                    if (pageReader.isNull(outputColumn)) {
                        pageBuilder.setNull(outputColumn);
                    } else {
                        pageBuilder.setString(outputColumn, pageReader.getString(outputColumn));
                    }
                }

                @Override
                public void timestampColumn(Column outputColumn) {
                    if (pageReader.isNull(outputColumn)) {
                        pageBuilder.setNull(outputColumn);
                    } else {
                        pageBuilder.setTimestamp(outputColumn, pageReader.getTimestamp(outputColumn));
                    }
                }
            }
        };
    }
}

@@ -68,16 +68,17 @@
 tional;%0A
+%0A
 import o
@@ -146,45 +146,8 @@
 lt;%0A
-import org.embulk.config.ConfigDiff;%0A
 impo
@@ -548,55 +548,8 @@
 ma;%0A
-import org.embulk.spi.time.TimestampFormatter;%0A
 impo
@@ -708,48 +708,120 @@
 Task
-, TimestampFormatter.FormatterTask%0A    %7B
+%0A    %7B%0A        @Config(%22timezone%22)%0A        @ConfigDefault(%22null%22)%0A        public Optional%3CString%3E getTimezone();
 %0A   
@@ -1672,16 +1672,40 @@
 ma, task
+.getTimezone().or(%22UTC%22)
 );%0A     
@@ -4511,32 +4511,401 @@
                %7D
+%0A%0A                @Override%0A                public void jsonColumn(Column outputColumn) %7B%0A                    if (pageReader.isNull(outputColumn)) %7B%0A                        pageBuilder.setNull(outputColumn);%0A                    %7D else %7B%0A                        pageBuilder.setJson(outputColumn, pageReader.getJson(outputColumn));%0A                    %7D%0A                %7D
 %0A            %7D%0A 

Fix for issue #1