<commit_before>package org.embulk.filter.copy.plugin;

import org.embulk.config.Config;
import org.embulk.config.ConfigDiff;
import org.embulk.config.ConfigInject;
import org.embulk.config.ConfigSource;
import org.embulk.config.Task;
import org.embulk.config.TaskReport;
import org.embulk.config.TaskSource;
import org.embulk.filter.copy.forward.InForwardEventReader;
import org.embulk.filter.copy.forward.InForwardService;
import org.embulk.filter.copy.forward.InForwardVisitor;
import org.embulk.spi.BufferAllocator;
import org.embulk.spi.Exec;
import org.embulk.spi.InputPlugin;
import org.embulk.spi.PageBuilder;
import org.embulk.spi.PageOutput;
import org.embulk.spi.Schema;
import org.embulk.spi.SchemaConfig;
import org.embulk.spi.time.TimestampParser;
import org.slf4j.Logger;

import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;

public class InternalForwardInputPlugin
        implements InputPlugin
{
    public final static String PLUGIN_NAME = "internal_forward";
    private final static Logger logger = Exec.getLogger(InternalForwardInputPlugin.class);

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

        @ConfigInject
        BufferAllocator getBufferAllocator();
    }

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

        Schema schema = task.getColumns().toSchema();
        int taskCount = 1;  // number of run() method calls

        return resume(task.dump(), schema, taskCount, control);
    }

    @Override
    public ConfigDiff resume(TaskSource taskSource,
            Schema schema, int taskCount,
            InputPlugin.Control control)
    {
        control.run(taskSource, schema, taskCount);
        return Exec.newConfigDiff();
    }

    @Override
    public void cleanup(TaskSource taskSource,
            Schema schema, int taskCount,
            List<TaskReport> successTaskReports)
    {
        logger.info("input-copy: cleanup");
    }

    @Override
    public TaskReport run(TaskSource taskSource,
            Schema schema, int taskIndex,
            PageOutput output)
    {
        PluginTask task = taskSource.loadTask(PluginTask.class);

        try (PageBuilder pageBuilder = new PageBuilder(task.getBufferAllocator(), schema, output)) {
            TimestampParser timestampParser = new TimestampParser(
                    task.getJRuby(),
                    task.getDefaultTimestampFormat(),
                    task.getDefaultTimeZone());
            InForwardEventReader eventReader = new InForwardEventReader(schema, timestampParser);
            InForwardVisitor inForwardVisitor = new InForwardVisitor(eventReader, pageBuilder);

            AtomicBoolean isShutdown = new AtomicBoolean(false);
            Runtime.getRuntime().addShutdownHook(new Thread(() -> isShutdown.set(true)));

            InForwardService.builder()
                    .task(task)
                    .forEachEventCallback(
                            event ->
                            {
                                // TODO: here is not thread-safe
                                eventReader.setEvent(event);
                                while (eventReader.nextMessage()) {
                                    schema.visitColumns(inForwardVisitor);
                                    pageBuilder.addRecord();
                                }
                            }
                    )
                    .build()
                    .runUntilShouldShutdown();

            pageBuilder.finish();
        }

        return Exec.newTaskReport(); // TODO
    }<commit_msg>Remove unnecessary shutdown hook<commit_after>package org.embulk.filter.copy.plugin;

import org.embulk.config.Config;
import org.embulk.config.ConfigDiff;
import org.embulk.config.ConfigInject;
import org.embulk.config.ConfigSource;
import org.embulk.config.Task;
import org.embulk.config.TaskReport;
import org.embulk.config.TaskSource;
import org.embulk.filter.copy.forward.InForwardEventReader;
import org.embulk.filter.copy.forward.InForwardService;
import org.embulk.filter.copy.forward.InForwardVisitor;
import org.embulk.spi.BufferAllocator;
import org.embulk.spi.Exec;
import org.embulk.spi.InputPlugin;
import org.embulk.spi.PageBuilder;
import org.embulk.spi.PageOutput;
import org.embulk.spi.Schema;
import org.embulk.spi.SchemaConfig;
import org.embulk.spi.time.TimestampParser;
import org.slf4j.Logger;

import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;

public class InternalForwardInputPlugin
        implements InputPlugin
{
    public final static String PLUGIN_NAME = "internal_forward";
    private final static Logger logger = Exec.getLogger(InternalForwardInputPlugin.class);

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

        @ConfigInject
        BufferAllocator getBufferAllocator();
    }

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

        Schema schema = task.getColumns().toSchema();
        int taskCount = 1;  // number of run() method calls

        return resume(task.dump(), schema, taskCount, control);
    }

    @Override
    public ConfigDiff resume(TaskSource taskSource,
            Schema schema, int taskCount,
            InputPlugin.Control control)
    {
        control.run(taskSource, schema, taskCount);
        return Exec.newConfigDiff();
    }

    @Override
    public void cleanup(TaskSource taskSource,
            Schema schema, int taskCount,
            List<TaskReport> successTaskReports)
    {
        logger.info("input-copy: cleanup");
    }

    @Override
    public TaskReport run(TaskSource taskSource,
            Schema schema, int taskIndex,
            PageOutput output)
    {
        PluginTask task = taskSource.loadTask(PluginTask.class);

        try (PageBuilder pageBuilder = new PageBuilder(task.getBufferAllocator(), schema, output)) {
            TimestampParser timestampParser = new TimestampParser(
                    task.getJRuby(),
                    task.getDefaultTimestampFormat(),
                    task.getDefaultTimeZone());
            InForwardEventReader eventReader = new InForwardEventReader(schema, timestampParser);
            InForwardVisitor inForwardVisitor = new InForwardVisitor(eventReader, pageBuilder);

            InForwardService.builder()
                    .task(task)
                    .forEachEventCallback(
                            event ->
                            {
                                // TODO: here is not thread-safe
                                eventReader.setEvent(event);
                                while (eventReader.nextMessage()) {
                                    schema.visitColumns(inForwardVisitor);
                                    pageBuilder.addRecord();
                                }
                            }
                    )
                    .build()
                    .runUntilShouldShutdown();

            pageBuilder.finish();
        }

        return Exec.newTaskReport(); // TODO
    }
{'fraction_non_alphanumeric': 0.08392356006949085, 'fraction_numerical': 0.0005345449685954831, 'mean_word_length': 2.171186440677966, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 8, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '261172', 'n_tokens_mistral': 1934, 'n_tokens_neox': 1869, 'n_words': 384}