package com.sagacloud.broadcast;

import com.sagacloud.cache.cachedObject.QueryObjects;
import com.sagacloud.utils.OssUtils;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.jms.*;

/**
 * Created by wo on 10/22/19.
 */
public class ActiveMsgHandler implements MsgHandler{
    private  ConnectionFactory factory;
    private  Connection connection;
    private  Session session;
    private  MessageProducer messageProducer;
    private static final Logger log = LoggerFactory.getLogger(MessageHandler.class);

    public  void init(){
        try {
            factory = new ActiveMQConnectionFactory(OssUtils.getProperty("jms.userName"), OssUtils.getProperty("jms.password"), OssUtils.getProperty("jms.brokerurl"));
            connection = factory.createConnection();
            connection.start();
            session = connection.createSession(false, TopicSession.AUTO_ACKNOWLEDGE);
            Topic topic = session.createTopic(OssUtils.getProperty("jms.topic"));
            messageProducer = session.createProducer(topic);
            messageProducer.setTimeToLive(1000 * 60 * 60 * 2);
            consume();
        } catch (JMSException e) {
            log.error("Failed to establish JMS topic connection.", e);
            System.exit(1);
        }
    }

    public  void publish(String msg) {
        int tryCount = 0;
        while (tryCount < 3){
            try {
                TextMessage textMessage = session.createTextMessage(msg);
                messageProducer.send(textMessage);
                break;
            } catch (JMSException e) {
                log.error("Failed to publish message : " + msg, e);
                synchronized (MessageHandler.class) {
                    try {
                        session.close();
                    } catch (JMSException e1) {
                        log.error("close session error, ", e1);
                    }
                    init();
                }
                tryCount++;
                if(tryCount < 3){
                    log.info("Try to publish again, tryCount: " + tryCount);
                }
            }
        }

    }

    public  void consume()  {
        try {
            MessageConsumer consumer = session.createConsumer(session.createTopic(OssUtils.getProperty("jms.topic")));
            MessageListener listner = new MessageListener() {
                public void onMessage(Message message) {
                    try {
                        if (message instanceof TextMessage) {
                            TextMessage textMessage = (TextMessage) message;
                            log.debug("Received message: " + textMessage.getText() );
                            MsgHandler.clearCache(textMessage.getText());
                            QueryObjects.clearCache(textMessage.getText());
                        }
                    } catch (JMSException e) {
                        log.error("Failed to handle message, ", e);
                    }
                }
            };

            consumer.setMessageListener(listner);
        } catch (JMSException e) {
            log.error("", e);
        }


    }

}
