diff --git a/CMakeLists.txt b/CMakeLists.txt index eaeb46b..84d7478 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -61,6 +61,7 @@ add_library(fgc_core STATIC src/core/Config.cpp src/core/IniEdit.cpp src/core/CameraStall.cpp + src/core/RemoteCommands.cpp src/core/Paths.cpp src/core/SingleInstance.cpp src/core/Logger.cpp diff --git a/README.md b/README.md index ea5fe56..985eff0 100644 --- a/README.md +++ b/README.md @@ -295,10 +295,12 @@ echo $? # 0 = PASS, 4 = FAIL, 5 = could not start (the log says why) ## Everyday commands -The same commands work at the `:` prompt in the dashboard and in the plain -console. The ground station, over MQTT, has a narrower remote control: it can -switch the tower between the automatic sweep and pointing at a bearing it -names (`ControlCode`, `target_HDG` - see [docs/mqtt-api.md](docs/mqtt-api.md)). +The same commands work at the `:` prompt in the dashboard, in the plain +console, and **over MQTT**: a web maintenance interface can send any of them on +`GGS/FWT//Command` and read the answer on `CommandReply` - the lines the +command logged, and whether it succeeded. That topic is always live and +withholds nothing (`exit` included), so who may publish to it is the broker's +job. Details, payloads and safety rules: [docs/mqtt-api.md](docs/mqtt-api.md). | Command | Does | |---|---| diff --git a/docs/mqtt-api.md b/docs/mqtt-api.md index 6612c31..9c1f52a 100644 --- a/docs/mqtt-api.md +++ b/docs/mqtt-api.md @@ -11,7 +11,7 @@ GGS/FWT//... ``` `` comes from `config.ini` (`General.tower_name`). All messages use **QoS 1**. Published messages -are **retained**. +are **retained**, except command replies. ## Connection @@ -106,14 +106,105 @@ Both `hdg` and `pit` are sampled **when the frame arrives**, not when its JPEG X finishes, so a slow encode or a backed-up queue cannot attribute a later pose to earlier pixels. +## Remote commands: `Command` / `CommandReply` + +For development and maintenance: **any console command** - everything that can +be typed at the dashboard's `:` prompt or the plain console - can be sent on +`Command`, and the tower answers each one on `CommandReply`. It is meant for a +web maintenance interface; a tower in normal service simply never receives any. + +| Topic | Direction | Payload | +|---|---|---| +| `GGS/FWT//Command` | to the tower | `{"id":"","cmd":""}`, or a bare command line | +| `GGS/FWT//CommandReply` | from the tower | `{"id":"...","cmd":"...","level":"info\|warn\|error","lines":[...],"truncated":false}` - **not retained** | + +**How a command runs.** Exactly as if typed: the same handler +(`Application::handleCommand()`), on the same control thread, with the same +refusals. `gimbal move` during a sweep is refused as it is at the console +(send `stop` first); a command sent while a test or homing runs gets "ignored +... in progress". Commands run in the order they arrive, one at a time, each +answered before the next starts. + +**The reply.** +- `id` is echoed from the request, so a page can match answers to questions. + Choose a fresh one per request. +- `lines` are what the command logged while it ran - what the console would + have shown in response, timestamps and all. `help` returns the help text, + `gimbal dump` the dump, a refusal the reason. +- `level` is the most severe of those lines: `info` (done), `warn` (refused or + questionable), `error` (failed, or not run - see below). +- `gimbal dump`'s answer comes back from the firmware a moment later; its reply + waits for it (up to 5 s) and carries the dump itself. Commands behind it wait + their turn. +- Long jobs - `gimbal home`, `test ...`, `camera reset` - are answered when they + have **started**. Their outcome is not published here; ask afterwards + (`gimbal dump`), or read the tower's log - a self-test writes its report to + `logs/test_*.log`. +- Some commands answer with nothing (`lines: []`): `gimbal status` only asks the + firmware to refresh the dashboard's telemetry, and `exit` is answered just + before the program stops. +- At most 200 lines / 16 KB; `truncated` says when an answer was cut. + +**What is refused, and never run** (answered with `level: error`): +- a **retained** message - the broker re-delivers retained messages on every + reconnect, so acting on one would re-run it each time the tower restarts. + Publish commands without the retain flag; +- an unreadable payload (bad JSON, no `cmd`, more than one line). + +**Repeats.** QoS 1 can deliver a message twice. The last 64 ids are +remembered: a repeated id is answered again but not run again. A bare command +has no id, so it cannot be recognised as a repeat. + +**Every remote command is logged** before it runs, as +`remote command []: `, so the tower's log records who drove +it, and how. + +> **Careful - this topic is always live, and nothing is withheld.** Anyone who +> can publish to `GGS/FWT//Command` can do anything the console can: +> move the gimbal, change settings, send raw text to the firmware +> (`gimbal raw`), and stop the program (`exit`). **After `exit`, a tower with the +> boot service does not come back by itself** - someone has to restart the +> service or reboot it. Who may command a tower is decided by who may publish +> to its broker: set that with the broker's users and permissions. + +### Reaching a tower + +The tower subscribes on **its own broker** (RabbitMQ with the MQTT plugin, in +`~/zkms-client` on the tower). Today that broker has one path to the ground +station, and it runs outwards only: `GGS.FWT.*.CamEvent` is bound into a +`CamEvents` queue that a shovel (`rabbitmq/shovel-declaration.sh`) carries to +10.11.12.13. **Nothing carries messages from the ground station to a tower** - +which is also true of `ControlCode` and `target_HDG`. + +So a web interface either: +- connects to each tower's broker directly (over the VPN) - `Command` and + `CommandReply` both live there, and nothing else is needed; or +- goes through the ground station, which needs, in zkms-client: a + `CommandReply` queue, binding and shovel out (mirroring `CamEvents`), and an + inbound shovel from the ground station to the tower for `Command` (and + `ControlCode` / `target_HDG`). + +### Example + +```text +# to GGS/FWT/dev/Command +{"id":"maint-17","cmd":"gimbal status"} + +# from GGS/FWT/dev/CommandReply +{"id":"maint-17","cmd":"gimbal status","level":"info", + "lines":["16:40:02.114 [INFO ] remote ..."],"truncated":false} +``` + ## Topic summary ``` subscribe: GGS/FWT//target_HDG (int heading) GGS/FWT//ControlCode (0 = auto sweep, 1 = directed) + GGS/FWT//Command (any console command: JSON {id, cmd} or a bare line) publish: GGS/FWT//StatusCode (echoed control code) GGS/FWT//CamEvent (JSON: fwt, cam, hdg×10, pit×10, time-ms) + GGS/FWT//CommandReply (JSON: id, cmd, level, lines, truncated - not retained) ``` ## Local testing @@ -124,4 +215,6 @@ Point `zkms_server_ip` at a local broker and exercise the topics with Mosquitto: mosquitto_sub -t 'GGS/FWT/ExampleTower/#' -v # watch everything for the tower mosquitto_pub -t 'GGS/FWT/ExampleTower/ControlCode' -m '0' mosquitto_pub -t 'GGS/FWT/ExampleTower/target_HDG' -m '180' +mosquitto_pub -q 1 -t 'GGS/FWT/ExampleTower/Command' -m '{"id":"t1","cmd":"gimbal status"}' +mosquitto_pub -q 1 -t 'GGS/FWT/ExampleTower/Command' -m 'help' # a bare line works too ``` diff --git a/include/fgc/ChannelTopics.h b/include/fgc/ChannelTopics.h index 71fa987..b3ee543 100644 --- a/include/fgc/ChannelTopics.h +++ b/include/fgc/ChannelTopics.h @@ -18,10 +18,27 @@ inline constexpr const char* kTopicTargetHdg = "target_HDG"; inline constexpr const char* kTopicControlCode = "ControlCode"; inline constexpr const char* kTopicStatusCode = "StatusCode"; inline constexpr const char* kTopicCamEvent = "CamEvent"; +// Any console command, in; its answer, out. See docs/mqtt-api.md. +inline constexpr const char* kTopicCommand = "Command"; +inline constexpr const char* kTopicCommandReply = "CommandReply"; std::string towerTopic(const std::string& tower, const std::string& leaf); // Payload builders. One definition each, used by both channels. std::string camEventJson(const CamEvent& e); +// A Command payload: {"id":"...","cmd":"..."} (a flat JSON object of strings; +// other keys ignored), or a bare command line with no id - for testing from a +// shell. On a payload that is neither, `error` says why. +RemoteCommand parseCommandPayload(const std::string& payload); + +// A CommandReply payload. Caps the lines at kReplyMaxLines / kReplyMaxBytes and +// sets `truncated` when it had to cut. +inline constexpr size_t kReplyMaxLines = 200; +inline constexpr size_t kReplyMaxBytes = 16 * 1024; +std::string commandReplyJson(const CommandReply& r); + +// A JSON string literal, quotes included, with the escapes JSON requires. +std::string jsonString(const std::string& s); + } // namespace fgc diff --git a/include/fgc/IControlChannel.h b/include/fgc/IControlChannel.h index 3c22ecd..7310600 100644 --- a/include/fgc/IControlChannel.h +++ b/include/fgc/IControlChannel.h @@ -23,6 +23,25 @@ struct CamEvent { long long timestamp_ms = 0; // Unix epoch ms; matches the image filename }; +// A console command that arrived over MQTT, on GGS/FWT//Command. It is +// run by the same handleCommand() as a typed one, on the control thread. +struct RemoteCommand { + std::string id; // chosen by the sender, echoed in the reply; may be empty + std::string cmd; // the command line, exactly as typed at the console + long long received_ms = 0; + bool retained = false; // a retained message: never run (it would re-run on reconnect) + std::string error; // non-empty: the payload could not be read; reply, do not run +}; + +// The tower's answer, on GGS/FWT//CommandReply. +struct CommandReply { + std::string id; + std::string cmd; + std::string level = "info"; // most severe line: info / warn / error + std::vector lines; // what the command logged while it ran + bool truncated = false; +}; + // What became of an outbound message. Dropped and Failed are separated because // they mean different things operationally: Dropped is "the link was down and we // discarded it silently", Failed is "the broker or library rejected it". @@ -68,6 +87,11 @@ public: // acted on once. virtual ControlCommand poll() = 0; + // Commands received since the last call, oldest first; clears them. And the + // answer to one. Defaults: none / dropped, for a channel that takes no commands. + virtual std::vector takeCommands() { return {}; } + virtual void publishCommandReply(const CommandReply& /*reply*/) {} + // Recent traffic and counters for the connectivity ('m') view. Default: none, // so an implementation that keeps no history needs no changes. virtual std::vector recentMessages() const { return {}; } diff --git a/include/fgc/Logger.h b/include/fgc/Logger.h index 44ce2b0..6ca50b5 100644 --- a/include/fgc/Logger.h +++ b/include/fgc/Logger.h @@ -84,6 +84,19 @@ public: using FileSink = std::function; static void setFileSink(FileSink sink); + // --- tap ---------------------------------------------------------------- + // + // A THIRD independent slot: sees every line that reaches the console (the + // same lines, the same filter), whether the console is the dashboard's pane + // or stdout. It exists so a command that arrived over MQTT can be answered + // with what it logged, without taking over either sink above. + // + // The same WARNING as the file sink applies: it runs under this logger's + // non-recursive mutex, so it must only record - never log, never block, + // never publish. Pass an empty function to remove it. + using TapSink = std::function; + static void setTap(TapSink tap); + // The file's own threshold, independent of the console's, so the dashboard // can stay at info while the file keeps debug. // diff --git a/include/fgc/MqttControlChannel.h b/include/fgc/MqttControlChannel.h index 1984942..9eab51b 100644 --- a/include/fgc/MqttControlChannel.h +++ b/include/fgc/MqttControlChannel.h @@ -7,15 +7,16 @@ #include #include #include +#include #include namespace fgc { // Real control channel over MQTT (Eclipse Paho C++). Subscribes to the tower's -// target_HDG / ControlCode topics and publishes StatusCode / CamEvent. Topic -// scheme: GGS/FWT//... (see docs/mqtt-api.md). Publishes are QoS 1, -// retained. +// target_HDG / ControlCode / Command topics and publishes StatusCode / CamEvent / +// CommandReply. Topic scheme: GGS/FWT//... (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. @@ -36,6 +37,8 @@ public: void publishStatus(int code) override; void publishCamEvent(const CamEvent& event) override; ControlCommand poll() override; + std::vector takeCommands() override; + void publishCommandReply(const CommandReply& reply) override; std::vector recentMessages() const override { return history_.recent(); } ChannelStats channelStats() const override { return history_.stats(); } @@ -49,7 +52,7 @@ private: 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); + void publish(const std::string& topic, const std::string& payload, bool retained = true); ChannelHistory history_; std::string broker_uri_; @@ -58,6 +61,8 @@ private: std::string topic_control_code_; std::string topic_status_; std::string topic_cam_event_; + std::string topic_command_; + std::string topic_command_reply_; mqtt::async_client client_; mqtt::connect_options conn_opts_; @@ -65,6 +70,7 @@ private: std::mutex mutex_; ControlCommand sub_data_; + std::vector commands_; // received, not yet taken (under mutex_) }; } // namespace fgc diff --git a/include/fgc/RemoteCommands.h b/include/fgc/RemoteCommands.h new file mode 100644 index 0000000..0b58935 --- /dev/null +++ b/include/fgc/RemoteCommands.h @@ -0,0 +1,71 @@ +#pragma once + +// Running a console command that arrived over MQTT, and building its answer. +// +// The command goes through exactly the function a typed one does - the caller +// passes it in - on the control thread, so its lock-outs, refusals and output +// are the console's. What this adds is what a remote caller needs and a person +// at the console does not: +// - the answer: every line the command logged while it ran, and the most +// severe level among them, sent back on CommandReply; +// - a retained message is never run - it would re-run on every reconnect; +// - a repeated id (QoS 1 may deliver twice) is answered again, not re-run; +// - an audit line in the log for every remote command, before it runs; +// - for a command whose answer comes back later from the firmware (`gimbal +// dump`), the reply waits for that answer - see Await. + +#include "fgc/IControlChannel.h" + +#include +#include +#include +#include +#include +#include +#include + +namespace fgc { + +// What a command's answer still waits for once the command has returned: the +// firmware answers `gimbal dump` on the serial thread, a moment later. While +// waiting, lines from any thread that contain one of `collect` join the reply, +// and the first that contains one of `ends` completes it. +struct Await { + std::vector collect; + std::vector ends; + long long timeout_ms = 5000; +}; + +class RemoteCommandRunner { +public: + using Handler = std::function; + + // Run `rc` through `handle` (the console's command handler). Returns the + // reply to publish - or nothing yet, when `await` is given and the command + // did not warn: then the reply comes from poll(). Must be called on the + // thread that runs the commands: of what is logged while `handle` runs, only + // the calling thread's lines count as this command's output. + std::optional run(const RemoteCommand& rc, const Handler& handle, + const Await* await = nullptr, long long now_ms = 0); + + // A reply is still waiting for its answer; run nothing else until poll() + // has returned it, so replies keep the order of their commands. + bool waiting() const { return pending_ != nullptr; } + + // The waiting reply, once its answer has arrived or the wait timed out. + std::optional poll(long long now_ms); + + // How many ids are remembered for de-duplication. + static constexpr size_t kRememberIds = 64; + + struct Capture; // the tap's buffer; defined in RemoteCommands.cpp + +private: + CommandReply finish(const RemoteCommand& rc, Capture& cap); + + struct Pending; + std::shared_ptr pending_; + std::deque> recent_; // id -> reply, newest last +}; + +} // namespace fgc diff --git a/include/fgc/mock/NullControlChannel.h b/include/fgc/mock/NullControlChannel.h index f2fc764..831d034 100644 --- a/include/fgc/mock/NullControlChannel.h +++ b/include/fgc/mock/NullControlChannel.h @@ -7,6 +7,7 @@ #include #include +#include namespace fgc { @@ -40,6 +41,25 @@ public: } ControlCommand poll() override { return {}; } + // Commands: none arrive by themselves; inject() stands in for the broker, + // so --no-mqtt runs and tests can drive the remote-command path. + void inject(RemoteCommand rc) { + history_.record(/*outbound=*/false, Delivery::Ok, towerTopic(tower_, kTopicCommand), + rc.id.empty() ? rc.cmd : "{\"id\":" + jsonString(rc.id) + ",\"cmd\":" + + jsonString(rc.cmd) + "}"); + pending_.push_back(std::move(rc)); + } + std::vector takeCommands() override { + std::vector out; + out.swap(pending_); + return out; + } + void publishCommandReply(const CommandReply& r) override { + replies_.push_back(r); + drop(kTopicCommandReply, commandReplyJson(r)); + } + const std::vector& replies() const { return replies_; } + std::vector recentMessages() const override { return history_.recent(); } ChannelStats channelStats() const override { return history_.stats(); } @@ -51,6 +71,8 @@ private: std::string tower_; ChannelHistory history_; + std::vector pending_; + std::vector replies_; }; } // namespace fgc diff --git a/src/core/Application.cpp b/src/core/Application.cpp index 32337e2..a002941 100644 --- a/src/core/Application.cpp +++ b/src/core/Application.cpp @@ -26,6 +26,7 @@ #include "fgc/ScanGrid.h" #include "fgc/SerialMotorController.h" #include "fgc/TestReport.h" +#include "fgc/RemoteCommands.h" #include "fgc/TestRunner.h" #include "fgc/mock/MockCameraSource.h" #include "fgc/mock/MockEnvSensor.h" @@ -45,11 +46,13 @@ #include #include #include +#include #include #include #include #include #include +#include #include #include #include @@ -280,6 +283,8 @@ struct Application::Impl { bool heading_setup_active_ = false; std::atomic running{true}; + RemoteCommandRunner remote_; // commands over MQTT (GGS/FWT//Command) + std::deque remote_queue_; // taken from the channel, not yet run // --test: what run() returns, and where the one-shot test has got to. int exit_code_ = 0; bool resume_capture_after_test_ = false; // was capturing when a test began @@ -2561,6 +2566,45 @@ struct Application::Impl { } } + // Commands that arrived over MQTT: run exactly as typed ones, answered on + // CommandReply. After the typed ones in the same tick, one at a time and each + // answered before the next runs - so an `exit` is still answered. A command + // whose answer the firmware sends later holds the queue until it arrives. + void drainRemoteCommands() { + const long long now = std::chrono::duration_cast( + std::chrono::steady_clock::now().time_since_epoch()).count(); + if (remote_.waiting()) { + auto reply = remote_.poll(now); + if (!reply) return; + channel->publishCommandReply(*reply); + } + for (auto& rc : channel->takeCommands()) remote_queue_.push_back(std::move(rc)); + while (!remote_queue_.empty() && running) { + const RemoteCommand rc = std::move(remote_queue_.front()); + remote_queue_.pop_front(); + const std::optional await = remoteAwait(rc.cmd); + auto reply = remote_.run( + rc, [this](const std::string& line) { handleCommand(line); }, + await ? &*await : nullptr, now); + if (!reply) return; // waiting; poll() answers it on a later tick + channel->publishCommandReply(*reply); + } + } + + // The firmware answers these on the serial thread after the command has + // returned; the console shows that answer when it lands, so the remote reply + // waits for it. + static std::optional remoteAwait(const std::string& line) { + std::istringstream in(line); + std::string a, b; + in >> a >> b; + if (lowerStr(a) == "gimbal" && lowerStr(b) == "dump") + return Await{{"firmware dump", "Motor controller:"}, + {"firmware dump:", "firmware dump still incomplete"}, + 5000}; + return std::nullopt; + } + void drainCommands() { std::queue local; { @@ -2751,6 +2795,7 @@ struct Application::Impl { oneshot_since_ = std::chrono::steady_clock::now(); while (running && !g_shutdown_requested.load()) { drainCommands(); + drainRemoteCommands(); driveOneShotTest(); answerFirmwareNeeds(); // the firmware asks; hardware.ini answers pollBackgroundResults(); // apply calibration result / emit DIAG summary+log diff --git a/src/core/ChannelTopics.cpp b/src/core/ChannelTopics.cpp index 280aa91..1157814 100644 --- a/src/core/ChannelTopics.cpp +++ b/src/core/ChannelTopics.cpp @@ -1,5 +1,8 @@ #include "fgc/ChannelTopics.h" +#include +#include + namespace fgc { std::string towerTopic(const std::string& tower, const std::string& leaf) { @@ -13,4 +16,143 @@ std::string camEventJson(const CamEvent& e) { ", \"time\":" + std::to_string(e.timestamp_ms) + " }"; } +std::string jsonString(const std::string& s) { + std::string out = "\""; + for (unsigned char c : s) { + switch (c) { + case '"': out += "\\\""; break; + case '\\': out += "\\\\"; break; + case '\n': out += "\\n"; break; + case '\r': out += "\\r"; break; + case '\t': out += "\\t"; break; + default: + if (c < 0x20) { + char b[8]; std::snprintf(b, sizeof(b), "\\u%04x", c); out += b; + } else { + out += static_cast(c); // UTF-8 passes through as is + } + } + } + return out + "\""; +} + +namespace { + +// A minimal reader for one flat JSON object whose values are strings. Enough +// for {"id":"...","cmd":"..."}; anything else it reports rather than guesses. +struct FlatJson { + const std::string& s; + size_t i = 0; + std::string err; + + void ws() { while (i < s.size() && std::isspace(static_cast(s[i]))) ++i; } + bool eat(char c) { ws(); if (i < s.size() && s[i] == c) { ++i; return true; } return false; } + + bool str(std::string& out) { + ws(); + if (i >= s.size() || s[i] != '"') { err = "expected a string"; return false; } + ++i; + while (i < s.size() && s[i] != '"') { + char c = s[i++]; + if (c != '\\') { out += c; continue; } + if (i >= s.size()) break; + char e = s[i++]; + switch (e) { + case 'n': out += '\n'; break; + case 'r': out += '\r'; break; + case 't': out += '\t'; break; + case 'b': out += '\b'; break; + case 'f': out += '\f'; break; + case 'u': { + if (i + 4 > s.size()) { err = "bad \\u escape"; return false; } + unsigned cp = static_cast(std::stoul(s.substr(i, 4), nullptr, 16)); + i += 4; + // Encode as UTF-8 (the BMP is enough for a command line). + if (cp < 0x80) out += static_cast(cp); + else if (cp < 0x800) { + out += static_cast(0xC0 | (cp >> 6)); + out += static_cast(0x80 | (cp & 0x3F)); + } else { + out += static_cast(0xE0 | (cp >> 12)); + out += static_cast(0x80 | ((cp >> 6) & 0x3F)); + out += static_cast(0x80 | (cp & 0x3F)); + } + break; + } + default: out += e; break; // \" \\ \/ + } + } + if (i >= s.size()) { err = "unterminated string"; return false; } + ++i; + return true; + } + + // Skip a non-string value (number, true/false/null) so other keys may exist. + void skipScalar() { + ws(); + while (i < s.size() && s[i] != ',' && s[i] != '}') ++i; + } +}; + +} // namespace + +RemoteCommand parseCommandPayload(const std::string& payload) { + RemoteCommand rc; + size_t a = payload.find_first_not_of(" \t\r\n"); + if (a == std::string::npos) { rc.error = "empty command"; return rc; } + if (payload[a] != '{') { + // A bare command line: what a person types into mosquitto_pub. + size_t b = payload.find_last_not_of(" \t\r\n"); + rc.cmd = payload.substr(a, b - a + 1); + if (rc.cmd.find('\n') != std::string::npos) rc.error = "one command per message"; + return rc; + } + FlatJson j{payload}; + j.i = a; + try { + if (!j.eat('{')) { rc.error = "not a JSON object"; return rc; } + bool first = true; + while (!j.eat('}')) { + if (!first && !j.eat(',')) { rc.error = "expected , or }"; return rc; } + first = false; + std::string key; + if (!j.str(key)) { rc.error = "JSON: " + j.err; return rc; } + if (!j.eat(':')) { rc.error = "expected :"; return rc; } + j.ws(); + if (j.i < payload.size() && payload[j.i] == '"') { + std::string val; + if (!j.str(val)) { rc.error = "JSON: " + j.err; return rc; } + if (key == "id") rc.id = val; + else if (key == "cmd") rc.cmd = val; + } else { + if (key == "id" || key == "cmd") { rc.error = key + " must be a string"; return rc; } + j.skipScalar(); + } + } + } catch (const std::exception&) { + rc.error = "unreadable JSON"; + return rc; + } + if (rc.cmd.find_first_not_of(" \t") == std::string::npos) rc.error = "no \"cmd\" in the message"; + else if (rc.cmd.find('\n') != std::string::npos) rc.error = "one command per message"; + return rc; +} + +std::string commandReplyJson(const CommandReply& r) { + std::string lines = "["; + size_t bytes = 0, n = 0; + bool truncated = r.truncated; + for (const auto& l : r.lines) { + if (n == kReplyMaxLines || bytes + l.size() > kReplyMaxBytes) { truncated = true; break; } + if (n) lines += ","; + lines += jsonString(l); + bytes += l.size(); + ++n; + } + lines += "]"; + return "{\"id\":" + jsonString(r.id) + ",\"cmd\":" + jsonString(r.cmd) + + ",\"level\":" + jsonString(r.level) + ",\"lines\":" + lines + + ",\"truncated\":" + (truncated ? "true" : "false") + "}"; +} + } // namespace fgc diff --git a/src/core/Logger.cpp b/src/core/Logger.cpp index ca246f0..95ff592 100644 --- a/src/core/Logger.cpp +++ b/src/core/Logger.cpp @@ -17,6 +17,7 @@ std::atomic g_cats{0}; // every wire-trace category OFF by default std::mutex g_mutex; Logger::LogSink g_sink; // empty => default stdout/stderr writer (below) Logger::FileSink g_file_sink; // empty => no file logging +Logger::TapSink g_tap; // empty => nothing listening // Default writer: the stdout/stderr split this logger has always used. void writeDefault(LogLevel level, const std::string& line) { @@ -99,6 +100,11 @@ LogLevel Logger::effectiveLevel() { bool Logger::enabled(LogLevel level) { return level >= effectiveLevel(); } +void Logger::setTap(TapSink tap) { + std::lock_guard lock(g_mutex); + g_tap = std::move(tap); +} + void Logger::setFileSink(FileSink sink) { std::lock_guard lock(g_mutex); g_file_sink = std::move(sink); @@ -200,6 +206,7 @@ void LogStream::commit() { g_file_sink(level_, cat_, line); } if (is_cat || level_ >= g_level.load()) { + if (g_tap) g_tap(level_, line); if (g_sink) g_sink(level_, line); else writeDefault(level_, line); } diff --git a/src/core/RemoteCommands.cpp b/src/core/RemoteCommands.cpp new file mode 100644 index 0000000..16cf917 --- /dev/null +++ b/src/core/RemoteCommands.cpp @@ -0,0 +1,167 @@ +#include "fgc/RemoteCommands.h" + +#include "fgc/Logger.h" + +#include +#include +#include +#include +#include + +namespace fgc { + +namespace { + +CommandReply refusal(const RemoteCommand& rc, const std::string& why) { + CommandReply r; + r.id = rc.id; + r.cmd = rc.cmd; + r.level = "error"; + r.lines = {why}; + return r; +} + +bool containsAny(const std::string& line, const std::vector& needles) { + for (const auto& n : needles) + if (line.find(n) != std::string::npos) return true; + return false; +} + +} // namespace + +// What the tap collects. The tap runs on whichever thread logs, under the +// logger's lock, so it only records - under its own small lock, since poll() +// reads it on the control thread while the tap may still be adding. +struct RemoteCommandRunner::Capture { + std::mutex m; + std::thread::id tid = std::this_thread::get_id(); + bool handling = true; // the command is still running + std::optional await; + bool ended = false; // the awaited answer has arrived + std::vector lines; + LogLevel worst = LogLevel::Info; + + void take(LogLevel level, const std::string& text) { + std::lock_guard lock(m); + // While the command runs: everything it logs, on its own thread - the + // camera, the MQTT client and the sensors log from theirs meanwhile, and + // none of that is its answer. From any thread: the awaited answer. + const bool mine = handling && std::this_thread::get_id() == tid; + const bool answer = await && containsAny(text, await->collect); + const bool closing = await && containsAny(text, await->ends); + if (!mine && !answer && !closing) return; + // A multi-line entry (the firmware dump) becomes one reply line per line. + size_t from = 0; + while (from <= text.size()) { + const size_t nl = text.find('\n', from); + const std::string part = text.substr(from, nl == std::string::npos ? nl : nl - from); + if (!part.empty() || nl != std::string::npos) lines.push_back(part); + if (nl == std::string::npos) break; + from = nl + 1; + } + if (!lines.empty() && lines.back().empty()) lines.pop_back(); + if (level > worst) worst = level; + if (closing) ended = true; + } +}; + +struct RemoteCommandRunner::Pending { + RemoteCommand rc; + std::shared_ptr cap; + long long deadline_ms = 0; +}; + +std::optional RemoteCommandRunner::run(const RemoteCommand& rc, + const Handler& handle, + const Await* await, long long now_ms) { + const std::string tag = rc.id.empty() ? "" : " [" + rc.id + "]"; + + if (!rc.error.empty()) { + LOG_WARN << "remote command" << tag << " not run: " << rc.error; + return refusal(rc, "not run: " + rc.error); + } + if (rc.retained) { + LOG_WARN << "remote command" << tag << " ignored: it was a retained message ('" << rc.cmd + << "'), which the broker re-delivers on every reconnect"; + return refusal(rc, "not run: retained messages are never run - publish commands " + "without the retain flag"); + } + if (!rc.id.empty()) { + auto it = std::find_if(recent_.begin(), recent_.end(), + [&](const std::pair& p) { + return p.first == rc.id; + }); + if (it != recent_.end()) { + LOG_INFO << "remote command" << tag << " is a repeat; answered again, not run again"; + return it->second; + } + } + + LOG_INFO << "remote command" << tag << ": " << rc.cmd; + + // The tap is armed before the command runs, so an answer that comes back + // quickly is not missed; reading the buffer after setTap({}) is ordered by + // the logger's lock. + auto cap = std::make_shared(); + if (await) cap->await = *await; + Logger::setTap([cap](LogLevel level, const std::string& line) { cap->take(level, line); }); + try { + handle(rc.cmd); + } catch (...) { + Logger::setTap({}); + throw; + } + + bool wait = false; + { + std::lock_guard lock(cap->m); + cap->handling = false; + // A command that warned (refused, busy, bad arguments) sent nothing to + // wait for. + wait = await && !cap->ended && cap->worst < LogLevel::Warn; + } + if (wait) { + pending_ = std::make_shared(Pending{rc, cap, now_ms + await->timeout_ms}); + return std::nullopt; + } + return finish(rc, *cap); +} + +std::optional RemoteCommandRunner::poll(long long now_ms) { + if (!pending_) return std::nullopt; + Capture& cap = *pending_->cap; + { + std::lock_guard lock(cap.m); + if (!cap.ended) { + if (now_ms < pending_->deadline_ms) return std::nullopt; + cap.lines.push_back("no answer from the firmware within " + + std::to_string(cap.await->timeout_ms / 1000) + " s"); + if (cap.worst < LogLevel::Warn) cap.worst = LogLevel::Warn; + } + } + const std::shared_ptr done = std::move(pending_); + return finish(done->rc, *done->cap); +} + +CommandReply RemoteCommandRunner::finish(const RemoteCommand& rc, Capture& cap) { + Logger::setTap({}); + + CommandReply r; + r.id = rc.id; + r.cmd = rc.cmd; + { + std::lock_guard lock(cap.m); + r.lines = std::move(cap.lines); + r.level = cap.worst >= LogLevel::Error ? "error" + : cap.worst >= LogLevel::Warn ? "warn" + : "info"; + } + + if (!rc.id.empty()) { + recent_.emplace_back(rc.id, r); + while (recent_.size() > kRememberIds) recent_.pop_front(); + } + return r; +} + +} // namespace fgc diff --git a/src/mqtt/MqttControlChannel.cpp b/src/mqtt/MqttControlChannel.cpp index a34c5d7..43ef59a 100644 --- a/src/mqtt/MqttControlChannel.cpp +++ b/src/mqtt/MqttControlChannel.cpp @@ -22,6 +22,8 @@ MqttControlChannel::MqttControlChannel(const std::string& broker_uri, const std: topic_control_code_(towerTopic(tower, kTopicControlCode)), 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); @@ -63,6 +65,7 @@ void MqttControlChannel::connected(const std::string& /*cause*/) { history_.noteConnect(); client_.subscribe(topic_target_hdg_, kQos); client_.subscribe(topic_control_code_, kQos); + client_.subscribe(topic_command_, kQos); connected_ = true; } @@ -83,6 +86,19 @@ void MqttControlChannel::message_arrived(mqtt::const_message_ptr msg) { LOG_TRACE_CAT(LogCat::Mqtt) << "RX " << topic << ' ' << payload; history_.record(/*outbound=*/false, Delivery::Ok, topic, payload); std::lock_guard 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::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; + return; + } try { if (topic == topic_target_hdg_) { (void)std::stoi(payload); // validate it is an integer @@ -105,7 +121,21 @@ ControlCommand MqttControlChannel::poll() { return out; } -void MqttControlChannel::publish(const std::string& topic, const std::string& payload) { +std::vector MqttControlChannel::takeCommands() { + std::lock_guard lock(mutex_); + std::vector 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. @@ -116,7 +146,7 @@ void MqttControlChannel::publish(const std::string& topic, const std::string& pa try { auto msg = mqtt::make_message(topic, payload); msg->set_qos(kQos); - msg->set_retained(true); + msg->set_retained(retained); client_.publish(msg)->wait_for(kPublishTimeout); history_.record(/*outbound=*/true, Delivery::Ok, topic, payload); } catch (const mqtt::exception& e) { diff --git a/src/ui/TuiUi.cpp b/src/ui/TuiUi.cpp index e06563f..148e475 100644 --- a/src/ui/TuiUi.cpp +++ b/src/ui/TuiUi.cpp @@ -633,7 +633,7 @@ Element connPanel(const ConnView& v) { })); } -// Expanded comms view ('m'): broker state, the five topics with their last +// Expanded comms view ('m'): broker state, the six topics with their last // traffic, the counters, and a scrollable history of recent messages. // // The history is the point of the view. `dropped` in particular has no other @@ -660,8 +660,8 @@ Element connDetailPanel(const ConnView& v, ScrollState& st) { : "") | dim})); rows.push_back(row("uri", text(v.broker.empty() ? "\xE2\x80\x94" : v.broker))); rows.push_back(row("client id", text(v.tower))); - rows.push_back(row("settings", text("QoS 1 \xC2\xB7 retained \xC2\xB7 keep-alive 20 s " - "\xC2\xB7 clean session") | dim)); + rows.push_back(row("settings", text("QoS 1 \xC2\xB7 retained (not replies) \xC2\xB7 " + "keep-alive 20 s \xC2\xB7 clean session") | dim)); rows.push_back(row("link", text(std::to_string(v.stats.connects) + " connect(s), last loss " + formatTimeAgo(now, v.stats.last_loss_ms)) | dim)); @@ -685,8 +685,8 @@ Element connDetailPanel(const ConnView& v, ScrollState& st) { rows.push_back(text("TOPICS") | bold | color(Color::Green)); struct TopicRow { const char* dir; const char* leaf; }; static const TopicRow kTopics[] = { - {"sub", kTopicTargetHdg}, {"sub", kTopicControlCode}, {"pub", kTopicStatusCode}, - {"pub", kTopicCamEvent}, + {"sub", kTopicTargetHdg}, {"sub", kTopicControlCode}, {"sub", kTopicCommand}, + {"pub", kTopicStatusCode}, {"pub", kTopicCamEvent}, {"pub", kTopicCommandReply}, }; for (const auto& t : kTopics) { const ChannelMessage* last = nullptr; @@ -697,7 +697,9 @@ Element connDetailPanel(const ConnView& v, ScrollState& st) { text(std::string(" ") + t.dir + " ") | dim, text(t.leaf) | size(WIDTH, EQUAL, 14), text(when) | dim | size(WIDTH, EQUAL, 10), - text(last ? last->payload : "") | dim, + // A long payload (a CommandReply) is cut at the edge, never the + // columns before it. + text(last ? last->payload : "") | dim | xflex_shrink, })); } @@ -718,8 +720,9 @@ Element connDetailPanel(const ConnView& v, ScrollState& st) { rows.push_back(hbox({ text(formatClockMs(m.timestamp_ms) + " ") | dim, tag, - text(" " + topicLeaf(m.topic)) | size(WIDTH, EQUAL, 14), - text(" " + m.payload + note) | (bad ? color(Color::Red) : color(Color::Default)), + text(" " + topicLeaf(m.topic)) | size(WIDTH, EQUAL, 16), + text(" " + m.payload + note) | (bad ? color(Color::Red) : color(Color::Default)) | + xflex_shrink, })); } } diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index 658a843..63fbd1f 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -24,6 +24,7 @@ add_executable(fgc_tests test_config.cpp test_iniedit.cpp test_camerastall.cpp + test_remotecommands.cpp test_tmccurrent.cpp test_hardwarepresence.cpp test_hardwarechecks.cpp diff --git a/tests/test_remotecommands.cpp b/tests/test_remotecommands.cpp new file mode 100644 index 0000000..e707c87 --- /dev/null +++ b/tests/test_remotecommands.cpp @@ -0,0 +1,223 @@ +#include + +#include "fgc/ChannelTopics.h" +#include "fgc/Logger.h" +#include "fgc/RemoteCommands.h" +#include "fgc/mock/NullControlChannel.h" + +#include +#include + +using namespace fgc; + +// --- the Command payload --------------------------------------------------------- + +TEST_CASE("a JSON command carries its id and command line") { + auto rc = parseCommandPayload(R"({"id":"web-42","cmd":"gimbal move 45,10"})"); + CHECK(rc.error.empty()); + CHECK(rc.id == "web-42"); + CHECK(rc.cmd == "gimbal move 45,10"); +} + +TEST_CASE("a bare line is a command with no id - for mosquitto_pub from a shell") { + auto rc = parseCommandPayload(" gimbal status\n"); + CHECK(rc.error.empty()); + CHECK(rc.id.empty()); + CHECK(rc.cmd == "gimbal status"); +} + +TEST_CASE("JSON escapes are honoured, and unknown keys are ignored") { + auto rc = parseCommandPayload( + R"({ "user": "maint", "n": 3, "cmd": "gimbal raw \"STATUS\"", "id": "aé" })"); + CHECK(rc.error.empty()); + CHECK(rc.cmd == "gimbal raw \"STATUS\""); + CHECK(rc.id == "a\xC3\xA9"); +} + +TEST_CASE("unreadable or empty payloads are errors, never commands") { + CHECK_FALSE(parseCommandPayload("").error.empty()); + CHECK_FALSE(parseCommandPayload("{\"id\":\"x\"}").error.empty()); // no cmd + CHECK_FALSE(parseCommandPayload("{\"cmd\":42}").error.empty()); // not a string + CHECK_FALSE(parseCommandPayload("{\"cmd\":\"start\"").error.empty()); // unterminated + CHECK_FALSE(parseCommandPayload("start\nexit").error.empty()); // two commands +} + +// --- the CommandReply payload ----------------------------------------------------- + +TEST_CASE("the reply is valid JSON, whatever the log lines contain") { + CommandReply r; + r.id = "7"; r.cmd = "help"; r.level = "warn"; + r.lines = {"a \"quoted\" word", "back\\slash", "tab\there"}; + const std::string j = commandReplyJson(r); + CHECK(j == "{\"id\":\"7\",\"cmd\":\"help\",\"level\":\"warn\",\"lines\":" + "[\"a \\\"quoted\\\" word\",\"back\\\\slash\",\"tab\\there\"],\"truncated\":false}"); +} + +TEST_CASE("a long answer is cut, and says so") { + CommandReply r; + r.lines.assign(kReplyMaxLines + 5, "x"); + const std::string j = commandReplyJson(r); + CHECK(j.find("\"truncated\":true") != std::string::npos); +} + +// --- running it ----------------------------------------------------------------- + +TEST_CASE("a remote command runs through the handler and is answered with what it logged") { + RemoteCommandRunner run; + RemoteCommand rc; rc.id = "1"; rc.cmd = "help"; + std::string seen; + auto reply = run.run(rc, [&](const std::string& line) { + seen = line; + LOG_INFO << "first line of help"; + LOG_INFO << "second line of help"; + }); + CHECK(seen == "help"); + CHECK(reply->id == "1"); + CHECK(reply->level == "info"); + REQUIRE(reply->lines.size() == 2); + CHECK(reply->lines[1].find("second line of help") != std::string::npos); +} + +TEST_CASE("the level is the most severe line: a refusal reads as warn") { + RemoteCommandRunner run; + RemoteCommand rc; rc.id = "2"; rc.cmd = "gimbal move 10,0"; + auto reply = run.run(rc, [](const std::string&) { + LOG_WARN << "gimbal move refused while a scan is running"; + }); + CHECK(reply->level == "warn"); + RemoteCommand bad; bad.id = "3"; bad.cmd = "camera reset"; + CHECK(run.run(bad, [](const std::string&) { LOG_ERROR << "reset FAILED"; })->level == "error"); +} + +TEST_CASE("a retained or unreadable command is answered but never run") { + RemoteCommandRunner run; + int ran = 0; + RemoteCommand retained; retained.cmd = "start"; retained.retained = true; + auto r1 = run.run(retained, [&](const std::string&) { ++ran; }); + CHECK(ran == 0); + CHECK(r1->level == "error"); + CHECK(r1->lines.at(0).find("retained") != std::string::npos); + + RemoteCommand garbled; garbled.error = "unreadable JSON"; + auto r2 = run.run(garbled, [&](const std::string&) { ++ran; }); + CHECK(ran == 0); + CHECK(r2->level == "error"); +} + +TEST_CASE("a repeated id is answered again, not run again") { + RemoteCommandRunner run; + int ran = 0; + RemoteCommand rc; rc.id = "dup"; rc.cmd = "snapshot"; + auto h = [&](const std::string&) { ++ran; LOG_INFO << "snap " << ran; }; + auto first = run.run(rc, h); + auto second = run.run(rc, h); + CHECK(ran == 1); + CHECK(second->lines == first->lines); + // Without an id nothing can be recognised as a repeat: it runs each time. + RemoteCommand bare; bare.cmd = "snapshot"; + run.run(bare, h); + run.run(bare, h); + CHECK(ran == 3); +} + +TEST_CASE("the oldest remembered id ages out") { + RemoteCommandRunner run; + int ran = 0; + auto h = [&](const std::string&) { ++ran; }; + RemoteCommand first; first.id = "id-0"; first.cmd = "x"; + run.run(first, h); + for (size_t i = 1; i <= RemoteCommandRunner::kRememberIds; ++i) { + RemoteCommand rc; rc.id = "id-" + std::to_string(i); rc.cmd = "x"; + run.run(rc, h); + } + run.run(first, h); // forgotten by now: runs again + CHECK(ran == static_cast(RemoteCommandRunner::kRememberIds) + 2); +} + +TEST_CASE("lines logged by other threads meanwhile are not part of the answer") { + RemoteCommandRunner run; + RemoteCommand rc; rc.id = "t"; rc.cmd = "gimbal status"; + auto reply = run.run(rc, [](const std::string&) { + std::thread other([] { LOG_INFO << "camera: frame from another thread"; }); + other.join(); + LOG_INFO << "status from the command itself"; + }); + REQUIRE(reply->lines.size() == 1); + CHECK(reply->lines[0].find("status from the command itself") != std::string::npos); +} + +// --- an answer the firmware sends later --------------------------------------- + +namespace { +const Await kDump{{"firmware dump", "Motor controller:"}, + {"firmware dump:", "firmware dump still incomplete"}, 5000}; +} + +TEST_CASE("a dump's reply waits for the dump, logged later on another thread") { + RemoteCommandRunner run; + RemoteCommand rc; rc.id = "d"; rc.cmd = "gimbal dump"; + auto now = run.run(rc, [](const std::string&) { LOG_INFO << "requesting firmware dump..."; }, + &kDump, 1000); + CHECK_FALSE(now); + CHECK(run.waiting()); + CHECK_FALSE(run.poll(2000)); // not arrived yet + std::thread serial([] { + LOG_INFO << "camera: an unrelated line"; + LOG_WARN << "firmware dump incomplete (lossy read); re-requesting (7 attempt(s) left)"; + LOG_INFO << "firmware dump:\nDUMP BEGIN\nDUMP Y state=1\nDUMP END\n"; + }); + serial.join(); + LOG_INFO << "sweep: a control-thread line after the command returned"; + auto reply = run.poll(2500); + REQUIRE(reply); + CHECK_FALSE(run.waiting()); + CHECK(reply->level == "warn"); // the lossy read is part of the answer + REQUIRE(reply->lines.size() == 6); + CHECK(reply->lines[0].find("requesting firmware dump") != std::string::npos); + CHECK(reply->lines[1].find("re-requesting") != std::string::npos); + CHECK(reply->lines[2].find("firmware dump:") != std::string::npos); + CHECK(reply->lines[3] == "DUMP BEGIN"); + CHECK(reply->lines[5] == "DUMP END"); +} + +TEST_CASE("a dump that never comes is answered when the wait runs out") { + RemoteCommandRunner run; + RemoteCommand rc; rc.id = "d2"; rc.cmd = "gimbal dump"; + CHECK_FALSE(run.run(rc, [](const std::string&) {}, &kDump, 1000)); + CHECK_FALSE(run.poll(5999)); + auto reply = run.poll(6000); + REQUIRE(reply); + CHECK(reply->level == "warn"); + CHECK(reply->lines.back().find("no answer from the firmware") != std::string::npos); + // ...and remembered like any other: a repeat is answered, not re-sent. + int ran = 0; + auto again = run.run(rc, [&](const std::string&) { ++ran; }, &kDump, 7000); + REQUIRE(again); + CHECK(ran == 0); +} + +TEST_CASE("a refused command sent nothing, so its reply does not wait") { + RemoteCommandRunner run; + RemoteCommand rc; rc.id = "d3"; rc.cmd = "gimbal dump"; + auto reply = run.run(rc, [](const std::string&) { LOG_WARN << "ignored: a test is in progress"; }, + &kDump, 1000); + REQUIRE(reply); + CHECK_FALSE(run.waiting()); + CHECK(reply->level == "warn"); +} + +TEST_CASE("the null channel carries an injected command through to a recorded reply") { + NullControlChannel ch("dev"); + RemoteCommand rc; rc.id = "n1"; rc.cmd = "help"; + ch.inject(rc); + auto got = ch.takeCommands(); + REQUIRE(got.size() == 1); + CHECK(ch.takeCommands().empty()); + CommandReply r; r.id = "n1"; r.cmd = "help"; + ch.publishCommandReply(r); + REQUIRE(ch.replies().size() == 1); + const auto recent = ch.recentMessages(); + REQUIRE(recent.size() == 2); + CHECK(recent[0].topic == "GGS/FWT/dev/Command"); + CHECK(recent[1].topic == "GGS/FWT/dev/CommandReply"); +}