mirror of
https://github.com/OrcaSlicer/OrcaSlicer.git
synced 2026-09-16 13:32:44 +00:00
153 lines
7.1 KiB
C++
153 lines
7.1 KiB
C++
#ifndef slic3r_OrcaMqttConnection_hpp_
|
|
#define slic3r_OrcaMqttConnection_hpp_
|
|
|
|
#include <boost/asio/ssl.hpp>
|
|
#include <boost/beast/core.hpp>
|
|
#include <boost/beast/ssl.hpp>
|
|
#include <boost/beast/websocket.hpp>
|
|
#include <atomic>
|
|
#include <condition_variable>
|
|
#include <deque>
|
|
#include <functional>
|
|
#include <map>
|
|
#include <memory>
|
|
#include <mutex>
|
|
#include <set>
|
|
#include <string>
|
|
#include <thread>
|
|
#include <utility>
|
|
#include <vector>
|
|
#include <cstddef>
|
|
#include <cstdint>
|
|
|
|
namespace Slic3r {
|
|
|
|
// Minimal MQTT 3.1.1 codec + WebSocket transport (ws:// and wss://), shared by the
|
|
// LAN (OrcaSonar) and cloud (per-printer) printer connections. Both PUBLISH
|
|
// commands to device/<id>/request and SUBSCRIBE device/<id>/report; Config is the
|
|
// only per-transport difference.
|
|
class OrcaMqttConnection
|
|
{
|
|
public:
|
|
using TokenProvider = std::function<std::string()>;
|
|
using MessageHandler = std::function<void(const std::string&, const std::string&)>;
|
|
using StateHandler = std::function<void(bool connected, bool initial)>;
|
|
|
|
struct Endpoint { std::string host; std::string port; std::string target; };
|
|
|
|
struct Config {
|
|
std::string url;
|
|
bool use_tls = false;
|
|
TokenProvider bearer_provider; // set => bearer on WS upgrade, CONNECT creds omitted
|
|
std::string username;
|
|
std::string password;
|
|
std::string client_id = "OrcaSlicer";
|
|
int keepalive_seconds = 60;
|
|
};
|
|
|
|
static bool parse_endpoint(const std::string& url, Endpoint& endpoint);
|
|
|
|
// Build an MQTT 3.1.1 CONNECT packet. Clean-session is always set; the
|
|
// username/password connect flags and payload fields are added only when
|
|
// username is non-empty (the cloud form authenticates via a bearer on the
|
|
// WebSocket upgrade and omits CONNECT credentials). Public for unit tests.
|
|
static std::vector<uint8_t> make_connect_packet(const std::string& client_id,
|
|
const std::string& username,
|
|
const std::string& password,
|
|
int keepalive_seconds);
|
|
|
|
// Topic-string helpers for the per-device request/report channels and the
|
|
// MQTT 3.1.1 PUBLISH / SUBSCRIBE / UNSUBSCRIBE packet builders. All public
|
|
// for unit tests. make_publish_packet emits QoS 0 (no packet identifier).
|
|
static std::string request_topic(const std::string& dev_id); // "device/<id>/request"
|
|
static std::string report_topic(const std::string& dev_id); // "device/<id>/report"
|
|
static std::vector<uint8_t> make_publish_packet(const std::string& topic, const std::string& payload);
|
|
static std::vector<uint8_t> make_subscribe_packet(uint16_t packet_id, const std::string& topic, uint8_t qos);
|
|
static std::vector<uint8_t> make_unsubscribe_packet(uint16_t packet_id, const std::string& topic);
|
|
|
|
~OrcaMqttConnection();
|
|
|
|
bool start(const Config& config, MessageHandler on_message, StateHandler on_state);
|
|
void stop();
|
|
// True while the worker thread is alive (connected OR retrying). Lets callers
|
|
// avoid restarting a healthy connection.
|
|
bool is_running() const;
|
|
// True once CONNACK has been received and the socket has not since dropped.
|
|
bool is_connected() const { return connected.load(); }
|
|
bool subscribe(const std::string& dev_id);
|
|
bool unsubscribe(const std::string& dev_id);
|
|
// Last MQTT CONNACK return code: 0 ok, 1..5 refusal, -1 none seen this attempt.
|
|
int last_connack_rc() const { return m_last_connack_rc.load(); }
|
|
void clear_subscriptions();
|
|
bool send_request(const std::string& dev_id, const std::string& payload);
|
|
|
|
private:
|
|
// The endpoint may be either a TLS (wss://) or a plaintext (ws://) WebSocket;
|
|
// Connection holds whichever one is engaged and the ws_* helpers below
|
|
// dispatch on it.
|
|
using TlsWebSocket = boost::beast::websocket::stream<
|
|
boost::asio::ssl::stream<boost::beast::tcp_stream>>;
|
|
using PlainWebSocket = boost::beast::websocket::stream<boost::beast::tcp_stream>;
|
|
struct Connection;
|
|
|
|
static void append_string(std::vector<uint8_t>& packet, const std::string& value);
|
|
static void prepend_remaining_length(std::vector<uint8_t>& packet, size_t length);
|
|
static std::vector<uint8_t> make_ping_packet();
|
|
|
|
// Transport dispatch: each forwards to conn.wss (TLS) or conn.ws (plaintext).
|
|
void ws_write(Connection& conn, const std::vector<uint8_t>& packet); // locks write_mutex
|
|
std::size_t ws_read(Connection& conn, boost::beast::flat_buffer& buffer, boost::system::error_code& ec);
|
|
void ws_handshake(Connection& conn, const Config& config, const Endpoint& endpoint);
|
|
void ws_close(Connection& conn);
|
|
// Emit a queued SUBSCRIBE/UNSUBSCRIBE on the live socket right now (from the
|
|
// caller thread), so a selection change is applied without waiting for the
|
|
// blocking read loop to next return. No-op if no CONNACKed socket exists yet
|
|
// (the worker sends the set on connect). The WebSocket is never dropped for a
|
|
// subscription change.
|
|
void flush_subscription_change();
|
|
void connect_and_read();
|
|
void send_current_subscriptions(Connection& conn);
|
|
void send_pending_subscriptions(Connection& conn);
|
|
void handle_packet(const std::string& packet);
|
|
void notify_state(bool is_now_connected);
|
|
void run();
|
|
|
|
std::atomic_bool stopping{true};
|
|
std::atomic_int reconnect_delay_seconds{1};
|
|
// Serialises the whole of start() and stop() against each other, so the UI
|
|
// thread's stop() (disconnect / dtor) cannot race the connect thread's start()
|
|
// into a concurrent worker.join(). Recursive because start() calls stop().
|
|
std::recursive_mutex lifecycle_mutex;
|
|
std::thread worker;
|
|
std::mutex mutex;
|
|
std::mutex connection_mutex;
|
|
std::mutex write_mutex; // serialises every websocket write (worker + caller threads)
|
|
std::shared_ptr<Connection> active_connection;
|
|
std::condition_variable initial_cv;
|
|
std::condition_variable state_cv;
|
|
Config current_config;
|
|
MessageHandler on_message;
|
|
StateHandler on_state;
|
|
// Full report-topic strings ("device/<id>/report"), not bare device ids.
|
|
std::set<std::string> subscriptions;
|
|
std::set<std::string> pending_subscriptions;
|
|
std::set<std::string> pending_unsubscriptions;
|
|
// Requests for a subscribed device wait until the corresponding SUBACK is
|
|
// received. Otherwise an immediate pushall response can be published by
|
|
// the broker before this client is actually subscribed to the report topic.
|
|
std::set<std::string> acknowledged_subscriptions;
|
|
std::map<uint16_t, std::string> pending_subscribe_packets;
|
|
std::deque<std::pair<std::string, std::string>> pending_requests;
|
|
std::atomic<uint16_t> next_packet_id{1};
|
|
std::atomic<int> m_last_connack_rc{-1};
|
|
uint64_t m_attempt_number{0}; // worker-thread diagnostic sequence
|
|
std::string m_connection_stage; // worker-thread diagnostic stage
|
|
bool initial_result{false};
|
|
bool initial_completed{false};
|
|
std::atomic_bool connected{false};
|
|
};
|
|
|
|
} // namespace Slic3r
|
|
|
|
#endif // slic3r_OrcaMqttConnection_hpp_
|