<commit_before>package org.komamitsu.fluency;

import org.junit.Test;
import org.komamitsu.fluency.buffer.Buffer;
import org.komamitsu.fluency.buffer.MessageBuffer;
import org.komamitsu.fluency.buffer.PackedForwardBuffer;
import org.komamitsu.fluency.flusher.AsyncFlusher;
import org.komamitsu.fluency.flusher.Flusher;
import org.komamitsu.fluency.flusher.SyncFlusher;
import org.komamitsu.fluency.sender.Sender;
import org.komamitsu.fluency.sender.TCPSender;
import org.msgpack.value.MapValue;
import org.msgpack.value.Value;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.SocketChannel;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

import static org.junit.Assert.*;

public class FluencyTest
{
    private static final Logger LOG = LoggerFactory.getLogger(FluencyTest.class);

    @Test
    public void testDefaultFluency()
            throws IOException
    {
        Fluency fluency = null;
        fluency = Fluency.defaultFluency();
        fluency = Fluency.defaultFluency(12345);
        fluency = Fluency.defaultFluency("333.333.333.333", 12345);
        fluency = Fluency.defaultFluency(Arrays.asList(new InetSocketAddress(43210)));
        Fluency.Config config = new Fluency.Config();
        config.setFlushIntervalMillis(200).setMaxBufferSize(64 * 1024 * 1024).setSenderMaxRetryCount(99);
        fluency = Fluency.defaultFluency(config);
        fluency = Fluency.defaultFluency(12345, config);
        fluency = Fluency.defaultFluency("333.333.333.333", 12345, config);
        fluency = Fluency.defaultFluency(Arrays.asList(new InetSocketAddress(43210)), config);
    }

    interface FluencyFactory
    {
        Fluency generate(int fluentdPort)
                throws IOException;
    }

    @Test
    public void testFluencyUsingPackedForwardBufferAndAsyncFlusher()
            throws Exception
    {
        testFluencyBase(new FluencyFactory() {
            @Override
            public Fluency generate(int fluentdPort)
                    throws IOException
            {
                Sender sender = new TCPSender(fluentdPort);
                Buffer.Config bufferConfig = new PackedForwardBuffer.Config();
                Flusher.Config flusherConfig = new AsyncFlusher.Config();
                return new Fluency.Builder(sender).setBufferConfig(bufferConfig).setFlusherConfig(flusherConfig).build();
            }
        });
    }

    @Test
    public void testFluencyUsingMessageAndAsyncFlusher()
            throws Exception
    {
        testFluencyBase(new FluencyFactory() {
            @Override
            public Fluency generate(int fluentdPort)
                    throws IOException
            {
                Sender sender = new TCPSender(fluentdPort);
                Buffer.Config bufferConfig = new MessageBuffer.Config();
                Flusher.Config flusherConfig = new AsyncFlusher.Config();
                return new Fluency.Builder(sender).setBufferConfig(bufferConfig).setFlusherConfig(flusherConfig).build();
            }
        });
    }

    @Test
    public void testFluencyUsingPackedForwardBufferAndSyncFlusher()
            throws Exception
    {
        testFluencyBase(new FluencyFactory() {
            @Override
            public Fluency generate(int fluentdPort)
                    throws IOException
            {
                Sender sender = new TCPSender(fluentdPort);
                Buffer.Config bufferConfig = new PackedForwardBuffer.Config();
                Flusher.Config flusherConfig = new SyncFlusher.Config();
                return new Fluency.Builder(sender).setBufferConfig(bufferConfig).setFlusherConfig(flusherConfig).build();
            }
        });
    }

    @Test
    public void testFluencyUsingMessageAndSyncFlusher()
            throws Exception
    {
        testFluencyBase(new FluencyFactory() {
            @Override
            public Fluency generate(int fluentdPort)
                    throws IOException
            {
                Sender sender = new TCPSender(fluentdPort);
                Buffer.Config bufferConfig = new MessageBuffer.Config();
                Flusher.Config flusherConfig = new SyncFlusher.Config();
                return new Fluency.Builder(sender).setBufferConfig(bufferConfig).setFlusherConfig(flusherConfig).build();
            }
        });
    }

    @Test
    public void testFluencyUsingMessageAndSyncFlusherWithAckResponse()
            throws Exception
    {
        testFluencyBase(new FluencyFactory() {
            @Override
            public Fluency generate(int fluentdPort)
                    throws IOException
            {
                Sender sender = new TCPSender(fluentdPort);
                Buffer.Config bufferConfig = new MessageBuffer.Config().setAckResponseMode(true);
                Flusher.Config flusherConfig = new SyncFlusher.Config();
                return new Fluency.Builder(sender).setBufferConfig(bufferConfig).setFlusherConfig(flusherConfig).build();
            }
        });
    }

    @Test
    public void testFluencyUsingMessageAndAsyncFlusherWithAckResponse()
            throws Exception
    {
        testFluencyBase(new FluencyFactory() {
            @Override
            public Fluency generate(int fluentdPort)
                    throws IOException
            {
                Sender sender = new TCPSender(fluentdPort);
                Buffer.Config bufferConfig = new MessageBuffer.Config().setAckResponseMode(true);
                Flusher.Config flusherConfig = new AsyncFlusher.Config();
                return new Fluency.Builder(sender).setBufferConfig(bufferConfig).setFlusherConfig(flusherConfig).build();
            }
        });
    }

    @Test
    public void testFluencyUsingPackedForwardAndSyncFlusherWithAckResponse()
            throws Exception
    {
        testFluencyBase(new FluencyFactory() {
            @Override
            public Fluency generate(int fluentdPort)
                    throws IOException
            {
                Sender sender = new TCPSender(fluentdPort);
                Buffer.Config bufferConfig = new PackedForwardBuffer.Config().setAckResponseMode(true);
                Flusher.Config flusherConfig = new SyncFlusher.Config();
                return new Fluency.Builder(sender).setBufferConfig(bufferConfig).setFlusherConfig(flusherConfig).build();
            }
        });
    }

    @Test
    public void testFluencyUsingPackedForwardAndAsyncFlusherWithAckResponse()
            throws Exception
    {
        testFluencyBase(new FluencyFactory() {
            @Override
            public Fluency generate(int fluentdPort)
                    throws IOException
            {
                Sender sender = new TCPSender(fluentdPort);
                Buffer.Config bufferConfig = new PackedForwardBuffer.Config().setAckResponseMode(true);
                Flusher.Config flusherConfig = new AsyncFlusher.Config();
                return new Fluency.Builder(sender).setBufferConfig(bufferConfig).setFlusherConfig(flusherConfig).build();
            }
        });
    }

    public void testFluencyBase(FluencyFactory fluencyFactory)
            throws Exception
    {
        MockFluentdServer fluentd = new MockFluentdServer();<commit_msg>Put each test in FluencyTest into one test method<commit_after>package org.komamitsu.fluency;

import org.junit.Test;
import org.komamitsu.fluency.buffer.Buffer;
import org.komamitsu.fluency.buffer.MessageBuffer;
import org.komamitsu.fluency.buffer.PackedForwardBuffer;
import org.komamitsu.fluency.flusher.AsyncFlusher;
import org.komamitsu.fluency.flusher.Flusher;
import org.komamitsu.fluency.flusher.SyncFlusher;
import org.komamitsu.fluency.sender.Sender;
import org.komamitsu.fluency.sender.TCPSender;
import org.msgpack.value.MapValue;
import org.msgpack.value.Value;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.SocketChannel;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

import static org.junit.Assert.*;

public class FluencyTest
{
    private static final Logger LOG = LoggerFactory.getLogger(FluencyTest.class);

    @Test
    public void testDefaultFluency()
            throws IOException
    {
        Fluency fluency = null;
        fluency = Fluency.defaultFluency();
        fluency = Fluency.defaultFluency(12345);
        fluency = Fluency.defaultFluency("333.333.333.333", 12345);
        fluency = Fluency.defaultFluency(Arrays.asList(new InetSocketAddress(43210)));
        Fluency.Config config = new Fluency.Config();
        config.setFlushIntervalMillis(200).setMaxBufferSize(64 * 1024 * 1024).setSenderMaxRetryCount(99);
        fluency = Fluency.defaultFluency(config);
        fluency = Fluency.defaultFluency(12345, config);
        fluency = Fluency.defaultFluency("333.333.333.333", 12345, config);
        fluency = Fluency.defaultFluency(Arrays.asList(new InetSocketAddress(43210)), config);
    }

    interface FluencyFactory
    {
        Fluency generate(int fluentdPort)
                throws IOException;
    }

    @Test
    public void testFluencyEachCombination()
            throws Exception
    {
        testFluencyUsingPackedForwardBufferAndAsyncFlusher();
        testFluencyUsingMessageAndAsyncFlusher();
        testFluencyUsingPackedForwardBufferAndSyncFlusher();
        testFluencyUsingMessageAndSyncFlusher();
        testFluencyUsingMessageAndSyncFlusherWithAckResponse();
        testFluencyUsingMessageAndAsyncFlusherWithAckResponse();
        testFluencyUsingPackedForwardAndSyncFlusherWithAckResponse();
        testFluencyUsingPackedForwardAndAsyncFlusherWithAckResponse();
    }

    public void testFluencyUsingPackedForwardBufferAndAsyncFlusher()
            throws Exception
    {
        testFluencyBase(new FluencyFactory() {
            @Override
            public Fluency generate(int fluentdPort)
                    throws IOException
            {
                Sender sender = new TCPSender(fluentdPort);
                Buffer.Config bufferConfig = new PackedForwardBuffer.Config();
                Flusher.Config flusherConfig = new AsyncFlusher.Config();
                return new Fluency.Builder(sender).setBufferConfig(bufferConfig).setFlusherConfig(flusherConfig).build();
            }
        });
    }

    public void testFluencyUsingMessageAndAsyncFlusher()
            throws Exception
    {
        testFluencyBase(new FluencyFactory() {
            @Override
            public Fluency generate(int fluentdPort)
                    throws IOException
            {
                Sender sender = new TCPSender(fluentdPort);
                Buffer.Config bufferConfig = new MessageBuffer.Config();
                Flusher.Config flusherConfig = new AsyncFlusher.Config();
                return new Fluency.Builder(sender).setBufferConfig(bufferConfig).setFlusherConfig(flusherConfig).build();
            }
        });
    }

    public void testFluencyUsingPackedForwardBufferAndSyncFlusher()
            throws Exception
    {
        testFluencyBase(new FluencyFactory() {
            @Override
            public Fluency generate(int fluentdPort)
                    throws IOException
            {
                Sender sender = new TCPSender(fluentdPort);
                Buffer.Config bufferConfig = new PackedForwardBuffer.Config();
                Flusher.Config flusherConfig = new SyncFlusher.Config();
                return new Fluency.Builder(sender).setBufferConfig(bufferConfig).setFlusherConfig(flusherConfig).build();
            }
        });
    }

    public void testFluencyUsingMessageAndSyncFlusher()
            throws Exception
    {
        testFluencyBase(new FluencyFactory() {
            @Override
            public Fluency generate(int fluentdPort)
                    throws IOException
            {
                Sender sender = new TCPSender(fluentdPort);
                Buffer.Config bufferConfig = new MessageBuffer.Config();
                Flusher.Config flusherConfig = new SyncFlusher.Config();
                return new Fluency.Builder(sender).setBufferConfig(bufferConfig).setFlusherConfig(flusherConfig).build();
            }
        });
    }

    public void testFluencyUsingMessageAndSyncFlusherWithAckResponse()
            throws Exception
    {
        testFluencyBase(new FluencyFactory() {
            @Override
            public Fluency generate(int fluentdPort)
                    throws IOException
            {
                Sender sender = new TCPSender(fluentdPort);
                Buffer.Config bufferConfig = new MessageBuffer.Config().setAckResponseMode(true);
                Flusher.Config flusherConfig = new SyncFlusher.Config();
                return new Fluency.Builder(sender).setBufferConfig(bufferConfig).setFlusherConfig(flusherConfig).build();
            }
        });
    }

    public void testFluencyUsingMessageAndAsyncFlusherWithAckResponse()
            throws Exception
    {
        testFluencyBase(new FluencyFactory() {
            @Override
            public Fluency generate(int fluentdPort)
                    throws IOException
            {
                Sender sender = new TCPSender(fluentdPort);
                Buffer.Config bufferConfig = new MessageBuffer.Config().setAckResponseMode(true);
                Flusher.Config flusherConfig = new AsyncFlusher.Config();
                return new Fluency.Builder(sender).setBufferConfig(bufferConfig).setFlusherConfig(flusherConfig).build();
            }
        });
    }

    public void testFluencyUsingPackedForwardAndSyncFlusherWithAckResponse()
            throws Exception
    {
        testFluencyBase(new FluencyFactory() {
            @Override
            public Fluency generate(int fluentdPort)
                    throws IOException
            {
                Sender sender = new TCPSender(fluentdPort);
                Buffer.Config bufferConfig = new PackedForwardBuffer.Config().setAckResponseMode(true);
                Flusher.Config flusherConfig = new SyncFlusher.Config();
                return new Fluency.Builder(sender).setBufferConfig(bufferConfig).setFlusherConfig(flusherConfig).build();
            }
        });
    }

    public void testFluencyUsingPackedForwardAndAsyncFlusherWithAckResponse()
            throws Exception
    {
        testFluencyBase(new FluencyFactory() {
            @Override
            public Fluency generate(int fluentdPort)
                    throws IOException
            {
                Sender sender = new TCPSender(fluentdPort);
                Buffer.Config bufferConfig = new PackedForwardBuffer.Config().setAckResponseMode(true);
                Flusher.Config flusherConfig = new AsyncFlusher.Config();
                return new Fluency.Builder(sender).setBufferConfig(bufferConfig).setFlusherConfig(flusherConfig).build();
            }
        });
    }

    public void testFluencyBase(FluencyFactory fluencyFactory)
            throws Exception
    {
        MockFluentdServer fluentd = new MockFluentdServer();
{'fraction_non_alphanumeric': 0.07786493208341305, 'fraction_numerical': 0.009055544926981698, 'mean_word_length': 2.457231040564374, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, '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': '2398170', 'n_tokens_mistral': 4313, 'n_tokens_neox': 4077, 'n_words': 733}