package com.att.dmf.mr.backends.kafka;

import java.util.Arrays;
import java.util.Properties;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.FutureTask;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RunnableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.common.KafkaException;

import com.att.dmf.mr.backends.Consumer;

//import org.slf4j.Logger;
//import org.slf4j.LoggerFactory;

import com.att.eelf.configuration.EELFLogger;
import com.att.eelf.configuration.EELFManager;

/**
 * A consumer instance that's created per-request. These are stateless so that
 * clients can connect to this service as a proxy.
 * 
 * @author peter
 *
 */
public class KafkaConsumer implements Consumer {
	private enum State {
		OPENED, CLOSED
	}

	/**
	 * KafkaConsumer() is constructor. It has following 4 parameters:-
	 * 
	 * @param topic
	 * @param group
	 * @param id
	 * @param cc
	 * 
	 */

	public KafkaConsumer(String topic, String group, String id, Properties prop) throws Exception {
		fTopic = topic;
		fGroup = group;
		fId = id;
		// fConnector = cc;

		fCreateTimeMs = System.currentTimeMillis();
		fLastTouch = fCreateTimeMs;
		fPendingMsgs = new LinkedBlockingQueue<ConsumerRecord<String,String>> ();
		fLogTag = fGroup + "(" + fId + ")/" + fTopic;
		offset = 0;

		state = KafkaConsumer.State.OPENED;

		// final Map<String, Integer> topicCountMap = new HashMap<String,
		// Integer>();
		// topicCountMap.put(fTopic, 1);
		// log.info(fLogTag +" kafka Consumer started at "
		// +System.currentTimeMillis());
		// final Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap =
		// fConnector.createMessageStreams(topicCountMap);
		// final List<KafkaStream<byte[], byte[]>> streams =
		// consumerMap.get(fTopic);

		kConsumer = new org.apache.kafka.clients.consumer.KafkaConsumer<>(prop);
		// System.out.println("I am in Consumer APP " + topic + "-- " +
		// fConsumer);
		kConsumer.subscribe(Arrays.asList(topic));
		log.info(fLogTag + " kafka stream created in " + (System.currentTimeMillis() - fCreateTimeMs));
	System.out.println("-----id " +id);
	
		
	  try { ConsumerRecords<String, String> records =
					           kConsumer.poll(500); System.out.println("---" +
					            records.count());
					  
					            for (ConsumerRecord<String, String> record : records) {
					            System.out.printf("offset = %d, key = %s, value = %s",
					            record.offset(), record.key(), record.value()); String t =
					            record.value();
					  
					            }
		  }catch(Exception e){
			  System.out.println( e);
		  }
			System.out.println(fLogTag + " kafka stream created in " + (System.currentTimeMillis() - fCreateTimeMs));
			kConsumer.commitSync();
		//  fConsumer.close();  
	

		/*
		 * ConsumerRecords<String, String> records = fConsumer.poll(500);
		 * System.out.println("---" + records.count());
		 * 
		 * for (ConsumerRecord<String, String> record : records) {
		 * System.out.printf("offset = %d, key = %s, value = %s",
		 * record.offset(), record.key(), record.value()); String t =
		 * record.value();
		 * 
		 * }
		 * 
		 * 
		 * fConsumer.commitSync(); fConsumer.close();
		 */

		// fStream = streams.iterator().next();
	}
	
	
	
	private Consumer.Message makeMessage ( final ConsumerRecord<String,String> msg )
	{
		return new Consumer.Message()
		{
			@Override
			public long getOffset ()
			{
				return msg.offset ();
			}
			
			@Override
			public String getMessage ()
			{
				return new String ( msg.value () );
			}
		};
	}
	
	@Override
	public synchronized Consumer.Message nextMessage ()
	{
		
		try
		{
			if ( fPendingMsgs.size () > 0 )
			{
				return makeMessage ( fPendingMsgs.take () );
			}
		}
		catch ( InterruptedException x )
		{
			log.warn ( "After size>0, pending msg take() threw InterruptedException. Ignoring. (" + x.getMessage () + ")", x );
		}
		
		
			try
			{
				boolean foundMsgs = false;
				System.out.println("entering into pollingWWWWWWWWWWWWWWWWW");
				final ConsumerRecords<String,String> records = kConsumer.poll ( 100 );
				System.out.println("polling doneXXXXXXXXXXXXXXXXXXXXXXXXXXX....");
				for ( ConsumerRecord<String,String> record : records )
				{
					foundMsgs = true;
					fPendingMsgs.offer ( record );
				}
			
			}
			catch ( KafkaException x )
			{
				log.debug ( fLogTag + ": KafkaException " + x.getMessage () );
				
			}
			catch ( java.lang.IllegalStateException | java.lang.IllegalArgumentException x )
			{
				log.error ( fLogTag + ": Illegal state/arg exception in Kafka consumer; dropping stream. " + x.getMessage () );
			
			}
				
		return null;
	}
	
	

	/**
	 * getName() method returns string type value. returns 3 parameters in
	 * string:- fTopic,fGroup,fId
	 * 
	 * @Override
	 */
	public String getName() {
		return fTopic + " : " + fGroup + " : " + fId;
	}

	/**
	 * getCreateTimeMs() method returns long type value. returns fCreateTimeMs
	 * variable value
	 * 
	 * @Override
	 * 
	 */
	public long getCreateTimeMs() {
		return fCreateTimeMs;
	}

	public org.apache.kafka.clients.consumer.KafkaConsumer getConsumer() {
		return kConsumer;
	}

	/**
	 * getLastAccessMs() method returns long type value. returns fLastTouch
	 * variable value
	 * 
	 * @Override
	 * 
	 */
	public long getLastAccessMs() {
		return fLastTouch;
	}

	

	/**
	 * getOffset() method returns long type value. returns offset variable value
	 * 
	 * @Override
	 * 
	 */
	public long getOffset() {
		return offset;
	}

	/**
	 * commit offsets commitOffsets() method will be called on closed of
	 * KafkaConsumer.
	 * 
	 * @Override
	 * 
	 *
	 * 			public void commitOffsets() { if (getState() ==
	 *           KafkaConsumer.State.CLOSED) { log.warn("commitOffsets() called
	 *           on closed KafkaConsumer " + getName()); return; }
	 *           fConnector.commitOffsets(); }
	 */

	/**
	 * updating fLastTouch with current time in ms
	 */
	public void touch() {
		fLastTouch = System.currentTimeMillis();
	}

	/**
	 * getLastTouch() method returns long type value. returns fLastTouch
	 * variable value
	 * 
	 */
	public long getLastTouch() {
		return fLastTouch;
	}

	/**
	 * setting the kafkaConsumer state to closed
	 */
	public synchronized boolean close() {

		if (getState() == KafkaConsumer.State.CLOSED) {

			log.warn("close() called on closed KafkaConsumer " + getName());
			return true;
		}

		setState(KafkaConsumer.State.CLOSED);
		// fConnector.shutdown();
		boolean retVal = kafkaConnectorshuttask();
		return retVal;

	}

	/* time out if the kafka shutdown fails for some reason */

	private boolean kafkaConnectorshuttask() {
		Callable<Boolean> run = new Callable<Boolean>() {
			@Override
			public Boolean call() throws Exception {
				// your code to be timed
				try {
				System.out.println("consumer closing....." + kConsumer);
					kConsumer.close();
				} catch (Exception e) {
					log.info("@@@@@@Kafka Stream shutdown erorr occurred " + getName() + " " + e);
				}
				log.info("Kafka connection closure with in 15 seconds by a Executors task");
				return true;
			}
		};

		RunnableFuture future = new FutureTask(run);
		ExecutorService service = Executors.newSingleThreadExecutor();
		service.execute(future);
		Boolean result = null;
		try {
			result = (Boolean) future.get(15, TimeUnit.SECONDS); // wait 1
																	// second
		} catch (TimeoutException ex) {
			// timed out. Try to stop the code if possible.
			log.info("Timeout Occured - Kafka connection closure with in 15 seconds by a Executors task");
			future.cancel(true);
		} catch (Exception ex) {
			// timed out. Try to stop the code if possible.
			log.info("Timeout Occured - Kafka connection closure with in 15 seconds by a Executors task" + ex);
			future.cancel(true);
			return false;
		}
		service.shutdown();
		return true;
	}

	/**
	 * getConsumerGroup() returns Consumer group
	 * 
	 * @return
	 */
	public String getConsumerGroup() {
		return fGroup;
	}

	/**
	 * getConsumerId returns Consumer Id
	 * 
	 * @return
	 */
	public String getConsumerId() {
		return fId;
	}

	/**
	 * getState returns kafkaconsumer state
	 * 
	 * @return
	 */
	private KafkaConsumer.State getState() {
		return this.state;
	}

	/**
	 * setState() sets the kafkaConsumer state
	 * 
	 * @param state
	 */
	private void setState(KafkaConsumer.State state) {
		this.state = state;
	}

	// private ConsumerConnector fConnector;
	private final String fTopic;
	private final String fGroup;
	private final String fId;
	private final String fLogTag;
	// private final KafkaStream<byte[], byte[]> fStream;
	private final org.apache.kafka.clients.consumer.KafkaConsumer<String, String> kConsumer;
	private long fCreateTimeMs;
	private long fLastTouch;
	private long offset;
	private KafkaConsumer.State state;
	private static final EELFLogger log = EELFManager.getInstance().getLogger(KafkaConsumer.class);
	private final LinkedBlockingQueue<ConsumerRecord<String,String>> fPendingMsgs;
	// private static final Logger log =
	// LoggerFactory.getLogger(KafkaConsumer.class);

	@Override
	public void commitOffsets() {
		if (getState() == KafkaConsumer.State.CLOSED) {
			log.warn("commitOffsets() called on closed KafkaConsumer " + getName());
			return;
		}
		kConsumer.commitSync();
		// fConsumer.close();

	}



	@Override
	public void setOffset(long offsetval) {
		// TODO Auto-generated method stub
		offset = offsetval;
	}
}