<commit_before>package gew.PubSub.mqtt;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;

import java.io.UnsupportedEncodingException;
import java.util.concurrent.BlockingQueue;

/**
 * MQTT Callback class, handle arriving message to message queue, etc.
 * @author Jason/<NAME>
 * @since 2017-08-01
 */
public class ClientCallback implements MqttCallback
{
    private boolean queueEnable;
    private MqttClient currentClient;
    private int RECONNECT_TRIAL = 10;
    private boolean autoReconnect = true;
    private BlockingQueue<String[]> messageQueue;       // Producer; [0]: Topic, [1]: Payload

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

    public ClientCallback(MqttClient currentClient)
    {
        this.currentClient = currentClient;
        queueEnable = false;
    }

    ClientCallback(MqttClient currentClient, BlockingQueue<String[]> messageQueue)
    {
        this.currentClient = currentClient;
        this.messageQueue = messageQueue;
        queueEnable = true;
    }

    public boolean isAutoReconnect() { return autoReconnect; }
    public void setAutoReconnect(boolean autoReconnect) { this.autoReconnect = autoReconnect; }

    @Override
    public void connectionLost(Throwable throwable) {
        logger.info("=> MQTT Connection Lost: " + throwable.getMessage());
        if(!autoReconnect) {
            while(RECONNECT_TRIAL != 0 && !currentClient.isConnected()) {
                try {
                    logger.info("=> System is trying to reconnect... (" + RECONNECT_TRIAL + ")");
                    currentClient.connect();            // or invoke reconnect()
                    if (currentClient.isConnected()) {
                        logger.info("=> System Reconnected!");
                        RECONNECT_TRIAL = 10;
                        break;
                    }
                } catch (MqttException e) {
//                    RECONNECT_TRIAL++;
                    logger.error("=> MQTT Reconnect: " + e.toString());
                }
                RECONNECT_TRIAL--;
                try {
                    Thread.sleep(10000);
                } catch (InterruptedException e) {
                    logger.warn("=> MQTT Reconnection Interrupted: " + e.toString());
                }
            }
        }
    }

    @Override
    public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
        if(queueEnable)
        {
            String[] data = new String[2];
            String msg = null;
            try {
                msg = new String(mqttMessage.getPayload(), "UTF-8");
            } catch (UnsupportedEncodingException e) {
                logger.error(e.toString());
            }
            data[0] = topic;
            data[1] = msg;
            try {
                messageQueue.put(data);
            } catch(InterruptedException err) {
                logger.info(err.toString());
            }
        } else {
            String msg = null;
            try {
                msg = new String(mqttMessage.getPayload(), "UTF-8");

            } catch (UnsupportedEncodingException e) {
                logger.error(e.toString());
            }
            logger.info("=> [" + topic + "] " + msg);
        }
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
        if(RECONNECT_TRIAL < 10)
            RECONNECT_TRIAL = 10;
    }
}<commit_msg>Modified mechanism when system not in auto-reconnect mode<commit_after>package gew.PubSub.mqtt;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;

import java.io.UnsupportedEncodingException;
import java.util.concurrent.BlockingQueue;

/**
 * MQTT Callback class, handle arriving message to message queue, etc.
 * @author Jason/<NAME>
 * @since 2017-08-01
 */
public class ClientCallback implements MqttCallback
{
    private boolean queueEnable;
    private MqttClient currentClient;
    private int RECONNECT_TRIAL = 10;
    private boolean autoReconnect = true;
    private BlockingQueue<String[]> messageQueue;       // Producer; [0]: Topic, [1]: Payload

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

    public ClientCallback(MqttClient currentClient)
    {
        this.currentClient = currentClient;
        queueEnable = false;
    }

    ClientCallback(MqttClient currentClient, BlockingQueue<String[]> messageQueue)
    {
        this.currentClient = currentClient;
        this.messageQueue = messageQueue;
        queueEnable = true;
    }

    public boolean isAutoReconnect() { return autoReconnect; }
    public void setAutoReconnect(boolean autoReconnect) { this.autoReconnect = autoReconnect; }

    @Override
    public void connectionLost(Throwable throwable) {
        logger.info("=> MQTT Connection Lost: " + throwable.getMessage());
        if(!autoReconnect) {
            while(RECONNECT_TRIAL != 0 && !currentClient.isConnected()) {
                try {
                    logger.info("=> System is trying to reconnect... (" + RECONNECT_TRIAL + ")");
                    currentClient.connect();            // or invoke reconnect()
                    if (currentClient.isConnected()) {
                        logger.info("=> System Reconnected!");
                        RECONNECT_TRIAL = 10;
                        break;
                    }
                } catch (MqttException e) {
//                    RECONNECT_TRIAL++;
                    logger.error("=> MQTT Reconnect: " + e.toString());
                }
                RECONNECT_TRIAL--;
                try {
                    Thread.sleep(10000);
                } catch (InterruptedException e) {
                    logger.warn("=> MQTT Reconnection Interrupted: " + e.toString());
                }
            }
        }
    }

    @Override
    public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
        if(queueEnable)
        {
            String[] data = new String[2];
            String msg = null;
            try {
                msg = new String(mqttMessage.getPayload(), "UTF-8");
            } catch (UnsupportedEncodingException e) {
                logger.error(e.toString());
            }
            data[0] = topic;
            data[1] = msg;
            try {
                messageQueue.put(data);
            } catch(InterruptedException err) {
                logger.info(err.toString());
            }
        } else {
            String msg = null;
            try {
                msg = new String(mqttMessage.getPayload(), "UTF-8");

            } catch (UnsupportedEncodingException e) {
                logger.error(e.toString());
            }
            logger.info("=> [" + topic + "] " + msg);
        }
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
        if(!autoReconnect && RECONNECT_TRIAL < 10)
            RECONNECT_TRIAL = 10;
    }
}
{'fraction_non_alphanumeric': 0.1029917924278528, 'fraction_numerical': 0.009531374106433678, 'mean_word_length': 1.9662347860227718, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 21, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '15937', 'n_tokens_mistral': 2121, 'n_tokens_neox': 2048, 'n_words': 424}