Take any console command over MQTT, and answer it
This commit is contained in:
parent
9dc9b90f6d
commit
4a69d65afb
GPG Key ID:
D4A6C394DF13F1D7
|
|
@ -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
|
||||
|
|
|
|||
10
README.md
10
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/<tower>/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 |
|
||||
|---|---|
|
||||
|
|
|
|||
|
|
@ -11,7 +11,7 @@ GGS/FWT/<tower_name>/...
|
|||
```
|
||||
|
||||
`<tower_name>` 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/<tower>/Command` | to the tower | `{"id":"<your id>","cmd":"<command line>"}`, or a bare command line |
|
||||
| `GGS/FWT/<tower>/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 [<id>]: <command line>`, 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/<tower>/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/<tower>/target_HDG (int heading)
|
||||
GGS/FWT/<tower>/ControlCode (0 = auto sweep, 1 = directed)
|
||||
GGS/FWT/<tower>/Command (any console command: JSON {id, cmd} or a bare line)
|
||||
|
||||
publish: GGS/FWT/<tower>/StatusCode (echoed control code)
|
||||
GGS/FWT/<tower>/CamEvent (JSON: fwt, cam, hdg×10, pit×10, time-ms)
|
||||
GGS/FWT/<tower>/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
|
||||
```
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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/<tower>/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/<tower>/CommandReply.
|
||||
struct CommandReply {
|
||||
std::string id;
|
||||
std::string cmd;
|
||||
std::string level = "info"; // most severe line: info / warn / error
|
||||
std::vector<std::string> 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<RemoteCommand> 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<ChannelMessage> recentMessages() const { return {}; }
|
||||
|
|
|
|||
|
|
@ -84,6 +84,19 @@ public:
|
|||
using FileSink = std::function<void(LogLevel level, LogCat cat, const std::string& line)>;
|
||||
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<void(LogLevel level, const std::string& line)>;
|
||||
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.
|
||||
//
|
||||
|
|
|
|||
|
|
@ -7,15 +7,16 @@
|
|||
#include <memory>
|
||||
#include <mutex>
|
||||
#include <string>
|
||||
#include <vector>
|
||||
|
||||
#include <mqtt/async_client.h>
|
||||
|
||||
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/<tower>/... (see docs/mqtt-api.md). Publishes are QoS 1,
|
||||
// retained.
|
||||
// target_HDG / ControlCode / Command topics 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.
|
||||
|
|
@ -36,6 +37,8 @@ public:
|
|||
void publishStatus(int code) override;
|
||||
void publishCamEvent(const CamEvent& event) override;
|
||||
ControlCommand poll() 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(); }
|
||||
|
|
@ -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<RemoteCommand> commands_; // received, not yet taken (under mutex_)
|
||||
};
|
||||
|
||||
} // namespace fgc
|
||||
|
|
|
|||
|
|
@ -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 <deque>
|
||||
#include <functional>
|
||||
#include <memory>
|
||||
#include <optional>
|
||||
#include <string>
|
||||
#include <utility>
|
||||
#include <vector>
|
||||
|
||||
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<std::string> collect;
|
||||
std::vector<std::string> ends;
|
||||
long long timeout_ms = 5000;
|
||||
};
|
||||
|
||||
class RemoteCommandRunner {
|
||||
public:
|
||||
using Handler = std::function<void(const std::string& line)>;
|
||||
|
||||
// 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<CommandReply> 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<CommandReply> 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> pending_;
|
||||
std::deque<std::pair<std::string, CommandReply>> recent_; // id -> reply, newest last
|
||||
};
|
||||
|
||||
} // namespace fgc
|
||||
|
|
@ -7,6 +7,7 @@
|
|||
|
||||
#include <string>
|
||||
#include <utility>
|
||||
#include <vector>
|
||||
|
||||
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<RemoteCommand> takeCommands() override {
|
||||
std::vector<RemoteCommand> out;
|
||||
out.swap(pending_);
|
||||
return out;
|
||||
}
|
||||
void publishCommandReply(const CommandReply& r) override {
|
||||
replies_.push_back(r);
|
||||
drop(kTopicCommandReply, commandReplyJson(r));
|
||||
}
|
||||
const std::vector<CommandReply>& replies() const { return replies_; }
|
||||
|
||||
std::vector<ChannelMessage> 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<RemoteCommand> pending_;
|
||||
std::vector<CommandReply> replies_;
|
||||
};
|
||||
|
||||
} // namespace fgc
|
||||
|
|
|
|||
|
|
@ -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 <csignal>
|
||||
#include <cstdio>
|
||||
#include <cstdlib>
|
||||
#include <deque>
|
||||
#include <ctime>
|
||||
#include <filesystem>
|
||||
#include <fstream>
|
||||
#include <memory>
|
||||
#include <mutex>
|
||||
#include <optional>
|
||||
#include <queue>
|
||||
#include <sstream>
|
||||
#include <string>
|
||||
|
|
@ -280,6 +283,8 @@ struct Application::Impl {
|
|||
bool heading_setup_active_ = false;
|
||||
|
||||
std::atomic<bool> running{true};
|
||||
RemoteCommandRunner remote_; // commands over MQTT (GGS/FWT/<tower>/Command)
|
||||
std::deque<RemoteCommand> 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::milliseconds>(
|
||||
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> 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<Await> 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<std::string> 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
|
||||
|
|
|
|||
|
|
@ -1,5 +1,8 @@
|
|||
#include "fgc/ChannelTopics.h"
|
||||
|
||||
#include <cctype>
|
||||
#include <cstdio>
|
||||
|
||||
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<char>(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<unsigned char>(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<unsigned>(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<char>(cp);
|
||||
else if (cp < 0x800) {
|
||||
out += static_cast<char>(0xC0 | (cp >> 6));
|
||||
out += static_cast<char>(0x80 | (cp & 0x3F));
|
||||
} else {
|
||||
out += static_cast<char>(0xE0 | (cp >> 12));
|
||||
out += static_cast<char>(0x80 | ((cp >> 6) & 0x3F));
|
||||
out += static_cast<char>(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
|
||||
|
|
|
|||
|
|
@ -17,6 +17,7 @@ std::atomic<unsigned> 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<std::mutex> lock(g_mutex);
|
||||
g_tap = std::move(tap);
|
||||
}
|
||||
|
||||
void Logger::setFileSink(FileSink sink) {
|
||||
std::lock_guard<std::mutex> 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);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,167 @@
|
|||
#include "fgc/RemoteCommands.h"
|
||||
|
||||
#include "fgc/Logger.h"
|
||||
|
||||
#include <algorithm>
|
||||
#include <memory>
|
||||
#include <mutex>
|
||||
#include <thread>
|
||||
#include <vector>
|
||||
|
||||
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<std::string>& 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> await;
|
||||
bool ended = false; // the awaited answer has arrived
|
||||
std::vector<std::string> lines;
|
||||
LogLevel worst = LogLevel::Info;
|
||||
|
||||
void take(LogLevel level, const std::string& text) {
|
||||
std::lock_guard<std::mutex> 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<Capture> cap;
|
||||
long long deadline_ms = 0;
|
||||
};
|
||||
|
||||
std::optional<CommandReply> 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<std::string, CommandReply>& 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<Capture>();
|
||||
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<std::mutex> 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>(Pending{rc, cap, now_ms + await->timeout_ms});
|
||||
return std::nullopt;
|
||||
}
|
||||
return finish(rc, *cap);
|
||||
}
|
||||
|
||||
std::optional<CommandReply> RemoteCommandRunner::poll(long long now_ms) {
|
||||
if (!pending_) return std::nullopt;
|
||||
Capture& cap = *pending_->cap;
|
||||
{
|
||||
std::lock_guard<std::mutex> 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<Pending> 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<std::mutex> 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
|
||||
|
|
@ -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<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;
|
||||
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<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.
|
||||
|
|
@ -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) {
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
}));
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -0,0 +1,223 @@
|
|||
#include <doctest/doctest.h>
|
||||
|
||||
#include "fgc/ChannelTopics.h"
|
||||
#include "fgc/Logger.h"
|
||||
#include "fgc/RemoteCommands.h"
|
||||
#include "fgc/mock/NullControlChannel.h"
|
||||
|
||||
#include <string>
|
||||
#include <thread>
|
||||
|
||||
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<int>(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");
|
||||
}
|
||||
Loading…
Reference in New Issue