fwt_software/src/mqtt/MqttControlChannel.cpp

#include "fgc/MqttControlChannel.h"

#include "fgc/ChannelTopics.h"

#include "fgc/Logger.h"

#include <chrono>

namespace fgc {

namespace {
constexpr int  kQos = 1;
const auto     kConnectTimeout = std::chrono::seconds(5);
const auto     kPublishTimeout = std::chrono::seconds(3);
}  // namespace

MqttControlChannel::MqttControlChannel(const std::string& broker_uri, const std::string& tower,
                                       const std::string& user, const std::string& password)
    : broker_uri_(broker_uri),
      tower_(tower),
      topic_status_(towerTopic(tower, kTopicStatusCode)),
      topic_cam_event_(towerTopic(tower, kTopicCamEvent)),
      topic_command_(towerTopic(tower, kTopicCommand)),
      topic_command_reply_(towerTopic(tower, kTopicCommandReply)),
      client_(broker_uri, tower) {
    conn_opts_.set_keep_alive_interval(20);
    conn_opts_.set_clean_session(true);
    conn_opts_.set_automatic_reconnect(true);  // replaces the old manual reconnect loop
    if (!user.empty()) conn_opts_.set_user_name(user);
    if (!password.empty()) conn_opts_.set_password(password);
    client_.set_callback(*this);
}

MqttControlChannel::~MqttControlChannel() { disconnect(); }

bool MqttControlChannel::connect() {
    try {
        client_.connect(conn_opts_, nullptr, *this)->wait_for(kConnectTimeout);
        connected_ = client_.is_connected();
    } catch (const mqtt::exception& e) {
        LOG_ERROR << "MQTT connect failed (" << broker_uri_ << "): " << e.what();
        connected_ = false;
    }
    if (connected_)
        LOG_INFO << "MQTT connected to " << broker_uri_ << " as '" << tower_ << "'";
    return connected_;
}

void MqttControlChannel::disconnect() {
    if (!client_.is_connected()) return;
    try {
        client_.disconnect()->wait();
    } catch (const mqtt::exception& e) {
        LOG_WARN << "MQTT disconnect: " << e.what();
    }
    connected_ = false;
}

bool MqttControlChannel::connected() const { return connected_; }

void MqttControlChannel::connected(const std::string& /*cause*/) {
    LOG_INFO << "MQTT (re)connected; subscribing";
    history_.noteConnect();
    client_.subscribe(topic_command_, kQos);
    connected_ = true;
    // Say the mode again: whoever watches StatusCode may have missed changes
    // while the link was down. Not waited on - this runs on Paho's callback
    // thread, which is the one that would deliver the acknowledgement.
    std::string status;
    {
        std::lock_guard<std::mutex> lock(mutex_);
        status = last_status_;
    }
    if (!status.empty()) {
        try {
            auto msg = mqtt::make_message(topic_status_, status);
            msg->set_qos(kQos);
            msg->set_retained(true);
            client_.publish(msg);
            history_.record(/*outbound=*/true, Delivery::Ok, topic_status_, status);
        } catch (const mqtt::exception& e) {
            LOG_WARN << "MQTT status republish failed: " << e.what();
        }
    }
}

void MqttControlChannel::connection_lost(const std::string& cause) {
    LOG_WARN << "MQTT connection lost" << (cause.empty() ? "" : ": " + cause);
    history_.noteLoss();
    connected_ = false;
    // Paho will auto-reconnect when configured; the broker callback re-subscribes.
}

void MqttControlChannel::on_failure(const mqtt::token& /*tok*/) {
    LOG_WARN << "MQTT action failed";
}

void MqttControlChannel::message_arrived(mqtt::const_message_ptr msg) {
    const std::string topic   = msg->get_topic();
    const std::string payload = msg->to_string();
    LOG_TRACE_CAT(LogCat::Mqtt) << "RX  " << topic << ' ' << payload;
    history_.record(/*outbound=*/false, Delivery::Ok, topic, payload);
    std::lock_guard<std::mutex> lock(mutex_);
    if (topic == topic_command_) {
        // Queued for the control thread, which runs it exactly as a typed
        // command. Retained ones are queued too, flagged, so they are answered
        // (and refused) rather than silently swallowed.
        RemoteCommand rc = parseCommandPayload(payload);
        rc.received_ms = std::chrono::duration_cast<std::chrono::milliseconds>(
                             std::chrono::system_clock::now().time_since_epoch()).count();
        rc.retained = msg->is_retained();
        // A flood - a stuck publisher - must not grow without bound.
        if (commands_.size() < 256) commands_.push_back(std::move(rc));
        else LOG_WARN << "MQTT command queue full; dropped: " << payload;
    }
}

std::vector<RemoteCommand> MqttControlChannel::takeCommands() {
    std::lock_guard<std::mutex> lock(mutex_);
    std::vector<RemoteCommand> out;
    out.swap(commands_);
    return out;
}

void MqttControlChannel::publishCommandReply(const CommandReply& reply) {
    // Not retained: a reply answers one request, and a retained one would be
    // handed to every web page that subscribes later, as if it were new.
    publish(topic_command_reply_, commandReplyJson(reply), /*retained=*/false);
}

void MqttControlChannel::publish(const std::string& topic, const std::string& payload,
                                 bool retained) {
    if (!client_.is_connected()) {
        // Discarded: there is no queue and no retry, so this message is simply
        // gone. Recording it is the only trace it ever existed.
        history_.record(/*outbound=*/true, Delivery::Dropped, topic, payload);
        return;
    }
    LOG_TRACE_CAT(LogCat::Mqtt) << "PUB " << topic << ' ' << payload;
    try {
        auto msg = mqtt::make_message(topic, payload);
        msg->set_qos(kQos);
        msg->set_retained(retained);
        client_.publish(msg)->wait_for(kPublishTimeout);
        history_.record(/*outbound=*/true, Delivery::Ok, topic, payload);
    } catch (const mqtt::exception& e) {
        LOG_WARN << "MQTT publish failed: " << e.what();
        history_.record(/*outbound=*/true, Delivery::Failed, topic, payload);
    }
}

void MqttControlChannel::publishStatus(int code) {
    const std::string payload = std::to_string(code);
    {
        std::lock_guard<std::mutex> lock(mutex_);
        last_status_ = payload;
    }
    history_.noteStatus(payload);
    publish(topic_status_, payload);
}

void MqttControlChannel::publishCamEvent(const CamEvent& e) {
    publish(topic_cam_event_, camEventJson(e));
}

}  // namespace fgc