package com.javahonk.messaging;


import java.util.List;
import java.util.Map;

import javax.annotation.PostConstruct;
import javax.jms.Connection;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import com.javahonk.interfaces.TradePublisher;

/**
 * @author Java Honk
 *
 */
public class TibcoPublisherConsumer implements TradePublisher, MessageListener {

	private static final Logger logger = LogManager.getLogger(TibcoPublisherConsumer.class.getName());


	private ConnectionFactory connectionFactory;
	private String msgType;
	private String inputQueueName;
	private String outputQueueName;
	
	private Connection connection;
	private Session session;
	private MessageProducer msgProducer;
	private Queue outputQueue;
	
	private MessageConsumer msgConsumer;
		
	public TibcoPublisherConsumer(ConnectionFactory connectionFactory,
			String msgType, 
			String inputQueueName, 
			String outputQueueName) {
		this.connectionFactory = connectionFactory;
		this.msgType = msgType;
		this.inputQueueName = inputQueueName;
		this.outputQueueName = outputQueueName;
		
		logger.info("Tibco inputQueue: {}", inputQueueName);
		logger.info("Tibco outputQueue: {}", outputQueueName);
		logger.info("Tibco msgType: {}", msgType);
	}
	
	@PostConstruct
	public void init() {
		try {
			connection = connectionFactory.createConnection();			
			session = connectionFactory.createSession(connection);
			
			Queue inputQueue = session.createQueue(inputQueueName);
			msgProducer = session.createProducer(inputQueue);
			
			outputQueue = session.createQueue(outputQueueName);			
			msgConsumer = session.createConsumer(outputQueue);
			msgConsumer.setMessageListener(this);
			
			connection.start();
		} catch (JMSException e) {
			logger.error("Error creating connection to tibco: ", e);
		} 
	}

	@Override
	public void sendTrade(String values)  {
		try {
			TextMessage msg = session.createTextMessage(values);
			msg.setJMSType(msgType);

			logger.info("Sending Trade {}", values);
			
			msgProducer.send(msg);
		} catch (JMSException ex) {
			ex.printStackTrace();
		}
		
	}
	
	@Override
	public void onMessage(Message msg) {

		try {
			if (msg instanceof TextMessage) {
				TextMessage txtMsg = (TextMessage) msg;
				logger.info("Received response {}", msg);

				Object msgTextObj = txtMsg.getText();
				
				logger.info("Received response {}", msgTextObj.toString());
				System.out.println(msgTextObj.toString());
				
				@SuppressWarnings("unchecked")
				List<Map<String, Object>> list = (List<Map<String, Object>>)msgTextObj;

				for(Map<String, Object> map: list) {
					if (isValidMap(map)) {
						
						String successString = (String) map.get("status");
						String processState = successString.equalsIgnoreCase("success") ? "EXPIRED" : "ERROR";
						logger.info("Process state:{}", processState);
						
					} else {
						logger.error("Invalid map received in response or subscriber not set.");
					}
				}
			}

		} catch (Exception e) {
			logger.error("Failed to read response", e);
		}

	}
	
	@SuppressWarnings("rawtypes")
	private boolean isValidMap(Map responseMap) {
		return responseMap.containsKey("status") 
				&& responseMap.containsKey("key"); 
	}
	
	
}
