<commit_before>#include "MqttPublisher.h"

MqttPublisher::MqttPublisher(string brokerUrl, string clientID)
        : brokerURL(brokerUrl),
          clientID(clientID)
{
    client = nullptr;
}
MqttPublisher::MqttPublisher(string brokerUrl, string userName, string password, string clientID)
        : brokerURL(brokerUrl),
    userName(userName),
    password(password),
    clientID(clientID)
{
    client = nullptr;
}

MqttPublisher::~MqttPublisher() {
    stop();
}

void MqttPublisher::start(bool asyncMode) {
    MqttPublisher::asyncMode = asyncMode;
    if(asyncMode)
        setupAsyncClient();
    else
        setupClient();
}

void onDisconnect(void* context, MQTTAsync_successData* response)
{
    printf("Successful disconnection\n");
}

void MqttPublisher::stop() {
    // Disconnect
    std::cout << "Disconnecting..." << std::flush;
    if(client != nullptr) {
        MQTTClient_disconnect(client, 10000);
        MQTTClient_destroy(&client);
    } else if(asyncClient != nullptr) {
        MQTTAsync_disconnectOptions opts = MQTTAsync_disconnectOptions_initializer;
        int rc;

        opts.onSuccess = onDisconnect;
        opts.context = client;

        if ((rc = MQTTAsync_disconnect(asyncClient, &opts)) != MQTTASYNC_SUCCESS)
        {
            printf("Failed to start sendMessage, return code %d\n", rc);
        }
        MQTTAsync_destroy(&asyncClient);
    }
    client = nullptr;
    asyncClient = nullptr;
    std::cout << "OK" << std::endl;
}

extern "C" void onSend(void* context, MQTTAsync_successData* response)
{
    MQTTAsync client = (MQTTAsync)context;
#ifdef PRINT_DEBUG
    printf("Message with token value %d delivery confirmed\n", response->token);
#endif
}

void MqttPublisher::queueForPublish(string topicName, MqttQOS qos, byte *payload, size_t len) {
    MQTTAsync_message pubmsg = MQTTAsync_message_initializer;
    MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
    int rc;

    opts.onSuccess = onSend;
    opts.context = asyncClient;

    pubmsg.payload = payload;
    pubmsg.payloadlen = len;
    pubmsg.qos = qos;
    pubmsg.retained = 0;
    if ((rc = MQTTAsync_sendMessage(asyncClient, topicName.c_str(), &pubmsg, &opts)) != MQTTASYNC_SUCCESS) {<commit_msg>Only set the onFailure sendMessage callback<commit_after>#include <MQTTAsync.h>
#include "MqttPublisher.h"

MqttPublisher::MqttPublisher(string brokerUrl, string clientID)
        : brokerURL(brokerUrl),
          clientID(clientID)
{
    client = nullptr;
}
MqttPublisher::MqttPublisher(string brokerUrl, string userName, string password, string clientID)
        : brokerURL(brokerUrl),
    userName(userName),
    password(password),
    clientID(clientID)
{
    client = nullptr;
}

MqttPublisher::~MqttPublisher() {
    stop();
}

void MqttPublisher::start(bool asyncMode) {
    MqttPublisher::asyncMode = asyncMode;
    if(asyncMode)
        setupAsyncClient();
    else
        setupClient();
}

void onDisconnect(void* context, MQTTAsync_successData* response)
{
    printf("Successful disconnection\n");
}

void MqttPublisher::stop() {
    // Disconnect
    std::cout << "Disconnecting..." << std::flush;
    if(client != nullptr) {
        MQTTClient_disconnect(client, 10000);
        MQTTClient_destroy(&client);
    } else if(asyncClient != nullptr) {
        MQTTAsync_disconnectOptions opts = MQTTAsync_disconnectOptions_initializer;
        int rc;

        opts.onSuccess = onDisconnect;
        opts.context = client;

        if ((rc = MQTTAsync_disconnect(asyncClient, &opts)) != MQTTASYNC_SUCCESS)
        {
            printf("Failed to start sendMessage, return code %d\n", rc);
        }
        MQTTAsync_destroy(&asyncClient);
    }
    client = nullptr;
    asyncClient = nullptr;
    std::cout << "OK" << std::endl;
}

extern "C" void onSend(void* context, MQTTAsync_successData* response)
{
    MQTTAsync client = (MQTTAsync)context;
#ifdef PRINT_DEBUG
    printf("Message with token value %d delivery confirmed\n", response->token);
#endif
}
extern "C" void onSendFailure(void* context,  MQTTAsync_failureData* response) {
    printf("delivery failed, token: %d \n", response->token);
}

void MqttPublisher::queueForPublish(string topicName, MqttQOS qos, byte *payload, size_t len) {
    MQTTAsync_message pubmsg = MQTTAsync_message_initializer;
    MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
    int rc;

    opts.onSuccess = onSend;
    opts.onFailure = onSendFailure;
    opts.context = asyncClient;

    pubmsg.payload = payload;
    pubmsg.payloadlen = len;
    pubmsg.qos = qos;
    pubmsg.retained = 0;
    if ((rc = MQTTAsync_sendMessage(asyncClient, topicName.c_str(), &pubmsg, &opts)) != MQTTASYNC_SUCCESS) {
{'fraction_non_alphanumeric': 0.11313559322033899, 'fraction_numerical': 0.002542372881355932, 'mean_word_length': 3.744723618090452, 'pattern_counts': {'":': 0, '<': 20, '<?xml version=': 0, '>': 7, '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': '6921196', 'n_tokens_mistral': 1588, 'n_tokens_neox': 1504, 'n_words': 308}