217 lines
6.8 KiB
Go
217 lines
6.8 KiB
Go
// Package cli shares safe flag parsing and role-specific MQTT startup.
|
|
package cli
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"flag"
|
|
"fmt"
|
|
"io"
|
|
"log/slog"
|
|
"os"
|
|
"time"
|
|
|
|
"robot-market/test-mqtt-5/internal/backend"
|
|
"robot-market/test-mqtt-5/internal/config"
|
|
"robot-market/test-mqtt-5/internal/events"
|
|
"robot-market/test-mqtt-5/internal/mqtt"
|
|
"robot-market/test-mqtt-5/internal/simulator"
|
|
"robot-market/test-mqtt-5/internal/web"
|
|
)
|
|
|
|
func Run(ctx context.Context, args []string, out, errOut io.Writer, sim bool) int {
|
|
name := "test-mqtt5"
|
|
if sim {
|
|
name = "device-sim"
|
|
}
|
|
fs := flag.NewFlagSet(name, flag.ContinueOnError)
|
|
fs.SetOutput(io.Discard)
|
|
path := fs.String("config", "", "TOML file path (optional)")
|
|
device := fs.String("device-id", "", "Override topics.device_id")
|
|
level := fs.String("log-level", "", "Override logging.level")
|
|
credentials := fs.String("credentials", "", "Private deployment client JSON file (username/password only)")
|
|
broker := fs.String("broker", "", "Override mqtt.url")
|
|
ca := fs.String("ca", "", "Override tls.ca_file")
|
|
clientID := fs.String("client-id", "", "Override mqtt.client_id; use a distinct ID for each process")
|
|
plaintext := fs.Bool("allow-plaintext", false, "Explicitly permit mqtt:// test connections")
|
|
requestID := fs.String("request-id", "", "Command request ID; generated when omitted")
|
|
fs.Usage = func() {
|
|
if sim {
|
|
fmt.Fprintln(out, "Usage: device-sim [FLAGS] [check-config]\nRuns the harmless MQTT 5 simulator by default.")
|
|
} else {
|
|
fmt.Fprintln(out, "Usage: test-mqtt5 [FLAGS] check-config|observe|command ping|get_version\nFlags precede the subcommand.")
|
|
}
|
|
fs.SetOutput(out)
|
|
fs.PrintDefaults()
|
|
fs.SetOutput(io.Discard)
|
|
}
|
|
if err := fs.Parse(args); err != nil {
|
|
if errors.Is(err, flag.ErrHelp) {
|
|
return 0
|
|
}
|
|
fmt.Fprintln(errOut, "invalid CLI arguments; use --help (values omitted)")
|
|
return 2
|
|
}
|
|
mode := "simulate"
|
|
command := ""
|
|
if !sim {
|
|
if fs.NArg() < 1 {
|
|
fmt.Fprintln(errOut, "expected check-config, observe or command; use --help")
|
|
return 2
|
|
}
|
|
mode = fs.Arg(0)
|
|
if mode == "command" {
|
|
if fs.NArg() != 2 || (fs.Arg(1) != "ping" && fs.Arg(1) != "get_version") {
|
|
fmt.Fprintln(errOut, "command requires ping or get_version")
|
|
return 2
|
|
}
|
|
command = fs.Arg(1)
|
|
} else if fs.NArg() != 1 || (mode != "observe" && mode != "check-config") {
|
|
fmt.Fprintln(errOut, "expected check-config, observe or command; use --help")
|
|
return 2
|
|
}
|
|
} else if fs.NArg() != 0 {
|
|
if fs.NArg() != 1 || fs.Arg(0) != "check-config" {
|
|
fmt.Fprintln(errOut, "simulator accepts only check-config or no subcommand")
|
|
return 2
|
|
}
|
|
mode = "check-config"
|
|
}
|
|
if ctx.Err() != nil {
|
|
fmt.Fprintln(errOut, "configuration check interrupted")
|
|
return 1
|
|
}
|
|
overrides := map[string]string{}
|
|
defaults := config.Defaults()
|
|
// Role defaults remain below TOML, environment and explicit flags.
|
|
if sim {
|
|
defaults.MQTT.ClientID = "rm5-lab-simulator-01"
|
|
defaults.MQTT.Username = "sim-001"
|
|
}
|
|
if mode == "command" {
|
|
defaults.MQTT.ClientID = "rm5-lab-command-" + backend.ID()[:12]
|
|
}
|
|
if *credentials != "" {
|
|
data, err := os.ReadFile(*credentials)
|
|
if err != nil || len(data) > 1<<20 {
|
|
fmt.Fprintln(errOut, "cannot read private client credentials")
|
|
return 1
|
|
}
|
|
var record struct {
|
|
Username string `json:"username"`
|
|
Password string `json:"password"`
|
|
}
|
|
if json.Unmarshal(data, &record) != nil || record.Username == "" || record.Password == "" {
|
|
fmt.Fprintln(errOut, "invalid private client credentials")
|
|
return 1
|
|
}
|
|
overrides["mqtt.username"] = record.Username
|
|
overrides["mqtt.password"] = record.Password
|
|
overrides["mqtt.password_file"] = ""
|
|
}
|
|
fs.Visit(func(f *flag.Flag) {
|
|
switch f.Name {
|
|
case "device-id":
|
|
overrides["topics.device_id"] = *device
|
|
case "log-level":
|
|
overrides["logging.level"] = *level
|
|
case "broker":
|
|
overrides["mqtt.url"] = *broker
|
|
case "ca":
|
|
overrides["tls.ca_file"] = *ca
|
|
case "client-id":
|
|
overrides["mqtt.client_id"] = *clientID
|
|
case "allow-plaintext":
|
|
overrides["mqtt.allow_plaintext"] = fmt.Sprint(*plaintext)
|
|
}
|
|
})
|
|
cfg, err := config.LoadWithDefaults(*path, overrides, defaults)
|
|
if err != nil {
|
|
fmt.Fprintln(errOut, "configuration invalid:", err)
|
|
return 1
|
|
}
|
|
if cfg.Web.Enabled && mode != "observe" {
|
|
fmt.Fprintln(errOut, "web.enabled requires observe mode")
|
|
return 1
|
|
}
|
|
if (cfg.Topics.ObserveDevices != "" || cfg.Topics.ObserveAll) && mode != "observe" && mode != "check-config" {
|
|
fmt.Fprintln(errOut, "topic observation settings require observe mode")
|
|
return 1
|
|
}
|
|
if cfg.Receipts.Enabled && (mode != "observe" && mode != "check-config" || !cfg.Topics.ObserveAll) {
|
|
fmt.Fprintln(errOut, "receipt replies require raw observe mode")
|
|
return 1
|
|
}
|
|
var logLevel slog.Level
|
|
_ = logLevel.UnmarshalText([]byte(cfg.Logging.Level))
|
|
options := &slog.HandlerOptions{Level: logLevel}
|
|
var handler slog.Handler = slog.NewTextHandler(out, options)
|
|
if cfg.Logging.Format == "json" {
|
|
handler = slog.NewJSONHandler(out, options)
|
|
}
|
|
log := slog.New(handler)
|
|
if mode == "check-config" {
|
|
log.Log(ctx, logLevel, "configuration valid", "event", "config_valid", "network_checked", false, "credentials_checked", false)
|
|
return 0
|
|
}
|
|
if !cfg.MQTT.CleanStart {
|
|
fmt.Fprintln(errOut, "runtime requires clean_start=true on initial process connection; durable client recovery is not implemented")
|
|
return 1
|
|
}
|
|
store := events.NewStore(cfg.Limits.EventHistory, log)
|
|
client := mqtt.New(cfg, store)
|
|
if sim {
|
|
err = simulator.New(client).Run(ctx)
|
|
} else {
|
|
b := backend.New(client)
|
|
var viewer *web.Viewer
|
|
if cfg.Web.Enabled {
|
|
viewer = web.New(cfg.Web, store, cfg.Deadlines.Shutdown)
|
|
viewer.SetReadiness(client.Ready)
|
|
if err = viewer.Start(); err != nil {
|
|
fmt.Fprintln(errOut, err)
|
|
return 1
|
|
}
|
|
}
|
|
if err = client.Start(ctx, []string{"telemetry", "status", "commands", "acks"}, nil); err == nil {
|
|
if mode == "observe" {
|
|
err = b.Serve(ctx)
|
|
} else {
|
|
workerCtx, cancel := context.WithCancel(ctx)
|
|
done := make(chan error, 1)
|
|
go func() { done <- b.Serve(workerCtx) }()
|
|
id := *requestID
|
|
if id == "" {
|
|
id = backend.ID()
|
|
}
|
|
_, err = b.Command(ctx, backend.NewCommand(cfg.Topics.DeviceID, command, id, cfg.Deadlines.Command))
|
|
cancel()
|
|
workerErr := <-done
|
|
if err == nil {
|
|
err = workerErr
|
|
}
|
|
}
|
|
closeErr := client.Close()
|
|
if err == nil {
|
|
err = closeErr
|
|
}
|
|
}
|
|
if viewer != nil {
|
|
shutdown, stop := context.WithTimeout(context.Background(), cfg.Deadlines.Shutdown)
|
|
viewErr := viewer.Close(shutdown)
|
|
stop()
|
|
if err == nil {
|
|
err = viewErr
|
|
}
|
|
}
|
|
log.Info("backend counters", "event", "counters", "summary", b.Summary())
|
|
}
|
|
if err != nil {
|
|
fmt.Fprintln(errOut, "operation failed:", err)
|
|
return 1
|
|
}
|
|
log.Info("shutdown complete", "event", "shutdown_complete", "time", time.Now().UTC())
|
|
return 0
|
|
}
|