94 lines
3.6 KiB
C++
94 lines
3.6 KiB
C++
#pragma once
|
|
|
|
#include "fgc/ChannelHistory.h"
|
|
#include "fgc/IControlChannel.h"
|
|
|
|
#include <atomic>
|
|
#include <condition_variable>
|
|
#include <deque>
|
|
#include <memory>
|
|
#include <mutex>
|
|
#include <string>
|
|
#include <thread>
|
|
#include <utility>
|
|
#include <vector>
|
|
|
|
#include <mqtt/async_client.h>
|
|
|
|
namespace fgc {
|
|
|
|
// Real control channel over MQTT (Eclipse Paho C++). Subscribes to the tower's
|
|
// Command topic and publishes StatusCode / CamEvent / CommandReply. Topic scheme: GGS/FWT/<tower>/... (see docs/mqtt-api.md).
|
|
// Publishes are QoS 1; retained, except command replies.
|
|
//
|
|
// Unlike the old MQTTClient it does NOT spin a busy-wait thread; Paho's async
|
|
// client delivers callbacks on its own threads.
|
|
class MqttControlChannel : public IControlChannel,
|
|
public virtual mqtt::callback,
|
|
public virtual mqtt::iaction_listener {
|
|
public:
|
|
// broker_uri: a full Paho serverURI, e.g. "tcp://10.0.0.5:1883"
|
|
// (see NetworkConfig::brokerUri).
|
|
MqttControlChannel(const std::string& broker_uri, const std::string& tower,
|
|
const std::string& user, const std::string& password);
|
|
~MqttControlChannel() override;
|
|
|
|
bool connect() override;
|
|
void disconnect() override;
|
|
bool connected() const override;
|
|
|
|
void publishStatus(int code) override;
|
|
void publishCamEvent(const CamEvent& event) override;
|
|
void publishHealth(const HealthReport& report) override;
|
|
std::vector<RemoteCommand> takeCommands() override;
|
|
void publishCommandReply(const CommandReply& reply) override;
|
|
|
|
std::vector<ChannelMessage> recentMessages() const override { return history_.recent(); }
|
|
ChannelStats channelStats() const override { return history_.stats(); }
|
|
|
|
private:
|
|
// mqtt::callback
|
|
void connected(const std::string& cause) override;
|
|
void connection_lost(const std::string& cause) override;
|
|
void message_arrived(mqtt::const_message_ptr msg) override;
|
|
// mqtt::iaction_listener
|
|
void on_failure(const mqtt::token& tok) override;
|
|
void on_success(const mqtt::token& tok) override {}
|
|
|
|
void publish(const std::string& topic, const std::string& payload, bool retained = true);
|
|
|
|
// CamEvent and Health go out through a small queue with its own thread, so a
|
|
// slow link (each publish waits up to 3 s for the broker) never holds an
|
|
// image-save worker or the control loop. Bounded: when full, the oldest
|
|
// waiting message is dropped and recorded as such.
|
|
void post(const std::string& topic, const std::string& payload);
|
|
void runOutbox();
|
|
void stopOutbox(); // sends what is waiting (a few seconds at most), then ends
|
|
|
|
ChannelHistory history_;
|
|
std::string broker_uri_;
|
|
std::string tower_;
|
|
std::string topic_status_;
|
|
std::string topic_cam_event_;
|
|
std::string topic_health_;
|
|
std::string topic_command_;
|
|
std::string topic_command_reply_;
|
|
|
|
mqtt::async_client client_;
|
|
mqtt::connect_options conn_opts_;
|
|
std::atomic<bool> connected_{false};
|
|
|
|
std::mutex mutex_;
|
|
std::string last_status_; // republished on reconnect (under mutex_)
|
|
std::string last_health_; // republished on reconnect (under mutex_)
|
|
std::vector<RemoteCommand> commands_; // received, not yet taken (under mutex_)
|
|
|
|
std::mutex outbox_mutex_;
|
|
std::condition_variable outbox_cv_;
|
|
std::deque<std::pair<std::string, std::string>> outbox_; // (topic, payload)
|
|
bool outbox_stop_ = false;
|
|
std::thread outbox_thread_;
|
|
};
|
|
|
|
} // namespace fgc
|