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

import com.google.common.collect.Maps;
import org.embulk.config.Config;
import org.embulk.config.ConfigDefault;
import org.embulk.spi.Exec;
import org.komamitsu.fluency.Fluency;
import org.slf4j.Logger;

import java.io.IOException;
import java.util.Map;

public class OutForwardService
{
    private final static Logger logger = Exec.getLogger(OutForwardService.class);

    public interface OutForwardTask
            extends org.embulk.config.Task
    {
        @Config("host")
        @ConfigDefault("\"localhost\"")
        String getHost();
        void setHost(String host);

        @Config("port")
        @ConfigDefault("24224")
        int getPort();
        void setPort(int port);
    }

    public interface Task
            extends org.embulk.config.Task
    {
        @Config("out_forward")
        @ConfigDefault("{}") // TODO
        OutForwardTask getOutForwardTask();

        @Config("message_tag")
        @ConfigDefault("\"message\"")
        String getMessageTag();

        @Config("shutdown_tag")
        @ConfigDefault("\"shutdown\"")
        String getShutdownTag();
    }

    public static void sendShutdownMessage(Task task)
    {
        logger.info("out_forward: send shutdown message.");
        OutForwardService outForward = new OutForwardService(task);
        outForward.emit(task.getShutdownTag(), Maps.newHashMap());
        outForward.finish();
        outForward.close();
    }

    private final Task task;
    private final Fluency client;

    public OutForwardService(Task task)
    {
        this.task = task;
        this.client = newFluency(task.getOutForwardTask());
    }

    private Fluency newFluency(OutForwardTask forwardTask)
    {
        try {
            return Fluency.defaultFluency(forwardTask.getHost(), forwardTask.getPort());
        }
        catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    public void emit(String tag, Map<String, Object> message)
    {
        try {
            client.emit(tag, message);
        }
        catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    public void emit(Map<String, Object> message)
    {
        emit(task.getMessageTag(), message);<commit_msg>Enable to set up OutForwardService detailed settings<commit_after>package org.embulk.filter.copy.service;

import com.google.common.collect.Maps;
import org.embulk.config.Config;
import org.embulk.config.ConfigDefault;
import org.embulk.spi.Exec;
import org.komamitsu.fluency.Fluency;
import org.slf4j.Logger;

import java.io.IOException;
import java.util.Map;
import java.util.Optional;

public class OutForwardService
{
    private final static Logger logger = Exec.getLogger(OutForwardService.class);

    public interface OutForwardTask
            extends org.embulk.config.Task
    {
        @Config("host")
        @ConfigDefault("\"localhost\"")
        String getHost();
        void setHost(String host);

        @Config("port")
        @ConfigDefault("24224")
        int getPort();
        void setPort(int port);

        @Config("max_buffer_size")
        @ConfigDefault("null")
        Optional<Long> getMaxBufferSize();

        @Config("buffer_chunk_initial_size")
        @ConfigDefault("null")
        Optional<Integer> getBufferChunkInitialSize();

        @Config("buffer_chunk_retention_size")
        @ConfigDefault("null")
        Optional<Integer> getBufferChunkRetentionSize();

        @Config("flush_interval_millis")
        @ConfigDefault("null")
        Optional<Integer> getFlushIntervalMillis();

        @Config("sender_max_retry_count")
        @ConfigDefault("null")
        Optional<Integer> getSenderMaxRetryCount();

        @Config("ack_response_mode")
        @ConfigDefault("null")
        Optional<Boolean> getAckResponseMode();

        @Config("file_backup_dir")
        @ConfigDefault("null")
        Optional<String> getFileBackupDir();

        @Config("wait_until_buffer_flushed")
        @ConfigDefault("null")
        Optional<Integer> getWaitUntilBufferFlushed();

        @Config("wait_until_flusher_terminated")
        @ConfigDefault("null")
        Optional<Integer> getWaitUntilFlusherTerminated();
    }

    public interface Task
            extends org.embulk.config.Task
    {
        @Config("out_forward")
        @ConfigDefault("{}") // TODO
        OutForwardTask getOutForwardTask();

        @Config("message_tag")
        @ConfigDefault("\"message\"")
        String getMessageTag();

        @Config("shutdown_tag")
        @ConfigDefault("\"shutdown\"")
        String getShutdownTag();
    }

    public static void sendShutdownMessage(Task task)
    {
        logger.info("out_forward: send shutdown message.");
        OutForwardService outForward = new OutForwardService(task);
        outForward.emit(task.getShutdownTag(), Maps.newHashMap());
        outForward.finish();
        outForward.close();
    }

    private final Task task;
    private final Fluency client;

    public OutForwardService(Task task)
    {
        this.task = task;
        this.client = newFluency(task.getOutForwardTask());
    }

    private Fluency.Config configureFluencyConfig(OutForwardTask t)
    {
        Fluency.Config c = new Fluency.Config();
        if (t.getMaxBufferSize().isPresent()) {
            c.setMaxBufferSize(t.getMaxBufferSize().get());
        }
        if (t.getBufferChunkInitialSize().isPresent()) {
            c.setBufferChunkInitialSize(t.getBufferChunkInitialSize().get());
        }
        if (t.getBufferChunkRetentionSize().isPresent()) {
            c.setBufferChunkRetentionSize(t.getBufferChunkRetentionSize().get());
        }
        if (t.getFlushIntervalMillis().isPresent()) {
            c.setFlushIntervalMillis(t.getFlushIntervalMillis().get());
        }
        if (t.getSenderMaxRetryCount().isPresent()) {
            c.setSenderMaxRetryCount(t.getSenderMaxRetryCount().get());
        }
        if (t.getAckResponseMode().isPresent()) {
            c.setAckResponseMode(t.getAckResponseMode().get());
        }
        if (t.getFileBackupDir().isPresent()) {
            c.setFileBackupDir(t.getFileBackupDir().get());
        }
        if (t.getWaitUntilBufferFlushed().isPresent()) {
            c.setWaitUntilBufferFlushed(t.getWaitUntilBufferFlushed().get());
        }
        if (t.getWaitUntilFlusherTerminated().isPresent()) {
            c.setWaitUntilFlusherTerminated(t.getWaitUntilFlusherTerminated().get());
        }
        return c;
    }

    private Fluency newFluency(OutForwardTask t)
    {
        Fluency.Config c = configureFluencyConfig(t);
        try {
            return Fluency.defaultFluency(t.getHost(), t.getPort(), c);
        }
        catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    public void emit(String tag, Map<String, Object> message)
    {
        try {
            client.emit(tag, message);
        }
        catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    public void emit(Map<String, Object> message)
    {
        emit(task.getMessageTag(), message);
{'fraction_non_alphanumeric': 0.11745013251499511, 'fraction_numerical': 0.0016738736225414982, 'mean_word_length': 2.6048265460030167, 'pattern_counts': {'":': 0, '<': 16, '<?xml version=': 0, '>': 16, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4824285', 'n_tokens_mistral': 2056, 'n_tokens_neox': 2015, 'n_words': 350}