package org.komamitsu.fluency;

import org.komamitsu.fluency.buffer.Buffer;
import org.komamitsu.fluency.buffer.BufferConfig;
import org.komamitsu.fluency.buffer.MessageBuffer;
import org.komamitsu.fluency.buffer.PackedForwardBuffer;
import org.komamitsu.fluency.flusher.Flusher;
import org.komamitsu.fluency.flusher.FlusherConfig;
import org.komamitsu.fluency.flusher.SyncFlusher;
import org.komamitsu.fluency.sender.RetryableSender;
import org.komamitsu.fluency.sender.Sender;
import org.komamitsu.fluency.sender.TCPSender;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Closeable;
import java.io.Flushable;
import java.io.IOException;
import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;
import java.util.Map;

public class Fluency
        implements Flushable, Closeable
{
    private static final Logger LOG = LoggerFactory.getLogger(Fluency.class);
    private final Sender sender;
    private final Buffer buffer;
    private final Flusher flusher;

    public static Fluency defaultFluency(String host, int port)
            throws IOException
    {
        return new Fluency.Builder(new RetryableSender(new TCPSender(host, port))).build();
    }

    private Fluency(Sender sender, Buffer buffer, Flusher flusher)
    {
        this.sender = sender;
        this.buffer = buffer;
        this.flusher = flusher;
    }

    public void emit(String tag, long timestamp, Map<String, Object> data)
            throws IOException
    {
        buffer.append(tag, timestamp, data);
        flusher.onUpdate(this, buffer);
    }

    public void emit(String tag, Map<String, Object> data)
            throws IOException
    {
        emit(tag, System.currentTimeMillis() / 1000, data);
    }

    @Override
    public void flush()
            throws IOException
    {
        buffer.flush(sender);
    }

    @Override
    public void close()
            throws IOException
    {
        flusher.flush(this, buffer);
        sender.close();
    }

    public static class Builder
    {
        private final Sender sender;
        private Class<? extends Flusher> flusherClass = SyncFlusher.class;
        private BufferConfig bufferConfig = new BufferConfig.Builder().build();
        private FlusherConfig flusherConfig = new FlusherConfig.Builder().build();
        public Builder(Sender sender)
        {
            this.sender = sender;
        }

        public Builder setFlusherClass(Class<? extends Flusher> flusherClass)
        {
            this.flusherClass = flusherClass;
            return this;
        }

        public Builder setBufferConfig(BufferConfig bufferConfig)
        {
            this.bufferConfig = bufferConfig;
            return this;
        }

        public Builder setFlusherConfig(FlusherConfig flusherConfig)
        {
            this.flusherConfig = flusherConfig;
            return this;
        }

        public Fluency build()
        {
            // TODO: Make the class of `buffer` configurable
            Buffer buffer = new PackedForwardBuffer(bufferConfig);
            Exception exception = null;
            Constructor<? extends Flusher> constructor = null;
            try {
                constructor = flusherClass.getConstructor(FlusherConfig.class);
                Flusher flusher = constructor.newInstance(flusherConfig);
                return new Fluency(sender, buffer, flusher);
            }
            catch (NoSuchMethodException e) {
                exception = e;
            }
            catch (InvocationTargetException e) {
                exception = e;
            }
            catch (InstantiationException e) {
                exception = e;
            }
            catch (IllegalAccessException e) {
                exception = e;
            }
            throw new IllegalStateException("Failed to build an instance. flusherClass=" + flusherClass, exception);
        }
    }
}

@@ -123,59 +123,8 @@
 ig;%0A
-import org.komamitsu.fluency.buffer.MessageBuffer;%0A
 impo
@@ -2067,41 +2067,47 @@
 nds 
-Flusher%3E flusherClass = SyncFlush
+Buffer%3E bufferClass = PackedForwardBuff
 er.c
@@ -2184,32 +2184,107 @@
 lder().build();%0A
+        private Class%3C? extends Flusher%3E flusherClass = SyncFlusher.class;%0A
         private 
@@ -2435,32 +2435,197 @@
 der;%0A        %7D%0A%0A
+        public Builder setBufferClass(Class%3C? extends Buffer%3E bufferClass)%0A        %7B%0A            this.bufferClass = bufferClass;%0A            return this;%0A        %7D%0A%0A
         public B
@@ -3156,110 +3156,148 @@
     
-// TODO: Make the class of %60buffer%60 configurable%0A           
+Exception exception = null;%0A            try %7B%0A                Constructor%3C? extends
  Buffer
+%3E
  buffer
- = new PackedForwardBuffer(b
+Constructor = bufferClass.getConstructor(B
 uffe
@@ -3295,32 +3295,38 @@
 tor(BufferConfig
+.class
 );%0A            E
@@ -3328,134 +3328,129 @@
     
-Exception exception = null;%0A            Constructor%3C? extends Flusher%3E constructor = null;%0A            try %7B%0A                c
+    Buffer buffer = bufferConstructor.newInstance(bufferConfig);%0A%0A                Constructor%3C? extends Flusher%3E flusherC
 onst
@@ -3542,17 +3542,24 @@
 usher = 
-c
+flusherC
 onstruct
@@ -3589,16 +3589,17 @@
 onfig);%0A
+%0A
         
@@ -4108,16 +4108,49 @@
 nstance.
+ bufferClass=%22 + bufferClass + %22,
  flusher

Make `Buffer` class which `Fluency` uses configurable