// Path: src/main/java/io/jafka/producer/SyncProducer.java
// @ThreadSafe
// public class SyncProducer implements Closeable {
// 
//     private final Logger logger = LoggerFactory.getLogger(SyncProducer.class);
// 
//     //private static final RequestKeys RequestKey = RequestKeys.Produce;//0
// 
//     /////////////////////////////////////////////////////////////////////
//     private final SyncProducerConfig config;
// 
//     private final BlockingChannel blockingChannel;
// 
//     private final Object lock = new Object();
// 
//     private volatile boolean shutdown = false;
// 
//     private final String host;
// 
//     private final int port;
// 
//     public SyncProducer(SyncProducerConfig config) {
//         super();
//         this.config = config;
//         this.host = config.getHost();
//         this.port = config.getPort();
//         //
//         this.blockingChannel = new BlockingChannel(host, port, BlockingChannel.DEFAULT_BUFFER_SIZE, config.bufferSize, config.socketTimeoutMs);
//     }
// 
//     public void send(String topic, ByteBufferMessageSet message) {
//         send(topic, ProducerRequest.RandomPartition, message);
//     }
// 
//     public void send(String topic, int partition, ByteBufferMessageSet messages) {
//         messages.verifyMessageSize(config.maxMessageSize);
//         send(new ProducerRequest(topic, partition, messages));
//     }
// 
//     private void send(Request request) {
//         synchronized (lock) {
//             long startTime = System.nanoTime();
//             int written = -1;
//             try {
//                 written = connect().send(request);
//             } catch (IOException e) {
//                 // no way to tell if write succeeded. Disconnect and re-throw exception to let client handle retry
//                 disconnect();
//                 throw new RuntimeException(e);
//             } finally {
//                 if (logger.isDebugEnabled()) {
//                     logger.debug(format("write %d bytes data to %s:%d", written, host, port));
//                 }
//             }
//             final long endTime = System.nanoTime();
//             SyncProducerStats.recordProduceRequest(endTime - startTime);
//         }
//     }
// 
//     private BlockingChannel connect() {
//         if (!blockingChannel.isConnected() && !shutdown) {
//             try {
//                 blockingChannel.connect();
//             } catch (IOException ioe) {
//                 throw new RuntimeException(ioe.getMessage(), ioe);
//             } finally {
//                 if (!blockingChannel.isConnected()) {
//                     disconnect();
//                 }
//             }
//         }
//         return blockingChannel;
//     }
// 
//     private void disconnect() {
//         if (blockingChannel.isConnected()) {
//             logger.info("Disconnecting from " + config.getHost() + ":" + config.getPort());
//             blockingChannel.disconnect();
//         }
//     }
// 
//     public void close() {
//         synchronized (lock) {
//             try {
//                 disconnect();
//             } finally {
//                 shutdown = true;
//             }
//         }
//     }
// 
// 
//     public void multiSend(List<ProducerRequest> produces) {
//         for (ProducerRequest request : produces) {
//             request.messages.verifyMessageSize(config.maxMessageSize);
//         }
//         send(new MultiProducerRequest(produces));
//     }
// 
// }


import io.jafka.common.IllegalQueueStateException;
import io.jafka.producer.SyncProducer;
import io.jafka.producer.serializer.Encoder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import static java.lang.String.format;
/**
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package io.jafka.producer.async;




/**
 * @author adyliu (imxylz@gmail.com)
 * @since 1.0
 */
public class ProducerSendThread<T> extends Thread {

    final String threadName;

    final BlockingQueue<QueueItem<T>> queue;

    final Encoder<T> serializer;

    final SyncProducer underlyingProducer;

    final EventHandler<T> eventHandler;

    final CallbackHandler<T> callbackHandler;

    final long queueTime;

    final int batchSize;

    private final Logger logger = LoggerFactory.getLogger(ProducerSendThread.class);

    /////////////////////////////////////////////////////////////////////
    private final CountDownLatch shutdownLatch = new CountDownLatch(1);

    private volatile boolean shutdown = false;

    public ProducerSendThread(String threadName, //
                              BlockingQueue<QueueItem<T>> queue, //
                              Encoder<T> serializer, //
// Path: src/main/java/io/jafka/producer/SyncProducer.java
// @ThreadSafe
// public class SyncProducer implements Closeable {
// 
//     private final Logger logger = LoggerFactory.getLogger(SyncProducer.class);
// 
//     //private static final RequestKeys RequestKey = RequestKeys.Produce;//0
// 
//     /////////////////////////////////////////////////////////////////////
//     private final SyncProducerConfig config;
// 
//     private final BlockingChannel blockingChannel;
// 
//     private final Object lock = new Object();
// 
//     private volatile boolean shutdown = false;
// 
//     private final String host;
// 
//     private final int port;
// 
//     public SyncProducer(SyncProducerConfig config) {
//         super();
//         this.config = config;
//         this.host = config.getHost();
//         this.port = config.getPort();
//         //
//         this.blockingChannel = new BlockingChannel(host, port, BlockingChannel.DEFAULT_BUFFER_SIZE, config.bufferSize, config.socketTimeoutMs);
//     }
// 
//     public void send(String topic, ByteBufferMessageSet message) {
//         send(topic, ProducerRequest.RandomPartition, message);
//     }
// 
//     public void send(String topic, int partition, ByteBufferMessageSet messages) {
//         messages.verifyMessageSize(config.maxMessageSize);
//         send(new ProducerRequest(topic, partition, messages));
//     }
// 
//     private void send(Request request) {
//         synchronized (lock) {
//             long startTime = System.nanoTime();
//             int written = -1;
//             try {
//                 written = connect().send(request);
//             } catch (IOException e) {
//                 // no way to tell if write succeeded. Disconnect and re-throw exception to let client handle retry
//                 disconnect();
//                 throw new RuntimeException(e);
//             } finally {
//                 if (logger.isDebugEnabled()) {
//                     logger.debug(format("write %d bytes data to %s:%d", written, host, port));
//                 }
//             }
//             final long endTime = System.nanoTime();
//             SyncProducerStats.recordProduceRequest(endTime - startTime);
//         }
//     }
// 
//     private BlockingChannel connect() {
//         if (!blockingChannel.isConnected() && !shutdown) {
//             try {
//                 blockingChannel.connect();
//             } catch (IOException ioe) {
//                 throw new RuntimeException(ioe.getMessage(), ioe);
//             } finally {
//                 if (!blockingChannel.isConnected()) {
//                     disconnect();
//                 }
//             }
//         }
//         return blockingChannel;
//     }
// 
//     private void disconnect() {
//         if (blockingChannel.isConnected()) {
//             logger.info("Disconnecting from " + config.getHost() + ":" + config.getPort());
//             blockingChannel.disconnect();
//         }
//     }
// 
//     public void close() {
//         synchronized (lock) {
//             try {
//                 disconnect();
//             } finally {
//                 shutdown = true;
//             }
//         }
//     }
// 
// 
//     public void multiSend(List<ProducerRequest> produces) {
//         for (ProducerRequest request : produces) {
//             request.messages.verifyMessageSize(config.maxMessageSize);
//         }
//         send(new MultiProducerRequest(produces));
//     }
// 
// }


// Path: src/main/java/io/jafka/producer/async/ProducerSendThread.java
import io.jafka.common.IllegalQueueStateException;
import io.jafka.producer.SyncProducer;
import io.jafka.producer.serializer.Encoder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import static java.lang.String.format;

/**
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package io.jafka.producer.async;




/**
 * @author adyliu (imxylz@gmail.com)
 * @since 1.0
 */
public class ProducerSendThread<T> extends Thread {

    final String threadName;

    final BlockingQueue<QueueItem<T>> queue;

    final Encoder<T> serializer;

    final SyncProducer underlyingProducer;

    final EventHandler<T> eventHandler;

    final CallbackHandler<T> callbackHandler;

    final long queueTime;

    final int batchSize;

    private final Logger logger = LoggerFactory.getLogger(ProducerSendThread.class);

    /////////////////////////////////////////////////////////////////////
    private final CountDownLatch shutdownLatch = new CountDownLatch(1);

    private volatile boolean shutdown = false;

    public ProducerSendThread(String threadName, //
                              BlockingQueue<QueueItem<T>> queue, //
                              Encoder<T> serializer, //
                              SyncProducer underlyingProducer,//