// Path: fluency-fluentd/src/main/java/org/komamitsu/fluency/fluentd/ingester/sender/FluentdSender.java
// public abstract class FluentdSender
//         implements Closeable, Sender
// {
//     private static final Logger LOG = LoggerFactory.getLogger(FluentdSender.class);
//     private final Config config;
// 
//     protected FluentdSender(Config config)
//     {
//         this.config = config;
//     }
// 
//     protected FluentdSender()
//     {
//         this(new FluentdSender.Config());
//     }
// 
//     public synchronized void send(ByteBuffer buffer)
//             throws IOException
//     {
//         sendInternalWithRestoreBufferPositions(Arrays.asList(buffer), null);
//     }
// 
//     public synchronized void send(List<ByteBuffer> buffers)
//             throws IOException
//     {
//         sendInternalWithRestoreBufferPositions(buffers, null);
//     }
// 
//     public void sendWithAck(List<ByteBuffer> buffers, String ackToken)
//             throws IOException
//     {
//         sendInternalWithRestoreBufferPositions(buffers, ackToken);
//     }
// 
//     private void sendInternalWithRestoreBufferPositions(List<ByteBuffer> buffers, String ackToken)
//             throws IOException
//     {
//         List<Integer> positions = new ArrayList<>(buffers.size());
//         for (ByteBuffer data : buffers) {
//             positions.add(data.position());
//         }
// 
//         try {
//             sendInternal(buffers, ackToken);
//         }
//         catch (Exception e) {
//             for (int i = 0; i < buffers.size(); i++) {
//                 buffers.get(i).position(positions.get(i));
//             }
// 
//             ErrorHandler errorHandler = config.getErrorHandler();
//             if (errorHandler != null) {
//                 try {
//                     errorHandler.handle(e);
//                 }
//                 catch (Exception ex) {
//                     LOG.warn("Failed to handle an error in the error handler {}", errorHandler, ex);
//                 }
//             }
// 
//             if (e instanceof IOException) {
//                 throw (IOException) e;
//             }
//             else {
//                 throw new IOException(e);
//             }
//         }
//     }
// 
//     public abstract boolean isAvailable();
// 
//     abstract protected void sendInternal(List<ByteBuffer> buffers, String ackToken)
//             throws IOException;
// 
//     public static class Config
//             extends Sender.Config
//     {
//     }
// }
// 
// Path: fluency-fluentd/src/main/java/org/komamitsu/fluency/fluentd/ingester/sender/RequestOption.java
// @JsonInclude(JsonInclude.Include.NON_NULL)
// public class RequestOption
// {
//     private final int size;
//     private final String chunk;
// 
//     public RequestOption(
//             @JsonProperty("size") int size,
//             @JsonProperty("chunk") String chunk)
//     {
//         this.size = size;
//         this.chunk = chunk;
//     }
// 
//     @JsonProperty("size")
//     public int getSize()
//     {
//         return size;
//     }
// 
//     @JsonProperty("chunk")
//     public String getChunk()
//     {
//         return chunk;
//     }
// 
//     @Override
//     public String toString()
//     {
//         return "RequestOption{" +
//                 "size=" + size +
//                 ", chunk=" + chunk +
//                 '}';
//     }
// }
// 
// Path: fluency-core/src/main/java/org/komamitsu/fluency/ingester/Ingester.java
// public interface Ingester
//         extends Closeable
// {
//     void ingest(String tag, ByteBuffer dataBuffer)
//             throws IOException;
// 
//     Sender getSender();
// }
// 
// Path: fluency-core/src/main/java/org/komamitsu/fluency/ingester/sender/Sender.java
// public interface Sender
// {
//     class Config
//     {
//         private ErrorHandler errorHandler;
// 
//         public ErrorHandler getErrorHandler()
//         {
//             return errorHandler;
//         }
// 
//         public void setErrorHandler(ErrorHandler errorHandler)
//         {
//             this.errorHandler = errorHandler;
//         }
// 
//         @Override
//         public String toString()
//         {
//             return "Config{" +
//                     "senderErrorHandler=" + errorHandler +
//                     '}';
//         }
//     }
// }


import java.util.UUID;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.komamitsu.fluency.fluentd.ingester.sender.FluentdSender;
import org.komamitsu.fluency.fluentd.ingester.sender.RequestOption;
import org.komamitsu.fluency.ingester.Ingester;
import org.komamitsu.fluency.ingester.sender.Sender;
import org.msgpack.core.MessagePack;
import org.msgpack.core.MessagePacker;
import org.msgpack.jackson.dataformat.MessagePackFactory;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.List;
    @Override
    public void ingest(String tag, ByteBuffer dataBuffer)
            throws IOException
    {
        ByteArrayOutputStream header = new ByteArrayOutputStream();
        MessagePacker messagePacker = MessagePack.newDefaultPacker(header);

        int dataLength = dataBuffer.limit();
        messagePacker.packArrayHeader(3);
        messagePacker.packString(tag);
        messagePacker.packRawStringHeader(dataLength);
        messagePacker.flush();

        ByteBuffer headerBuffer = ByteBuffer.wrap(header.toByteArray());

        if (config.isAckResponseMode()) {
            // The spec (https://github.com/fluent/fluentd/wiki/Forward-Protocol-Specification-v1#entry)
            // says to encode a 128 bit value as base64, but fluent-bit currently does something different
            // and in fluent-bit and fluentd there is no validation. So for now we keep it simple, see
            // discussion on issue #181.
            String token = UUID.randomUUID().toString();

            ByteBuffer optionBuffer = ByteBuffer.wrap(objectMapper.writeValueAsBytes(new RequestOption(dataLength, token)));
            List<ByteBuffer> buffers = Arrays.asList(headerBuffer, dataBuffer, optionBuffer);

            synchronized (sender) {
                sender.sendWithAck(buffers, token);
            }
        }
        else {
// Path: fluency-fluentd/src/main/java/org/komamitsu/fluency/fluentd/ingester/sender/FluentdSender.java
// public abstract class FluentdSender
//         implements Closeable, Sender
// {
//     private static final Logger LOG = LoggerFactory.getLogger(FluentdSender.class);
//     private final Config config;
// 
//     protected FluentdSender(Config config)
//     {
//         this.config = config;
//     }
// 
//     protected FluentdSender()
//     {
//         this(new FluentdSender.Config());
//     }
// 
//     public synchronized void send(ByteBuffer buffer)
//             throws IOException
//     {
//         sendInternalWithRestoreBufferPositions(Arrays.asList(buffer), null);
//     }
// 
//     public synchronized void send(List<ByteBuffer> buffers)
//             throws IOException
//     {
//         sendInternalWithRestoreBufferPositions(buffers, null);
//     }
// 
//     public void sendWithAck(List<ByteBuffer> buffers, String ackToken)
//             throws IOException
//     {
//         sendInternalWithRestoreBufferPositions(buffers, ackToken);
//     }
// 
//     private void sendInternalWithRestoreBufferPositions(List<ByteBuffer> buffers, String ackToken)
//             throws IOException
//     {
//         List<Integer> positions = new ArrayList<>(buffers.size());
//         for (ByteBuffer data : buffers) {
//             positions.add(data.position());
//         }
// 
//         try {
//             sendInternal(buffers, ackToken);
//         }
//         catch (Exception e) {
//             for (int i = 0; i < buffers.size(); i++) {
//                 buffers.get(i).position(positions.get(i));
//             }
// 
//             ErrorHandler errorHandler = config.getErrorHandler();
//             if (errorHandler != null) {
//                 try {
//                     errorHandler.handle(e);
//                 }
//                 catch (Exception ex) {
//                     LOG.warn("Failed to handle an error in the error handler {}", errorHandler, ex);
//                 }
//             }
// 
//             if (e instanceof IOException) {
//                 throw (IOException) e;
//             }
//             else {
//                 throw new IOException(e);
//             }
//         }
//     }
// 
//     public abstract boolean isAvailable();
// 
//     abstract protected void sendInternal(List<ByteBuffer> buffers, String ackToken)
//             throws IOException;
// 
//     public static class Config
//             extends Sender.Config
//     {
//     }
// }
// 
// Path: fluency-fluentd/src/main/java/org/komamitsu/fluency/fluentd/ingester/sender/RequestOption.java
// @JsonInclude(JsonInclude.Include.NON_NULL)
// public class RequestOption
// {
//     private final int size;
//     private final String chunk;
// 
//     public RequestOption(
//             @JsonProperty("size") int size,
//             @JsonProperty("chunk") String chunk)
//     {
//         this.size = size;
//         this.chunk = chunk;
//     }
// 
//     @JsonProperty("size")
//     public int getSize()
//     {
//         return size;
//     }
// 
//     @JsonProperty("chunk")
//     public String getChunk()
//     {
//         return chunk;
//     }
// 
//     @Override
//     public String toString()
//     {
//         return "RequestOption{" +
//                 "size=" + size +
//                 ", chunk=" + chunk +
//                 '}';
//     }
// }
// 
// Path: fluency-core/src/main/java/org/komamitsu/fluency/ingester/Ingester.java
// public interface Ingester
//         extends Closeable
// {
//     void ingest(String tag, ByteBuffer dataBuffer)
//             throws IOException;
// 
//     Sender getSender();
// }
// 
// Path: fluency-core/src/main/java/org/komamitsu/fluency/ingester/sender/Sender.java
// public interface Sender
// {
//     class Config
//     {
//         private ErrorHandler errorHandler;
// 
//         public ErrorHandler getErrorHandler()
//         {
//             return errorHandler;
//         }
// 
//         public void setErrorHandler(ErrorHandler errorHandler)
//         {
//             this.errorHandler = errorHandler;
//         }
// 
//         @Override
//         public String toString()
//         {
//             return "Config{" +
//                     "senderErrorHandler=" + errorHandler +
//                     '}';
//         }
//     }
// }


// Path: fluency-fluentd/src/main/java/org/komamitsu/fluency/fluentd/ingester/FluentdIngester.java
import java.util.UUID;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.komamitsu.fluency.fluentd.ingester.sender.FluentdSender;
import org.komamitsu.fluency.fluentd.ingester.sender.RequestOption;
import org.komamitsu.fluency.ingester.Ingester;
import org.komamitsu.fluency.ingester.sender.Sender;
import org.msgpack.core.MessagePack;
import org.msgpack.core.MessagePacker;
import org.msgpack.jackson.dataformat.MessagePackFactory;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.List;

    @Override
    public void ingest(String tag, ByteBuffer dataBuffer)
            throws IOException
    {
        ByteArrayOutputStream header = new ByteArrayOutputStream();
        MessagePacker messagePacker = MessagePack.newDefaultPacker(header);

        int dataLength = dataBuffer.limit();
        messagePacker.packArrayHeader(3);
        messagePacker.packString(tag);
        messagePacker.packRawStringHeader(dataLength);
        messagePacker.flush();

        ByteBuffer headerBuffer = ByteBuffer.wrap(header.toByteArray());

        if (config.isAckResponseMode()) {
            // The spec (https://github.com/fluent/fluentd/wiki/Forward-Protocol-Specification-v1#entry)
            // says to encode a 128 bit value as base64, but fluent-bit currently does something different
            // and in fluent-bit and fluentd there is no validation. So for now we keep it simple, see
            // discussion on issue #181.
            String token = UUID.randomUUID().toString();

            ByteBuffer optionBuffer = ByteBuffer.wrap(objectMapper.writeValueAsBytes(new RequestOption(dataLength, token)));
            List<ByteBuffer> buffers = Arrays.asList(headerBuffer, dataBuffer, optionBuffer);

            synchronized (sender) {
                sender.sendWithAck(buffers, token);
            }
        }
        else {
            ByteBuffer optionBuffer = ByteBuffer.wrap(objectMapper.writeValueAsBytes(new RequestOption(dataLength, null)));