Greencell: the charger on your own broker, not a cloud it never had
The HabuDen has no cloud API to connect to. It is commissioned over Bluetooth in
the Greencell GC app, pointed at an MQTT broker the owner runs, and from then on
publishes there — so the connector is an MQTT client rather than an HTTP one,
and nothing in it reaches Greencell. The wire contract is Home Assistant's own
greencell component and the greencell_client 1.0.3 library beneath it, which is
the only published description of the topics: a BROADCAST on /greencell/broadcast
draws device announcements, and /greencell/evse/{sn}/ carries current in
milliamps, voltage, power under "momentary", the EVSE state, and the access level
chosen in the app.
That meant an MQTT client, and the server takes no dependencies, so internal/mqtt
is hand-rolled the way internal/ocpp's RFC 6455 layer is. It is scoped to what
this connector needs and says so: QoS 0 for everything we send, clean session,
no reconnect — a connection lives for one plugin call, which is exactly how the
manager builds and tears down an instance. Inbound PUBLISH is accepted at QoS 0,
1 and 2 with the acknowledgements each requires, because the QoS of a delivery is
the broker's choice and not ours; an unacknowledged QoS 1 is redelivered forever.
Read-only, and the reason is worth writing down rather than rediscovering. A
device in EXECUTE mode accepts START, STOP, SET_CURRENT and QUERY — but the topic
those go to appears in no source: not Greencell's integration page, not
greencell_client, and Home Assistant ships sensor-only for that same reason.
Publishing to a guessed topic would be a control feature whose failure mode is a
driver believing they stopped a charge. So the access level is reported, and
commandTopic is the seam: an operator who has watched their own broker and found
theirs sets it, and a state read then sends QUERY — the one command a READ-mode
device also honours — instead of waiting out the charger's publish cadence. The
day the topic is public, control is a payload away from the same field.
What the cascade resolves here is a broker, not an account, so host, port, TLS and
credentials resolve together from the highest layer that names a host: an
organization's address paired with a user's password would address a broker with
credentials never meant for it. The serial, the QUERY topic and the listen window
each describe the charger rather than the endpoint, so each resolves on its own.
Two reading rules the tests pin. A phase the device did not report stays nil
rather than zero, because zero amps on a charger is a real measurement — a JSON
null decoding to 0.0 was a live bug until a test caught it — and a partial read
returns with received/complete flags instead of failing, since a device that
publishes some topics on a slower cadence is still worth reading. And a reachable
broker with no charger on it is degraded, not down: the half we configure works
and the missing half is the device. The plugin's end-to-end tests run against an
in-process broker written to the raw wire format, so a bug in the client cannot
hide behind a matching bug in the fixture.
The apps get the third connector card. The panel needed nothing — it renders a
plugin's ConfigFields itself — but the per-user panes are still hand-written per
integration, which is now three near-copies and the argument for the generic
version already noted in the plugins README. The web form splits the broker from
the charger because the server resolves them differently. The phone card is a
declarative config against the shared widget, which gained a number field type, a
degraded state that reads amber rather than red, and a fix for a locked field
that was covering its own displayed value with dots. Twenty keys in three
languages across both apps; Greencell, HabuDen and the literal QUERY join the
proper nouns that stay in English.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
tajniak81
co-authored by
Claude Opus 5
Claude Opus 5
parent
5e6b8b4b1c
commit
340a81b0d6
@@ -167,10 +167,12 @@ go build -o bin/api-server.exe ./cmd/server
|
||||
Restart the server. The plugin appears in the panel's **Plugins** card,
|
||||
**disabled** by default.
|
||||
|
||||
> DriverVault ships two built-in connectors today — `toyota` (Toyota Connected /
|
||||
> MyToyota, read-only vehicle data) and `anker-solix` (Anker Solix V1 EV charger)
|
||||
> — both blank-imported from `builtin/builtin.go`. The **external** kind below
|
||||
> needs no rebuild and is the easier place to start a new one.
|
||||
> DriverVault ships three built-in connectors today — `toyota` (Toyota Connected /
|
||||
> MyToyota, read-only vehicle data), `anker-solix` (Anker Solix V1 EV charger) and
|
||||
> `greencell` (Greencell HabuDen EV charger, read over the owner's MQTT broker
|
||||
> rather than a cloud API) — all blank-imported from `builtin/builtin.go`. The
|
||||
> **external** kind below needs no rebuild and is the easier place to start a new
|
||||
> one.
|
||||
|
||||
---
|
||||
|
||||
@@ -320,10 +322,11 @@ The contract is shaped for these; see [`doc.go`](doc.go):
|
||||
exist (`Manager.InvokeWith` / `InvokeBatchWith`, driven by the integration routes
|
||||
and `internal/api/vehicleproviders.go`); what is missing is the generic route.
|
||||
- **Resilience** — retry/backoff, circuit breaker, per-plugin latency/error metrics.
|
||||
- **Per-tenant credentials _for arbitrary plugins_** — the two built-in connectors
|
||||
already have them, through the hand-written `/api/integrations/toyota` and
|
||||
`/api/integrations/anker-solix` routes and their **superadmin → org admin →
|
||||
user** config cascade. What is missing is the generic version: per-org/per-user
|
||||
- **Per-tenant credentials _for arbitrary plugins_** — the built-in connectors
|
||||
already have them, through the hand-written `/api/integrations/toyota`,
|
||||
`/api/integrations/anker-solix` and `/api/integrations/greencell` routes and
|
||||
their **superadmin → org admin → user** config cascade. Each is a near-copy of
|
||||
the last, which is the argument for the generic version: per-org/per-user
|
||||
config keyed off `ConfigFields`, so a newly registered plugin gets the same
|
||||
treatment without new endpoints.
|
||||
- **Audit logging** of plugin access. (Charger *control* commands are already
|
||||
|
||||
@@ -2,7 +2,8 @@
|
||||
// register them with the plugin registry. Import this package once (from the api
|
||||
// package) to make all built-in connectors available.
|
||||
//
|
||||
// Built-in connectors that ship today: toyota and ankersolix (imported below).
|
||||
// Built-in connectors that ship today: toyota, ankersolix and greencell
|
||||
// (imported below).
|
||||
// Add another under internal/plugins/builtin/<name>/ and blank-import it here, e.g.
|
||||
//
|
||||
// import _ "drivervault/apiserver/internal/plugins/builtin/acme"
|
||||
@@ -14,6 +15,8 @@ package builtin
|
||||
import (
|
||||
// anker-solix — Anker Solix V1 Smart EV Charger read-only cloud data.
|
||||
_ "drivervault/apiserver/internal/plugins/builtin/ankersolix"
|
||||
// greencell — Greencell HabuDen EV charger, read over the owner's MQTT broker.
|
||||
_ "drivervault/apiserver/internal/plugins/builtin/greencell"
|
||||
// toyota — Toyota Connected Europe (MyToyota) read-only vehicle data.
|
||||
_ "drivervault/apiserver/internal/plugins/builtin/toyota"
|
||||
)
|
||||
|
||||
@@ -0,0 +1,805 @@
|
||||
// Package greencell is a built-in connector for Greencell EV charging stations —
|
||||
// today the HabuDen wallbox (11/22 kW, 32 A), the one device Greencell itself
|
||||
// supports for third-party integration.
|
||||
//
|
||||
// There is no Greencell cloud API to talk to. The charger is commissioned over
|
||||
// Bluetooth in the Greencell GC app, where the owner points it at an MQTT broker
|
||||
// of their own and switches the "Home Assistant" integration on; from then on the
|
||||
// wallbox publishes its telemetry to that broker. This connector is therefore an
|
||||
// MQTT client, not an HTTP one: it joins the same broker and reads the topics the
|
||||
// device publishes. The wire contract is the one Home Assistant's own greencell
|
||||
// integration speaks (homeassistant/components/greencell, and the greencell_client
|
||||
// 1.0.3 library it builds on), which is the only published description of it:
|
||||
//
|
||||
// publish /greencell/broadcast {"name":"BROADCAST"} — ask devices to announce
|
||||
// subscribe /greencell/broadcast/device {"id":"<serial>", …} — a device announcing itself
|
||||
// subscribe /greencell/evse/<sn>/current {"l1":…,"l2":…,"l3":…} milliamps
|
||||
// subscribe /greencell/evse/<sn>/voltage {"l1":…,"l2":…,"l3":…} volts
|
||||
// subscribe /greencell/evse/<sn>/power {"momentary":…} watts
|
||||
// subscribe /greencell/evse/<sn>/status {"state":"CHARGING"}
|
||||
// subscribe /greencell/evse/<sn>/device_state {"level":"EXECUTE"} access level
|
||||
//
|
||||
// Scope & limitations:
|
||||
// - Read-only. The GC app can put a device in EXECUTE mode, in which it accepts
|
||||
// START / STOP / SET_CURRENT / QUERY commands — but the topic those commands
|
||||
// are published on is documented nowhere: not in Greencell's integration page,
|
||||
// not in greencell_client, and Home Assistant's own integration ships without
|
||||
// control for exactly that reason. The access level is reported (CanExecute)
|
||||
// so the UI can say what the device would allow; acting on it needs that
|
||||
// topic, which is the one missing piece. An operator who has found theirs can
|
||||
// set commandTopic, and a state read will then send QUERY (the one command a
|
||||
// READ-mode device also honours) to prompt an immediate publish.
|
||||
// - Local, not cloud. The broker is the owner's; nothing here reaches Greencell.
|
||||
// The charger and this server must both be able to reach it.
|
||||
// - Pull-shaped over a push protocol. The plugin contract builds an instance per
|
||||
// call, so each Invoke opens a short-lived session, asks for a broadcast, reads
|
||||
// what arrives within a bounded window, and disconnects. It does not hold a
|
||||
// subscription open between calls, so a reading is as fresh as the device's own
|
||||
// publish cadence within that window.
|
||||
// - HabuDen. Serials matching the HabuDen pattern are named as such; any other
|
||||
// Greencell device that speaks these topics is still read, just generically.
|
||||
package greencell
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net"
|
||||
"regexp"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"drivervault/apiserver/internal/mqtt"
|
||||
"drivervault/apiserver/internal/plugins"
|
||||
)
|
||||
|
||||
// Topics. The leading slash is part of the contract — Greencell publishes on
|
||||
// absolute topics, not the relative ones MQTT conventions would suggest.
|
||||
const (
|
||||
broadcastTopic = "/greencell/broadcast"
|
||||
discoveryTopic = "/greencell/broadcast/device"
|
||||
evsePrefix = "/greencell/evse/"
|
||||
)
|
||||
|
||||
// Per-device topic suffixes, which double as the keys of the raw/received maps.
|
||||
const (
|
||||
topicCurrent = "current"
|
||||
topicVoltage = "voltage"
|
||||
topicPower = "power"
|
||||
topicStatus = "status"
|
||||
topicDeviceState = "device_state"
|
||||
)
|
||||
|
||||
// telemetryTopics are the per-device suffixes a state read subscribes to.
|
||||
var telemetryTopics = []string{topicCurrent, topicVoltage, topicPower, topicStatus, topicDeviceState}
|
||||
|
||||
// Device naming, mirroring the upstream integration.
|
||||
const (
|
||||
nameHabuDen = "Habu Den"
|
||||
nameGeneric = "Greencell Device"
|
||||
)
|
||||
|
||||
// habuDenSerial matches a HabuDen serial (greencell_client GreencellUtils).
|
||||
var habuDenSerial = regexp.MustCompile(`^EVGC021[A-Z][0-9]{8}ZM[0-9]{4}$`)
|
||||
|
||||
// EVSE states the device reports on the status topic, lowercased. Anything the
|
||||
// device sends that is not in this set is reported as stateUnknown rather than
|
||||
// passed through, so a consumer can switch on a closed set.
|
||||
const (
|
||||
stateIdle = "idle"
|
||||
stateConnected = "connected"
|
||||
stateWaitingForCar = "waiting_for_car"
|
||||
stateCharging = "charging"
|
||||
stateFinished = "finished"
|
||||
stateErrorCar = "error_car"
|
||||
stateErrorEVSE = "error_evse"
|
||||
stateUnavailable = "unavailable"
|
||||
stateUnknown = "unknown"
|
||||
)
|
||||
|
||||
var evseStates = map[string]bool{
|
||||
stateIdle: true, stateConnected: true, stateWaitingForCar: true,
|
||||
stateCharging: true, stateFinished: true, stateErrorCar: true,
|
||||
stateErrorEVSE: true, stateUnavailable: true,
|
||||
}
|
||||
|
||||
// Access levels the device reports on device_state — the mode chosen in the GC
|
||||
// app. OFFLINE is the deprecated spelling of UNAVAILABLE and is folded into it,
|
||||
// as greencell_client does.
|
||||
const (
|
||||
accessDisabled = "disabled"
|
||||
accessRead = "read"
|
||||
accessExecute = "execute"
|
||||
accessUnavailable = "unavailable"
|
||||
accessUnknown = "unknown"
|
||||
)
|
||||
|
||||
// Broker defaults. 1883 is plain MQTT, 8883 the TLS port.
|
||||
const (
|
||||
defaultPort = "1883"
|
||||
defaultTLSPort = "8883"
|
||||
)
|
||||
|
||||
// Collection window bounds. A device answers a broadcast within 30 s per
|
||||
// Greencell's own troubleshooting note, but that is far too long to hold an HTTP
|
||||
// request open, so the default is shorter and the operator can raise it.
|
||||
const (
|
||||
defaultTimeout = 12 * time.Second
|
||||
minTimeout = 1 * time.Second
|
||||
maxTimeout = 60 * time.Second
|
||||
// discoveryGrace is how much longer discovery listens after the first device
|
||||
// replies, to catch the rest of a multi-charger site. Upstream uses the same
|
||||
// half second.
|
||||
discoveryGrace = 500 * time.Millisecond
|
||||
)
|
||||
|
||||
func init() {
|
||||
plugins.Register("greencell", func() plugins.Plugin { return &Plugin{} })
|
||||
}
|
||||
|
||||
// snPlaceholder is substituted with the charger serial in a configured command
|
||||
// topic, so one setting can serve every charger on a broker.
|
||||
const snPlaceholder = "{sn}"
|
||||
|
||||
// queryCommand is the payload that asks a device to publish its state at once.
|
||||
// Greencell documents QUERY as honoured in both READ and EXECUTE mode; what it
|
||||
// does not document is the topic to send it on, hence commandTopic being an
|
||||
// operator-supplied opt-in.
|
||||
const queryCommand = "QUERY"
|
||||
|
||||
// Plugin is the Greencell EVSE connector.
|
||||
type Plugin struct {
|
||||
mu sync.Mutex // guards the config below; Invoke may run concurrently
|
||||
address string
|
||||
useTLS bool
|
||||
username string
|
||||
password string
|
||||
serial string
|
||||
commandTopic string
|
||||
timeout time.Duration
|
||||
}
|
||||
|
||||
// Descriptor returns the plugin's static metadata for the admin panel.
|
||||
func (p *Plugin) Descriptor() plugins.Descriptor {
|
||||
return plugins.Descriptor{
|
||||
Name: "greencell",
|
||||
Provider: "Greencell (HabuDen EV charger)",
|
||||
Version: "1.0.0",
|
||||
Kind: plugins.KindBuiltin,
|
||||
Category: plugins.CategoryAPIsExternal,
|
||||
AuthType: plugins.AuthBasic,
|
||||
Capabilities: []plugins.Capability{
|
||||
{ID: "chargers", Method: "SUB", Endpoint: discoveryTopic,
|
||||
Description: "Discover Greencell chargers on the broker by publishing a broadcast and collecting the announcements."},
|
||||
{ID: "charger-state", Method: "SUB", Endpoint: evsePrefix + "{sn}/#",
|
||||
Description: "Normalized live state of one charger: EVSE status, access level, power, per-phase current and voltage (needs sn, or the configured serial)."},
|
||||
},
|
||||
ConfigFields: []plugins.ConfigField{
|
||||
// As with the Toyota and Anker connectors, nothing is Required at the
|
||||
// global layer: an operator may configure a shared broker here or leave
|
||||
// it to the per-user cascade. Missing values surface as a clear error
|
||||
// when a call is actually made.
|
||||
{Key: "host", Label: "MQTT broker host", Type: "text",
|
||||
Help: "Hostname or IP of the MQTT broker the charger was pointed at in the Greencell GC app (for example 10.2.1.10)."},
|
||||
{Key: "port", Label: "MQTT broker port", Type: "number", Default: defaultPort,
|
||||
Help: "Broker TCP port. Defaults to 1883, or 8883 when TLS is on."},
|
||||
{Key: "tls", Label: "Use TLS", Type: "select", Default: "off",
|
||||
Help: "Connect to the broker over TLS. Must match how the broker is configured.",
|
||||
Options: []plugins.SelectOption{
|
||||
{Value: "off", Label: "Off (plain MQTT)"},
|
||||
{Value: "on", Label: "On (MQTTS)"},
|
||||
}},
|
||||
{Key: "username", Label: "MQTT username", Type: "text",
|
||||
Help: "Broker username, if the broker requires authentication. Leave blank for an open broker."},
|
||||
{Key: "password", Label: "MQTT password", Type: "password", Secret: true,
|
||||
Help: "Broker password for the username above."},
|
||||
{Key: "serial", Label: "Charger serial", Type: "text",
|
||||
Help: "Serial of the charger, e.g. EVGC021B22752405ZM0018. Optional — leave blank and discovery will find whatever is on the broker."},
|
||||
{Key: "commandTopic", Label: "QUERY command topic", Type: "text",
|
||||
Help: "Optional. Greencell documents a QUERY command that makes the charger publish its state immediately, but not the topic to send it on — no published source names it. If you find yours (watch your broker while the GC app talks to the charger), put it here and reads stop waiting for the device's own cadence. Use " + snPlaceholder + " for the serial, e.g. /greencell/evse/" + snPlaceholder + "/command. Leave blank to listen only."},
|
||||
{Key: "timeout", Label: "Listen window (seconds)", Type: "number", Default: "12",
|
||||
Help: "How long to wait for the charger to publish before answering. Greencell allows a device up to 30 s to respond to a broadcast; raise this if discovery comes back empty."},
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
// Init applies resolved config. It performs no network I/O — every call opens its
|
||||
// own short-lived broker session.
|
||||
func (p *Plugin) Init(_ context.Context, config map[string]string) error {
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
|
||||
p.useTLS = truthy(config["tls"])
|
||||
host := strings.TrimSpace(config["host"])
|
||||
port := strings.TrimSpace(config["port"])
|
||||
if port == "" {
|
||||
port = defaultPort
|
||||
if p.useTLS {
|
||||
port = defaultTLSPort
|
||||
}
|
||||
}
|
||||
p.address = ""
|
||||
if host != "" {
|
||||
p.address = net.JoinHostPort(host, port)
|
||||
}
|
||||
p.username = strings.TrimSpace(config["username"])
|
||||
p.password = config["password"]
|
||||
p.serial = strings.TrimSpace(config["serial"])
|
||||
p.commandTopic = strings.TrimSpace(config["commandTopic"])
|
||||
p.timeout = parseTimeout(config["timeout"])
|
||||
return nil
|
||||
}
|
||||
|
||||
// HealthCheck connects to the broker and asks whatever is listening to announce
|
||||
// itself. A reachable broker with no charger on it is degraded rather than down:
|
||||
// the half we configure works, and the missing half is the device.
|
||||
func (p *Plugin) HealthCheck(ctx context.Context) plugins.Health {
|
||||
start := time.Now()
|
||||
found, err := p.discover(ctx)
|
||||
lat := time.Since(start).Milliseconds()
|
||||
if err != nil {
|
||||
return plugins.Health{Status: plugins.StatusDown, LatencyMs: lat, Detail: shorten(err.Error())}
|
||||
}
|
||||
if len(found) == 0 {
|
||||
return plugins.Health{Status: plugins.StatusDegraded, LatencyMs: lat,
|
||||
Detail: "broker reachable, but no Greencell device answered the discovery broadcast"}
|
||||
}
|
||||
return plugins.Health{Status: plugins.StatusOK, LatencyMs: lat,
|
||||
Detail: fmt.Sprintf("broker reachable; %d Greencell device(s) answered", len(found))}
|
||||
}
|
||||
|
||||
// invokeParams is what an action may be given. Per-charger actions take "sn"; it
|
||||
// falls back to the configured serial when omitted.
|
||||
type invokeParams struct {
|
||||
SN string `json:"sn"`
|
||||
}
|
||||
|
||||
// Invoke runs a named read-only capability.
|
||||
func (p *Plugin) Invoke(ctx context.Context, action string, params json.RawMessage) (json.RawMessage, error) {
|
||||
var pp invokeParams
|
||||
if len(params) > 0 {
|
||||
if err := json.Unmarshal(params, &pp); err != nil {
|
||||
return nil, fmt.Errorf("greencell: invalid params: %w", err)
|
||||
}
|
||||
}
|
||||
sn := strings.TrimSpace(pp.SN)
|
||||
if sn == "" {
|
||||
sn = p.configuredSerial()
|
||||
}
|
||||
|
||||
switch action {
|
||||
case "chargers":
|
||||
found, err := p.discover(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return json.Marshal(map[string]any{"chargers": found})
|
||||
case "charger-state":
|
||||
if sn == "" {
|
||||
return nil, errors.New("greencell: action \"charger-state\" requires an sn (charger serial), or a serial in the plugin config")
|
||||
}
|
||||
st, err := p.chargerState(ctx, sn)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return json.Marshal(st)
|
||||
default:
|
||||
return nil, fmt.Errorf("greencell: unknown action %q", action)
|
||||
}
|
||||
}
|
||||
|
||||
// Shutdown has nothing to release: sessions do not outlive a call.
|
||||
func (p *Plugin) Shutdown(context.Context) error { return nil }
|
||||
|
||||
// ---- broker session ----------------------------------------------------------
|
||||
|
||||
// snapshot copies the config under the lock so a call is not affected by a
|
||||
// concurrent Init.
|
||||
func (p *Plugin) snapshot() (opts mqtt.Options, timeout time.Duration, err error) {
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
if p.address == "" {
|
||||
return mqtt.Options{}, 0, errors.New("greencell: no MQTT broker configured — set the broker host the charger publishes to")
|
||||
}
|
||||
return mqtt.Options{
|
||||
Address: p.address,
|
||||
TLS: p.useTLS,
|
||||
Username: p.username,
|
||||
Password: p.password,
|
||||
// One keep-alive period comfortably outlives a listen window, so the
|
||||
// session never has to ping mid-collection.
|
||||
Keepalive: 60 * time.Second,
|
||||
ConnectTimeout: 10 * time.Second,
|
||||
}, p.timeout, nil
|
||||
}
|
||||
|
||||
func (p *Plugin) configuredSerial() string {
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
return p.serial
|
||||
}
|
||||
|
||||
// commandTopicFor resolves the configured QUERY topic for one charger, or ""
|
||||
// when the operator has not supplied one.
|
||||
func (p *Plugin) commandTopicFor(sn string) string {
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
if p.commandTopic == "" {
|
||||
return ""
|
||||
}
|
||||
return strings.ReplaceAll(p.commandTopic, snPlaceholder, sn)
|
||||
}
|
||||
|
||||
// session opens a broker connection, subscribes to topics, and publishes the
|
||||
// discovery broadcast that prompts devices to speak up. The caller drains
|
||||
// client.Messages until it has what it came for or its window runs out, then
|
||||
// closes the client.
|
||||
func (p *Plugin) session(ctx context.Context, topics []string) (*mqtt.Client, time.Duration, error) {
|
||||
opts, timeout, err := p.snapshot()
|
||||
if err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
client, err := mqtt.Connect(ctx, opts)
|
||||
if err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
if err := client.Subscribe(ctx, topics...); err != nil {
|
||||
_ = client.Close()
|
||||
return nil, 0, fmt.Errorf("greencell: subscribing on the broker: %w", err)
|
||||
}
|
||||
// The broadcast is what makes an otherwise-quiet device announce itself; the
|
||||
// telemetry topics then follow on the device's own cadence.
|
||||
if err := client.Publish(ctx, broadcastTopic, []byte(`{"name":"BROADCAST"}`)); err != nil {
|
||||
_ = client.Close()
|
||||
return nil, 0, fmt.Errorf("greencell: publishing the discovery broadcast: %w", err)
|
||||
}
|
||||
return client, timeout, nil
|
||||
}
|
||||
|
||||
// ---- discovery ---------------------------------------------------------------
|
||||
|
||||
// Device is one charger that answered the discovery broadcast.
|
||||
type Device struct {
|
||||
SN string `json:"sn"`
|
||||
Name string `json:"name"`
|
||||
Model string `json:"model"`
|
||||
// Announcement is the device's own broadcast payload, passed through
|
||||
// unchanged: Greencell may carry firmware or capability fields there that
|
||||
// this connector has no schema for.
|
||||
Announcement json.RawMessage `json:"announcement,omitempty"`
|
||||
}
|
||||
|
||||
// discover collects the devices that answer a broadcast. It listens for the full
|
||||
// window, cut short by discoveryGrace once at least one device has replied.
|
||||
func (p *Plugin) discover(ctx context.Context) ([]Device, error) {
|
||||
client, timeout, err := p.session(ctx, []string{discoveryTopic})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer client.Close()
|
||||
|
||||
found := map[string]Device{}
|
||||
var order []string
|
||||
|
||||
deadline := time.NewTimer(timeout)
|
||||
defer deadline.Stop()
|
||||
var grace *time.Timer
|
||||
graceC := func() <-chan time.Time {
|
||||
if grace == nil {
|
||||
return nil
|
||||
}
|
||||
return grace.C
|
||||
}
|
||||
|
||||
for {
|
||||
select {
|
||||
case msg, ok := <-client.Messages():
|
||||
if !ok {
|
||||
if err := client.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return devicesInOrder(found, order), nil
|
||||
}
|
||||
if msg.Topic != discoveryTopic {
|
||||
continue
|
||||
}
|
||||
dev, ok := parseAnnouncement(msg.Payload)
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
if _, seen := found[dev.SN]; !seen {
|
||||
order = append(order, dev.SN)
|
||||
}
|
||||
found[dev.SN] = dev
|
||||
if grace == nil {
|
||||
grace = time.NewTimer(discoveryGrace)
|
||||
defer grace.Stop()
|
||||
}
|
||||
case <-graceC():
|
||||
return devicesInOrder(found, order), nil
|
||||
case <-deadline.C:
|
||||
return devicesInOrder(found, order), nil
|
||||
case <-ctx.Done():
|
||||
return nil, ctx.Err()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// devicesInOrder returns the collected devices in the order they answered.
|
||||
func devicesInOrder(found map[string]Device, order []string) []Device {
|
||||
out := make([]Device, 0, len(order))
|
||||
for _, sn := range order {
|
||||
out = append(out, found[sn])
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// parseAnnouncement reads a device announcement. A payload without a usable "id"
|
||||
// is not a device and is ignored.
|
||||
func parseAnnouncement(payload []byte) (Device, bool) {
|
||||
var doc struct {
|
||||
ID string `json:"id"`
|
||||
}
|
||||
if err := json.Unmarshal(payload, &doc); err != nil {
|
||||
return Device{}, false
|
||||
}
|
||||
sn := strings.TrimSpace(doc.ID)
|
||||
if sn == "" {
|
||||
return Device{}, false
|
||||
}
|
||||
model := deviceModel(sn)
|
||||
return Device{
|
||||
SN: sn,
|
||||
Name: model + " " + sn,
|
||||
Model: model,
|
||||
Announcement: json.RawMessage(append([]byte(nil), payload...)),
|
||||
}, true
|
||||
}
|
||||
|
||||
// deviceModel names a device from its serial.
|
||||
func deviceModel(sn string) string {
|
||||
if habuDenSerial.MatchString(sn) {
|
||||
return nameHabuDen
|
||||
}
|
||||
return nameGeneric
|
||||
}
|
||||
|
||||
// ---- live state --------------------------------------------------------------
|
||||
|
||||
// Phases holds a per-phase measurement. A phase the device did not report stays
|
||||
// nil rather than reading as zero, which on a charger would be a real value.
|
||||
type Phases struct {
|
||||
L1 *float64 `json:"l1"`
|
||||
L2 *float64 `json:"l2"`
|
||||
L3 *float64 `json:"l3"`
|
||||
}
|
||||
|
||||
// State is one charger's normalized live state.
|
||||
type State struct {
|
||||
SN string `json:"sn"`
|
||||
Name string `json:"name"`
|
||||
Model string `json:"model"`
|
||||
|
||||
// Status is the EVSE state, lowercased and constrained to the known set.
|
||||
Status string `json:"status"`
|
||||
Charging bool `json:"charging"`
|
||||
// Plugged is true in every state that implies a cable in the socket.
|
||||
Plugged bool `json:"plugged"`
|
||||
// Fault is true for the two error states.
|
||||
Fault bool `json:"fault"`
|
||||
|
||||
// AccessLevel is the integration mode set in the Greencell GC app, and
|
||||
// CanExecute is whether that mode would accept commands. Control is not
|
||||
// implemented (see the package comment); this reports what the device allows.
|
||||
AccessLevel string `json:"accessLevel"`
|
||||
CanExecute bool `json:"canExecute"`
|
||||
Available bool `json:"available"`
|
||||
|
||||
PowerW *float64 `json:"powerW"`
|
||||
CurrentA Phases `json:"currentA"`
|
||||
VoltageV Phases `json:"voltageV"`
|
||||
// LivePhases counts the phases currently drawing a meaningful current, which
|
||||
// is what separates a 1-phase from a 3-phase charge on a 22 kW wallbox.
|
||||
LivePhases int `json:"livePhases"`
|
||||
|
||||
// Received says which topics were heard inside the listen window; a false
|
||||
// entry means that field is unset, not zero.
|
||||
Received map[string]bool `json:"received"`
|
||||
// Raw is each topic's last payload, verbatim, for fields this connector has
|
||||
// no schema for.
|
||||
Raw map[string]json.RawMessage `json:"raw,omitempty"`
|
||||
ObservedAt time.Time `json:"observedAt"`
|
||||
// Complete is true when every telemetry topic reported inside the window.
|
||||
Complete bool `json:"complete"`
|
||||
}
|
||||
|
||||
// chargerState listens for one charger's telemetry and normalizes it. It returns
|
||||
// as soon as every topic has been heard, or at the end of the window with
|
||||
// whatever arrived — a device that publishes some topics on a slower cadence
|
||||
// still yields a useful partial reading, flagged by Received/Complete. Total
|
||||
// silence is an error: that means the charger is not on this broker.
|
||||
func (p *Plugin) chargerState(ctx context.Context, sn string) (State, error) {
|
||||
topics := make([]string, 0, len(telemetryTopics)+1)
|
||||
topics = append(topics, discoveryTopic)
|
||||
for _, suffix := range telemetryTopics {
|
||||
topics = append(topics, evsePrefix+sn+"/"+suffix)
|
||||
}
|
||||
|
||||
client, timeout, err := p.session(ctx, topics)
|
||||
if err != nil {
|
||||
return State{}, err
|
||||
}
|
||||
defer client.Close()
|
||||
|
||||
// When the operator has told us where the charger listens, ask it to publish
|
||||
// now rather than waiting out its own cadence. A device in DISABLE or READ
|
||||
// mode ignores every other command but this one, so it is safe to send
|
||||
// whatever mode the charger is in. Failing to publish is not fatal: the
|
||||
// listen-only path still works.
|
||||
if topic := p.commandTopicFor(sn); topic != "" {
|
||||
_ = client.Publish(ctx, topic, []byte(queryCommand))
|
||||
}
|
||||
|
||||
raw := map[string]json.RawMessage{}
|
||||
deadline := time.NewTimer(timeout)
|
||||
defer deadline.Stop()
|
||||
|
||||
collect:
|
||||
for len(raw) < len(telemetryTopics) {
|
||||
select {
|
||||
case msg, ok := <-client.Messages():
|
||||
if !ok {
|
||||
if err := client.Err(); err != nil {
|
||||
return State{}, err
|
||||
}
|
||||
break collect
|
||||
}
|
||||
if suffix, match := topicSuffix(msg.Topic, sn); match && json.Valid(msg.Payload) {
|
||||
raw[suffix] = json.RawMessage(append([]byte(nil), msg.Payload...))
|
||||
}
|
||||
case <-deadline.C:
|
||||
break collect
|
||||
case <-ctx.Done():
|
||||
return State{}, ctx.Err()
|
||||
}
|
||||
}
|
||||
|
||||
if len(raw) == 0 {
|
||||
return State{}, fmt.Errorf("greencell: no data from charger %s within %s — check that it is powered, on this broker, and not set to DISABLE in the GC app", sn, timeout)
|
||||
}
|
||||
return buildState(sn, raw), nil
|
||||
}
|
||||
|
||||
// topicSuffix maps a received topic back to its per-device suffix.
|
||||
func topicSuffix(topic, sn string) (string, bool) {
|
||||
prefix := evsePrefix + sn + "/"
|
||||
if !strings.HasPrefix(topic, prefix) {
|
||||
return "", false
|
||||
}
|
||||
suffix := strings.TrimPrefix(topic, prefix)
|
||||
for _, known := range telemetryTopics {
|
||||
if suffix == known {
|
||||
return suffix, true
|
||||
}
|
||||
}
|
||||
return "", false
|
||||
}
|
||||
|
||||
// buildState turns the collected payloads into a normalized reading.
|
||||
func buildState(sn string, raw map[string]json.RawMessage) State {
|
||||
model := deviceModel(sn)
|
||||
st := State{
|
||||
SN: sn,
|
||||
Name: model + " " + sn,
|
||||
Model: model,
|
||||
Status: stateUnknown,
|
||||
AccessLevel: accessUnknown,
|
||||
Received: map[string]bool{},
|
||||
Raw: raw,
|
||||
ObservedAt: time.Now().UTC(),
|
||||
}
|
||||
for _, suffix := range telemetryTopics {
|
||||
st.Received[suffix] = raw[suffix] != nil
|
||||
}
|
||||
st.Complete = len(raw) == len(telemetryTopics)
|
||||
|
||||
// Current arrives in milliamps; every other consumer wants amperes.
|
||||
st.CurrentA = scalePhases(parsePhases(raw[topicCurrent]), 1.0/1000.0)
|
||||
st.VoltageV = parsePhases(raw[topicVoltage])
|
||||
st.PowerW = numberField(raw[topicPower], "momentary")
|
||||
|
||||
if s, ok := stringField(raw[topicStatus], "state"); ok {
|
||||
st.Status = normalizeState(s)
|
||||
}
|
||||
// The device also signals unavailability through the status topic itself,
|
||||
// which upstream detects by substring because the payload is not a state
|
||||
// document in that case.
|
||||
if isUnavailable(raw[topicStatus]) {
|
||||
st.Status = stateUnavailable
|
||||
}
|
||||
|
||||
if s, ok := stringField(raw[topicDeviceState], "level"); ok {
|
||||
st.AccessLevel = normalizeAccess(s)
|
||||
}
|
||||
st.CanExecute = st.AccessLevel == accessExecute
|
||||
st.Available = st.AccessLevel != accessDisabled && st.AccessLevel != accessUnavailable &&
|
||||
st.Status != stateUnavailable
|
||||
|
||||
st.Charging = st.Status == stateCharging
|
||||
switch st.Status {
|
||||
case stateConnected, stateWaitingForCar, stateCharging, stateFinished, stateErrorCar:
|
||||
st.Plugged = true
|
||||
}
|
||||
st.Fault = st.Status == stateErrorCar || st.Status == stateErrorEVSE
|
||||
st.LivePhases = countLivePhases(st.CurrentA)
|
||||
return st
|
||||
}
|
||||
|
||||
// livePhaseThreshold is the current above which a phase counts as carrying a
|
||||
// charge, in amperes. Greencell's own floor for a charging session is 6 A, so
|
||||
// anything under an amp is measurement noise on an idle phase.
|
||||
const livePhaseThreshold = 1.0
|
||||
|
||||
func countLivePhases(p Phases) int {
|
||||
n := 0
|
||||
for _, v := range []*float64{p.L1, p.L2, p.L3} {
|
||||
if v != nil && *v >= livePhaseThreshold {
|
||||
n++
|
||||
}
|
||||
}
|
||||
return n
|
||||
}
|
||||
|
||||
// normalizeState constrains a reported EVSE state to the known set.
|
||||
func normalizeState(s string) string {
|
||||
v := strings.ToLower(strings.TrimSpace(s))
|
||||
if evseStates[v] {
|
||||
return v
|
||||
}
|
||||
return stateUnknown
|
||||
}
|
||||
|
||||
// normalizeAccess maps a reported access level to the known set, folding the
|
||||
// deprecated OFFLINE into UNAVAILABLE as greencell_client does.
|
||||
func normalizeAccess(s string) string {
|
||||
switch strings.ToLower(strings.TrimSpace(s)) {
|
||||
case "execute":
|
||||
return accessExecute
|
||||
case "read":
|
||||
return accessRead
|
||||
case "disabled", "disable":
|
||||
return accessDisabled
|
||||
case "offline", "unavailable":
|
||||
return accessUnavailable
|
||||
default:
|
||||
return accessUnknown
|
||||
}
|
||||
}
|
||||
|
||||
// isUnavailable reports whether a status payload is one of the out-of-band
|
||||
// unavailability markers rather than a state document.
|
||||
func isUnavailable(payload json.RawMessage) bool {
|
||||
if payload == nil {
|
||||
return false
|
||||
}
|
||||
up := strings.ToUpper(string(payload))
|
||||
return strings.Contains(up, "UNAVAILABLE") || strings.Contains(up, "OFFLINE")
|
||||
}
|
||||
|
||||
// ---- payload parsing ---------------------------------------------------------
|
||||
|
||||
// parsePhases reads an {"l1":…,"l2":…,"l3":…} payload. A phase that is absent or
|
||||
// not a number stays nil — the device is documented to send numbers, and a
|
||||
// non-numeric value is better reported as missing than coerced to zero.
|
||||
func parsePhases(payload json.RawMessage) Phases {
|
||||
if payload == nil {
|
||||
return Phases{}
|
||||
}
|
||||
var doc map[string]json.RawMessage
|
||||
if err := json.Unmarshal(payload, &doc); err != nil {
|
||||
return Phases{}
|
||||
}
|
||||
return Phases{L1: asNumber(doc["l1"]), L2: asNumber(doc["l2"]), L3: asNumber(doc["l3"])}
|
||||
}
|
||||
|
||||
// scalePhases multiplies every present phase by factor.
|
||||
func scalePhases(p Phases, factor float64) Phases {
|
||||
scale := func(v *float64) *float64 {
|
||||
if v == nil {
|
||||
return nil
|
||||
}
|
||||
out := *v * factor
|
||||
return &out
|
||||
}
|
||||
return Phases{L1: scale(p.L1), L2: scale(p.L2), L3: scale(p.L3)}
|
||||
}
|
||||
|
||||
// numberField reads one numeric field out of a payload.
|
||||
func numberField(payload json.RawMessage, key string) *float64 {
|
||||
if payload == nil {
|
||||
return nil
|
||||
}
|
||||
var doc map[string]json.RawMessage
|
||||
if err := json.Unmarshal(payload, &doc); err != nil {
|
||||
return nil
|
||||
}
|
||||
return asNumber(doc[key])
|
||||
}
|
||||
|
||||
// stringField reads one string field out of a payload.
|
||||
func stringField(payload json.RawMessage, key string) (string, bool) {
|
||||
if payload == nil {
|
||||
return "", false
|
||||
}
|
||||
var doc map[string]json.RawMessage
|
||||
if err := json.Unmarshal(payload, &doc); err != nil {
|
||||
return "", false
|
||||
}
|
||||
var s string
|
||||
if err := json.Unmarshal(doc[key], &s); err != nil {
|
||||
return "", false
|
||||
}
|
||||
return s, true
|
||||
}
|
||||
|
||||
// asNumber decodes a JSON number, also accepting one quoted as a string, which
|
||||
// some firmware revisions do.
|
||||
func asNumber(raw json.RawMessage) *float64 {
|
||||
if len(raw) == 0 {
|
||||
return nil
|
||||
}
|
||||
// json.Unmarshal accepts null into a float64 and leaves it at zero, which on a
|
||||
// charger is a real reading; an explicit null means "no value".
|
||||
if string(raw) == "null" {
|
||||
return nil
|
||||
}
|
||||
var f float64
|
||||
if err := json.Unmarshal(raw, &f); err == nil {
|
||||
return &f
|
||||
}
|
||||
var s string
|
||||
if err := json.Unmarshal(raw, &s); err != nil {
|
||||
return nil
|
||||
}
|
||||
f, err := strconv.ParseFloat(strings.TrimSpace(s), 64)
|
||||
if err != nil {
|
||||
return nil
|
||||
}
|
||||
return &f
|
||||
}
|
||||
|
||||
// ---- config helpers ----------------------------------------------------------
|
||||
|
||||
// truthy reads a boolean-ish config value; the panel writes selects as strings.
|
||||
func truthy(v string) bool {
|
||||
switch strings.ToLower(strings.TrimSpace(v)) {
|
||||
case "on", "true", "yes", "1", "tls", "mqtts":
|
||||
return true
|
||||
default:
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
// parseTimeout reads the listen window in seconds, clamped to a range that keeps
|
||||
// an HTTP request honest at one end and useful at the other.
|
||||
func parseTimeout(v string) time.Duration {
|
||||
n, err := strconv.Atoi(strings.TrimSpace(v))
|
||||
if err != nil || n <= 0 {
|
||||
return defaultTimeout
|
||||
}
|
||||
d := time.Duration(n) * time.Second
|
||||
return max(minTimeout, min(d, maxTimeout))
|
||||
}
|
||||
|
||||
// shorten trims a message to something a health detail can carry.
|
||||
func shorten(s string) string {
|
||||
const limit = 200
|
||||
s = strings.TrimSpace(s)
|
||||
if len(s) <= limit {
|
||||
return s
|
||||
}
|
||||
return s[:limit] + "…"
|
||||
}
|
||||
@@ -0,0 +1,852 @@
|
||||
package greencell
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"context"
|
||||
"encoding/binary"
|
||||
"encoding/json"
|
||||
"io"
|
||||
"net"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"drivervault/apiserver/internal/plugins"
|
||||
)
|
||||
|
||||
// ---- descriptor & registration -----------------------------------------------
|
||||
|
||||
func TestDescriptor(t *testing.T) {
|
||||
d := (&Plugin{}).Descriptor()
|
||||
if d.Name != "greencell" {
|
||||
t.Fatalf("name = %q, want greencell", d.Name)
|
||||
}
|
||||
if d.Kind != plugins.KindBuiltin {
|
||||
t.Fatalf("kind = %q, want builtin", d.Kind)
|
||||
}
|
||||
if len(d.Capabilities) == 0 {
|
||||
t.Fatal("expected capabilities")
|
||||
}
|
||||
|
||||
fields := map[string]plugins.ConfigField{}
|
||||
for _, f := range d.ConfigFields {
|
||||
fields[f.Key] = f
|
||||
}
|
||||
for _, k := range []string{"host", "port", "tls", "username", "password", "serial", "commandTopic", "timeout"} {
|
||||
if _, ok := fields[k]; !ok {
|
||||
t.Errorf("config field %q should be present", k)
|
||||
}
|
||||
}
|
||||
// Nothing is required at the global layer — the per-user cascade may supply
|
||||
// the broker instead, exactly as it does for the other connectors.
|
||||
for k, f := range fields {
|
||||
if f.Required {
|
||||
t.Errorf("config field %q must not be required at the global layer", k)
|
||||
}
|
||||
}
|
||||
if !fields["password"].Secret {
|
||||
t.Error("the broker password must be marked secret")
|
||||
}
|
||||
if fields["username"].Secret {
|
||||
t.Error("the broker username is not a secret")
|
||||
}
|
||||
if fields["tls"].Type != "select" {
|
||||
t.Errorf("tls type = %q, want select", fields["tls"].Type)
|
||||
}
|
||||
if fields["port"].Default != defaultPort {
|
||||
t.Errorf("port default = %q, want %q", fields["port"].Default, defaultPort)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRegistered(t *testing.T) {
|
||||
// init() must have registered the factory; registering twice panics, so a
|
||||
// duplicate name would surface at startup rather than here.
|
||||
defer func() {
|
||||
if r := recover(); r == nil {
|
||||
t.Fatal("expected a panic on duplicate registration, meaning the plugin registered itself")
|
||||
}
|
||||
}()
|
||||
plugins.Register("greencell", func() plugins.Plugin { return &Plugin{} })
|
||||
}
|
||||
|
||||
// ---- config ------------------------------------------------------------------
|
||||
|
||||
func TestInitBuildsBrokerAddress(t *testing.T) {
|
||||
cases := []struct {
|
||||
name string
|
||||
config map[string]string
|
||||
address string
|
||||
tls bool
|
||||
}{
|
||||
{"explicit port", map[string]string{"host": "10.2.1.10", "port": "1884"}, "10.2.1.10:1884", false},
|
||||
{"default plain port", map[string]string{"host": "mqtt.local"}, "mqtt.local:1883", false},
|
||||
{"default tls port", map[string]string{"host": "mqtt.local", "tls": "on"}, "mqtt.local:8883", true},
|
||||
{"tls with explicit port", map[string]string{"host": "mqtt.local", "tls": "true", "port": "9001"}, "mqtt.local:9001", true},
|
||||
{"ipv6 host", map[string]string{"host": "fd00::1", "port": "1883"}, "[fd00::1]:1883", false},
|
||||
{"no host", map[string]string{}, "", false},
|
||||
}
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
var p Plugin
|
||||
if err := p.Init(context.Background(), tc.config); err != nil {
|
||||
t.Fatalf("Init: %v", err)
|
||||
}
|
||||
if p.address != tc.address {
|
||||
t.Errorf("address = %q, want %q", p.address, tc.address)
|
||||
}
|
||||
if p.useTLS != tc.tls {
|
||||
t.Errorf("useTLS = %v, want %v", p.useTLS, tc.tls)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestSnapshotWithoutBrokerIsAnError(t *testing.T) {
|
||||
var p Plugin
|
||||
_ = p.Init(context.Background(), map[string]string{})
|
||||
if _, _, err := p.snapshot(); err == nil {
|
||||
t.Fatal("expected an error when no broker is configured")
|
||||
}
|
||||
}
|
||||
|
||||
func TestParseTimeout(t *testing.T) {
|
||||
cases := map[string]time.Duration{
|
||||
"": defaultTimeout,
|
||||
"abc": defaultTimeout,
|
||||
"0": defaultTimeout,
|
||||
"-5": defaultTimeout,
|
||||
" 20 ": 20 * time.Second,
|
||||
"1": minTimeout,
|
||||
"600": maxTimeout, // clamped
|
||||
"999999": maxTimeout,
|
||||
}
|
||||
for in, want := range cases {
|
||||
if got := parseTimeout(in); got != want {
|
||||
t.Errorf("parseTimeout(%q) = %v, want %v", in, got, want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestTruthy(t *testing.T) {
|
||||
for _, v := range []string{"on", "ON", "true", "yes", "1", " tls "} {
|
||||
if !truthy(v) {
|
||||
t.Errorf("truthy(%q) = false", v)
|
||||
}
|
||||
}
|
||||
for _, v := range []string{"", "off", "false", "no", "0", "maybe"} {
|
||||
if truthy(v) {
|
||||
t.Errorf("truthy(%q) = true", v)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ---- serials & announcements -------------------------------------------------
|
||||
|
||||
func TestDeviceModel(t *testing.T) {
|
||||
// The HabuDen pattern comes from greencell_client's GreencellUtils.
|
||||
if got := deviceModel("EVGC021B22752405ZM0018"); got != nameHabuDen {
|
||||
t.Errorf("HabuDen serial named %q, want %q", got, nameHabuDen)
|
||||
}
|
||||
for _, sn := range []string{"EVGC031B22752405ZM0018", "EVGC022B22752405ZM0018", "HABU_DEN", ""} {
|
||||
if got := deviceModel(sn); got != nameGeneric {
|
||||
t.Errorf("deviceModel(%q) = %q, want %q", sn, got, nameGeneric)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestParseAnnouncement(t *testing.T) {
|
||||
dev, ok := parseAnnouncement([]byte(`{"id":"EVGC021B22752405ZM0018","fw":"1.2.3"}`))
|
||||
if !ok {
|
||||
t.Fatal("a payload with an id should parse")
|
||||
}
|
||||
if dev.SN != "EVGC021B22752405ZM0018" {
|
||||
t.Errorf("sn = %q", dev.SN)
|
||||
}
|
||||
if dev.Model != nameHabuDen || !strings.HasPrefix(dev.Name, nameHabuDen+" ") {
|
||||
t.Errorf("name/model = %q / %q", dev.Name, dev.Model)
|
||||
}
|
||||
// Fields this connector has no schema for must survive.
|
||||
if !strings.Contains(string(dev.Announcement), `"fw":"1.2.3"`) {
|
||||
t.Errorf("announcement should be passed through verbatim, got %s", dev.Announcement)
|
||||
}
|
||||
|
||||
for _, bad := range []string{`{"id":""}`, `{"id":" "}`, `{"name":"BROADCAST"}`, `not json`, ``} {
|
||||
if _, ok := parseAnnouncement([]byte(bad)); ok {
|
||||
t.Errorf("parseAnnouncement(%q) should have been rejected", bad)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestTopicSuffix(t *testing.T) {
|
||||
sn := "SN1"
|
||||
for _, want := range telemetryTopics {
|
||||
got, ok := topicSuffix(evsePrefix+sn+"/"+want, sn)
|
||||
if !ok || got != want {
|
||||
t.Errorf("topicSuffix for %q = %q/%v", want, got, ok)
|
||||
}
|
||||
}
|
||||
for _, topic := range []string{
|
||||
evsePrefix + "OTHER/current", // another charger
|
||||
evsePrefix + sn + "/unknown", // a topic we do not model
|
||||
discoveryTopic,
|
||||
evsePrefix + sn, // no suffix at all
|
||||
} {
|
||||
if _, ok := topicSuffix(topic, sn); ok {
|
||||
t.Errorf("topicSuffix(%q) should not match", topic)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ---- payload parsing ---------------------------------------------------------
|
||||
|
||||
func TestAsNumber(t *testing.T) {
|
||||
f := asNumber(json.RawMessage(`230.5`))
|
||||
if f == nil || *f != 230.5 {
|
||||
t.Errorf("number = %v", f)
|
||||
}
|
||||
// Some firmware quotes its numbers.
|
||||
f = asNumber(json.RawMessage(`"16000"`))
|
||||
if f == nil || *f != 16000 {
|
||||
t.Errorf("quoted number = %v", f)
|
||||
}
|
||||
for _, bad := range []string{`null`, `"abc"`, `{}`, ``} {
|
||||
if got := asNumber(json.RawMessage(bad)); got != nil {
|
||||
t.Errorf("asNumber(%q) = %v, want nil", bad, *got)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestParsePhasesLeavesMissingPhasesUnset(t *testing.T) {
|
||||
// A single-phase charge reports only l1; l2/l3 must stay nil rather than
|
||||
// reading as a real zero.
|
||||
p := parsePhases(json.RawMessage(`{"l1":16000}`))
|
||||
if p.L1 == nil || *p.L1 != 16000 {
|
||||
t.Errorf("l1 = %v", p.L1)
|
||||
}
|
||||
if p.L2 != nil || p.L3 != nil {
|
||||
t.Errorf("missing phases should be nil, got %v %v", p.L2, p.L3)
|
||||
}
|
||||
if got := parsePhases(json.RawMessage(`not json`)); got.L1 != nil {
|
||||
t.Error("a malformed payload should yield no phases")
|
||||
}
|
||||
if got := parsePhases(nil); got.L1 != nil {
|
||||
t.Error("a missing payload should yield no phases")
|
||||
}
|
||||
}
|
||||
|
||||
func TestNormalizeState(t *testing.T) {
|
||||
if got := normalizeState("CHARGING"); got != stateCharging {
|
||||
t.Errorf("CHARGING -> %q", got)
|
||||
}
|
||||
if got := normalizeState(" waiting_for_car "); got != stateWaitingForCar {
|
||||
t.Errorf("waiting_for_car -> %q", got)
|
||||
}
|
||||
for _, in := range []string{"", "SOMETHING_NEW", "UNKNOWN"} {
|
||||
if got := normalizeState(in); got != stateUnknown {
|
||||
t.Errorf("normalizeState(%q) = %q, want unknown", in, got)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestNormalizeAccess(t *testing.T) {
|
||||
cases := map[string]string{
|
||||
"EXECUTE": accessExecute, "read": accessRead, "DISABLED": accessDisabled,
|
||||
"UNAVAILABLE": accessUnavailable,
|
||||
// OFFLINE is greencell_client's deprecated spelling of UNAVAILABLE.
|
||||
"OFFLINE": accessUnavailable,
|
||||
"": accessUnknown, "nonsense": accessUnknown,
|
||||
}
|
||||
for in, want := range cases {
|
||||
if got := normalizeAccess(in); got != want {
|
||||
t.Errorf("normalizeAccess(%q) = %q, want %q", in, got, want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestBuildStateThreePhaseCharge(t *testing.T) {
|
||||
raw := map[string]json.RawMessage{
|
||||
topicCurrent: json.RawMessage(`{"l1":16000,"l2":15980,"l3":16010}`),
|
||||
topicVoltage: json.RawMessage(`{"l1":230.1,"l2":229.7,"l3":231.0}`),
|
||||
topicPower: json.RawMessage(`{"momentary":11040}`),
|
||||
topicStatus: json.RawMessage(`{"state":"CHARGING"}`),
|
||||
topicDeviceState: json.RawMessage(`{"level":"EXECUTE"}`),
|
||||
}
|
||||
st := buildState("EVGC021B22752405ZM0018", raw)
|
||||
|
||||
if !st.Complete {
|
||||
t.Error("all five topics reported, so the reading is complete")
|
||||
}
|
||||
// Current is published in milliamps.
|
||||
if st.CurrentA.L1 == nil || *st.CurrentA.L1 != 16 {
|
||||
t.Errorf("l1 current = %v A, want 16", st.CurrentA.L1)
|
||||
}
|
||||
if st.VoltageV.L1 == nil || *st.VoltageV.L1 != 230.1 {
|
||||
t.Errorf("l1 voltage = %v", st.VoltageV.L1)
|
||||
}
|
||||
if st.PowerW == nil || *st.PowerW != 11040 {
|
||||
t.Errorf("power = %v", st.PowerW)
|
||||
}
|
||||
if st.Status != stateCharging || !st.Charging || !st.Plugged || st.Fault {
|
||||
t.Errorf("status flags: %q charging=%v plugged=%v fault=%v", st.Status, st.Charging, st.Plugged, st.Fault)
|
||||
}
|
||||
if st.AccessLevel != accessExecute || !st.CanExecute || !st.Available {
|
||||
t.Errorf("access: %q canExecute=%v available=%v", st.AccessLevel, st.CanExecute, st.Available)
|
||||
}
|
||||
if st.LivePhases != 3 {
|
||||
t.Errorf("livePhases = %d, want 3", st.LivePhases)
|
||||
}
|
||||
if st.Model != nameHabuDen {
|
||||
t.Errorf("model = %q", st.Model)
|
||||
}
|
||||
if st.ObservedAt.IsZero() {
|
||||
t.Error("observedAt should be stamped")
|
||||
}
|
||||
}
|
||||
|
||||
func TestBuildStateSinglePhaseAndPartial(t *testing.T) {
|
||||
// Only two topics reported: the rest must read as unset, not as zero.
|
||||
raw := map[string]json.RawMessage{
|
||||
topicCurrent: json.RawMessage(`{"l1":16000,"l2":30,"l3":0}`),
|
||||
topicStatus: json.RawMessage(`{"state":"CHARGING"}`),
|
||||
}
|
||||
st := buildState("SN1", raw)
|
||||
|
||||
if st.Complete {
|
||||
t.Error("a partial reading must not claim to be complete")
|
||||
}
|
||||
if st.Received[topicCurrent] != true || st.Received[topicPower] != false {
|
||||
t.Errorf("received map = %v", st.Received)
|
||||
}
|
||||
if st.PowerW != nil {
|
||||
t.Errorf("power = %v, want nil when the topic was silent", *st.PowerW)
|
||||
}
|
||||
// 30 mA is 0.03 A — noise on an idle phase, not a live one.
|
||||
if st.LivePhases != 1 {
|
||||
t.Errorf("livePhases = %d, want 1", st.LivePhases)
|
||||
}
|
||||
if st.AccessLevel != accessUnknown {
|
||||
t.Errorf("access level = %q, want unknown when device_state was silent", st.AccessLevel)
|
||||
}
|
||||
if st.Model != nameGeneric {
|
||||
t.Errorf("model = %q, want the generic name for a non-HabuDen serial", st.Model)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBuildStateUnavailable(t *testing.T) {
|
||||
// The device signals unavailability through the status topic itself, which
|
||||
// upstream detects by substring because the payload is not a state document.
|
||||
for _, payload := range []string{`{"state":"UNAVAILABLE"}`, `"OFFLINE"`, `{"status":"UNAVAILABLE"}`} {
|
||||
st := buildState("SN1", map[string]json.RawMessage{
|
||||
topicStatus: json.RawMessage(payload),
|
||||
topicDeviceState: json.RawMessage(`{"level":"READ"}`),
|
||||
})
|
||||
if st.Status != stateUnavailable {
|
||||
t.Errorf("payload %s -> status %q, want unavailable", payload, st.Status)
|
||||
}
|
||||
if st.Available {
|
||||
t.Errorf("payload %s should not read as available", payload)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestBuildStateErrorStates(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
state string
|
||||
fault bool
|
||||
plugged bool
|
||||
}{
|
||||
{"ERROR_CAR", true, true},
|
||||
{"ERROR_EVSE", true, false},
|
||||
{"IDLE", false, false},
|
||||
{"FINISHED", false, true},
|
||||
} {
|
||||
st := buildState("SN1", map[string]json.RawMessage{
|
||||
topicStatus: json.RawMessage(`{"state":"` + tc.state + `"}`),
|
||||
})
|
||||
if st.Fault != tc.fault || st.Plugged != tc.plugged {
|
||||
t.Errorf("%s: fault=%v plugged=%v, want %v/%v", tc.state, st.Fault, st.Plugged, tc.fault, tc.plugged)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestBuildStateDisabledIsNotAvailable(t *testing.T) {
|
||||
st := buildState("SN1", map[string]json.RawMessage{
|
||||
topicStatus: json.RawMessage(`{"state":"IDLE"}`),
|
||||
topicDeviceState: json.RawMessage(`{"level":"DISABLED"}`),
|
||||
})
|
||||
if st.Available || st.CanExecute {
|
||||
t.Errorf("a DISABLED device is neither available nor executable: %+v", st)
|
||||
}
|
||||
}
|
||||
|
||||
// ---- Invoke dispatch ---------------------------------------------------------
|
||||
|
||||
func TestInvokeUnknownAction(t *testing.T) {
|
||||
var p Plugin
|
||||
_ = p.Init(context.Background(), map[string]string{"host": "127.0.0.1"})
|
||||
if _, err := p.Invoke(context.Background(), "nope", nil); err == nil {
|
||||
t.Fatal("expected an error for an unknown action")
|
||||
}
|
||||
}
|
||||
|
||||
func TestInvokeChargerStateRequiresSerial(t *testing.T) {
|
||||
var p Plugin
|
||||
_ = p.Init(context.Background(), map[string]string{"host": "127.0.0.1"})
|
||||
_, err := p.Invoke(context.Background(), "charger-state", nil)
|
||||
if err == nil || !strings.Contains(err.Error(), "serial") {
|
||||
t.Fatalf("error = %v, want a missing-serial error", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestInvokeRejectsMalformedParams(t *testing.T) {
|
||||
var p Plugin
|
||||
_ = p.Init(context.Background(), map[string]string{"host": "127.0.0.1"})
|
||||
if _, err := p.Invoke(context.Background(), "chargers", json.RawMessage(`[`)); err == nil {
|
||||
t.Fatal("expected an error for malformed params")
|
||||
}
|
||||
}
|
||||
|
||||
// ---- end to end against a broker ---------------------------------------------
|
||||
|
||||
func TestDiscoverCollectsEveryDeviceThatAnswers(t *testing.T) {
|
||||
addr := startBroker(t, []brokerMsg{
|
||||
{discoveryTopic, `{"id":"EVGC021B22752405ZM0018"}`},
|
||||
{discoveryTopic, `{"id":"SOMETHINGELSE"}`},
|
||||
{discoveryTopic, `{"id":"EVGC021B22752405ZM0018"}`}, // a repeat is one device
|
||||
{discoveryTopic, `{"name":"BROADCAST"}`}, // our own echo, not a device
|
||||
})
|
||||
|
||||
p := pluginAt(t, addr, map[string]string{"timeout": "5"})
|
||||
devices, err := p.discover(context.Background())
|
||||
if err != nil {
|
||||
t.Fatalf("discover: %v", err)
|
||||
}
|
||||
if len(devices) != 2 {
|
||||
t.Fatalf("found %d devices, want 2: %+v", len(devices), devices)
|
||||
}
|
||||
if devices[0].SN != "EVGC021B22752405ZM0018" || devices[0].Model != nameHabuDen {
|
||||
t.Errorf("first device = %+v", devices[0])
|
||||
}
|
||||
if devices[1].Model != nameGeneric {
|
||||
t.Errorf("second device model = %q, want the generic name", devices[1].Model)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDiscoverReturnsEmptyOnSilence(t *testing.T) {
|
||||
addr := startBroker(t, nil)
|
||||
p := pluginAt(t, addr, map[string]string{"timeout": "1"})
|
||||
devices, err := p.discover(context.Background())
|
||||
if err != nil {
|
||||
t.Fatalf("discover: %v", err)
|
||||
}
|
||||
if len(devices) != 0 {
|
||||
t.Fatalf("want no devices, got %+v", devices)
|
||||
}
|
||||
}
|
||||
|
||||
func TestChargerStateReadsTelemetry(t *testing.T) {
|
||||
sn := "EVGC021B22752405ZM0018"
|
||||
addr := startBroker(t, []brokerMsg{
|
||||
{evsePrefix + sn + "/current", `{"l1":16000,"l2":0,"l3":0}`},
|
||||
{evsePrefix + sn + "/voltage", `{"l1":230,"l2":231,"l3":229}`},
|
||||
{evsePrefix + sn + "/power", `{"momentary":3680}`},
|
||||
{evsePrefix + sn + "/status", `{"state":"CHARGING"}`},
|
||||
{evsePrefix + sn + "/device_state", `{"level":"READ"}`},
|
||||
// Another charger on the same broker must not bleed into this reading.
|
||||
{evsePrefix + "OTHER/power", `{"momentary":99999}`},
|
||||
})
|
||||
|
||||
p := pluginAt(t, addr, map[string]string{"timeout": "5"})
|
||||
st, err := p.chargerState(context.Background(), sn)
|
||||
if err != nil {
|
||||
t.Fatalf("chargerState: %v", err)
|
||||
}
|
||||
if !st.Complete {
|
||||
t.Errorf("expected a complete reading, got %v", st.Received)
|
||||
}
|
||||
if st.PowerW == nil || *st.PowerW != 3680 {
|
||||
t.Errorf("power = %v, want 3680 (the other charger's reading must not leak in)", st.PowerW)
|
||||
}
|
||||
if st.CurrentA.L1 == nil || *st.CurrentA.L1 != 16 {
|
||||
t.Errorf("l1 = %v A, want 16", st.CurrentA.L1)
|
||||
}
|
||||
if st.LivePhases != 1 {
|
||||
t.Errorf("livePhases = %d, want 1", st.LivePhases)
|
||||
}
|
||||
if st.AccessLevel != accessRead || st.CanExecute {
|
||||
t.Errorf("READ mode should not report as executable: %+v", st)
|
||||
}
|
||||
}
|
||||
|
||||
func TestChargerStateReturnsPartialAtTheDeadline(t *testing.T) {
|
||||
sn := "SN1"
|
||||
addr := startBroker(t, []brokerMsg{
|
||||
{evsePrefix + sn + "/status", `{"state":"IDLE"}`},
|
||||
})
|
||||
|
||||
p := pluginAt(t, addr, map[string]string{"timeout": "1"})
|
||||
st, err := p.chargerState(context.Background(), sn)
|
||||
if err != nil {
|
||||
t.Fatalf("a partial reading should not be an error: %v", err)
|
||||
}
|
||||
if st.Complete || st.Status != stateIdle {
|
||||
t.Errorf("state = %+v", st)
|
||||
}
|
||||
if st.Received[topicVoltage] {
|
||||
t.Error("voltage was never published, so it must read as not received")
|
||||
}
|
||||
}
|
||||
|
||||
func TestChargerStateErrorsOnSilence(t *testing.T) {
|
||||
addr := startBroker(t, nil)
|
||||
p := pluginAt(t, addr, map[string]string{"timeout": "1"})
|
||||
_, err := p.chargerState(context.Background(), "SN1")
|
||||
if err == nil || !strings.Contains(err.Error(), "no data from charger") {
|
||||
t.Fatalf("error = %v, want a no-data error", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestChargerStateFailsWithoutABroker(t *testing.T) {
|
||||
// Nothing is listening, so the session cannot be opened at all.
|
||||
p := pluginAt(t, "127.0.0.1:1", map[string]string{"timeout": "1"})
|
||||
if _, err := p.chargerState(context.Background(), "SN1"); err == nil {
|
||||
t.Fatal("expected a connection error")
|
||||
}
|
||||
}
|
||||
|
||||
func TestHealthCheckClassifiesTheThreeOutcomes(t *testing.T) {
|
||||
t.Run("ok", func(t *testing.T) {
|
||||
addr := startBroker(t, []brokerMsg{{discoveryTopic, `{"id":"EVGC021B22752405ZM0018"}`}})
|
||||
p := pluginAt(t, addr, map[string]string{"timeout": "5"})
|
||||
h := p.HealthCheck(context.Background())
|
||||
if h.Status != plugins.StatusOK {
|
||||
t.Fatalf("status = %q (%s), want ok", h.Status, h.Detail)
|
||||
}
|
||||
if !strings.Contains(h.Detail, "1 Greencell device") {
|
||||
t.Errorf("detail = %q", h.Detail)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("degraded when the broker answers but no charger does", func(t *testing.T) {
|
||||
addr := startBroker(t, nil)
|
||||
p := pluginAt(t, addr, map[string]string{"timeout": "1"})
|
||||
h := p.HealthCheck(context.Background())
|
||||
if h.Status != plugins.StatusDegraded {
|
||||
t.Fatalf("status = %q (%s), want degraded", h.Status, h.Detail)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("down when the broker is unreachable", func(t *testing.T) {
|
||||
p := pluginAt(t, "127.0.0.1:1", map[string]string{"timeout": "1"})
|
||||
h := p.HealthCheck(context.Background())
|
||||
if h.Status != plugins.StatusDown {
|
||||
t.Fatalf("status = %q (%s), want down", h.Status, h.Detail)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func TestInvokeChargersShape(t *testing.T) {
|
||||
addr := startBroker(t, []brokerMsg{{discoveryTopic, `{"id":"EVGC021B22752405ZM0018"}`}})
|
||||
p := pluginAt(t, addr, map[string]string{"timeout": "5"})
|
||||
|
||||
raw, err := p.Invoke(context.Background(), "chargers", nil)
|
||||
if err != nil {
|
||||
t.Fatalf("Invoke: %v", err)
|
||||
}
|
||||
var doc struct {
|
||||
Chargers []Device `json:"chargers"`
|
||||
}
|
||||
if err := json.Unmarshal(raw, &doc); err != nil {
|
||||
t.Fatalf("unmarshal: %v", err)
|
||||
}
|
||||
if len(doc.Chargers) != 1 || doc.Chargers[0].SN != "EVGC021B22752405ZM0018" {
|
||||
t.Fatalf("chargers = %+v", doc.Chargers)
|
||||
}
|
||||
}
|
||||
|
||||
func TestInvokeChargerStateFallsBackToTheConfiguredSerial(t *testing.T) {
|
||||
sn := "EVGC021B22752405ZM0018"
|
||||
addr := startBroker(t, []brokerMsg{{evsePrefix + sn + "/status", `{"state":"FINISHED"}`}})
|
||||
p := pluginAt(t, addr, map[string]string{"timeout": "1", "serial": sn})
|
||||
|
||||
raw, err := p.Invoke(context.Background(), "charger-state", nil)
|
||||
if err != nil {
|
||||
t.Fatalf("Invoke: %v", err)
|
||||
}
|
||||
var st State
|
||||
if err := json.Unmarshal(raw, &st); err != nil {
|
||||
t.Fatalf("unmarshal: %v", err)
|
||||
}
|
||||
if st.SN != sn || st.Status != stateFinished {
|
||||
t.Fatalf("state = %+v", st)
|
||||
}
|
||||
}
|
||||
|
||||
// ---- test broker -------------------------------------------------------------
|
||||
|
||||
// brokerMsg is one publication the test broker pushes once the plugin has
|
||||
// subscribed and sent its discovery broadcast.
|
||||
type brokerMsg struct {
|
||||
topic string
|
||||
payload string
|
||||
}
|
||||
|
||||
// pluginAt returns a plugin configured to talk to addr.
|
||||
func pluginAt(t *testing.T, addr string, extra map[string]string) *Plugin {
|
||||
t.Helper()
|
||||
host, port, err := net.SplitHostPort(addr)
|
||||
if err != nil {
|
||||
t.Fatalf("bad address %q: %v", addr, err)
|
||||
}
|
||||
config := map[string]string{"host": host, "port": port}
|
||||
for k, v := range extra {
|
||||
config[k] = v
|
||||
}
|
||||
p := &Plugin{}
|
||||
if err := p.Init(context.Background(), config); err != nil {
|
||||
t.Fatalf("Init: %v", err)
|
||||
}
|
||||
return p
|
||||
}
|
||||
|
||||
// startBroker runs an MQTT 3.1.1 broker that does just enough for these tests:
|
||||
// accept a connection, acknowledge the subscription, and — once the client sends
|
||||
// its discovery broadcast — push the scripted messages back. It speaks the wire
|
||||
// format directly rather than reusing internal/mqtt, so a bug in the client
|
||||
// cannot hide behind a matching bug in the fixture.
|
||||
func startBroker(t *testing.T, script []brokerMsg) string {
|
||||
t.Helper()
|
||||
return startBrokerRecording(t, nil, script)
|
||||
}
|
||||
|
||||
// startBrokerRecording is startBroker with a channel that receives everything the
|
||||
// client publishes, so a test can assert on what went out as well as what came
|
||||
// back. A nil channel records nothing.
|
||||
func startBrokerRecording(t *testing.T, seen chan<- brokerMsg, script []brokerMsg) string {
|
||||
t.Helper()
|
||||
ln, err := net.Listen("tcp", "127.0.0.1:0")
|
||||
if err != nil {
|
||||
t.Fatalf("listen: %v", err)
|
||||
}
|
||||
var wg sync.WaitGroup
|
||||
// Close the listener before waiting: the accept loop only ends when the
|
||||
// listener does, and cleanups run last-registered first.
|
||||
t.Cleanup(func() { _ = ln.Close(); wg.Wait() })
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
for {
|
||||
conn, err := ln.Accept()
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
defer conn.Close()
|
||||
serveBroker(conn, seen, script)
|
||||
}()
|
||||
}
|
||||
}()
|
||||
return ln.Addr().String()
|
||||
}
|
||||
|
||||
func serveBroker(conn net.Conn, seen chan<- brokerMsg, script []brokerMsg) {
|
||||
br := bufio.NewReader(conn)
|
||||
for {
|
||||
typ, _, body, err := readRaw(br)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
switch typ {
|
||||
case 1: // CONNECT
|
||||
if _, err := conn.Write(frame(2, 0, []byte{0, 0})); err != nil {
|
||||
return
|
||||
}
|
||||
case 8: // SUBSCRIBE — acknowledge every requested filter with QoS 0
|
||||
if len(body) < 2 {
|
||||
return
|
||||
}
|
||||
ack := append([]byte{body[0], body[1]}, make([]byte, countFilters(body[2:]))...)
|
||||
if _, err := conn.Write(frame(9, 0, ack)); err != nil {
|
||||
return
|
||||
}
|
||||
case 3: // PUBLISH — from the client; record it, then answer with the script
|
||||
if seen != nil {
|
||||
if topic, rest, err := splitRawString(body); err == nil {
|
||||
select {
|
||||
case seen <- brokerMsg{topic, string(rest)}:
|
||||
default:
|
||||
}
|
||||
}
|
||||
}
|
||||
for _, m := range script {
|
||||
pub := append(encRawString(m.topic), m.payload...)
|
||||
if _, err := conn.Write(frame(3, 0, pub)); err != nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
case 12: // PINGREQ
|
||||
if _, err := conn.Write(frame(13, 0, nil)); err != nil {
|
||||
return
|
||||
}
|
||||
case 14: // DISCONNECT
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// countFilters counts the topic filters in a SUBSCRIBE payload so the SUBACK can
|
||||
// carry one return code per filter.
|
||||
func countFilters(b []byte) int {
|
||||
n := 0
|
||||
for len(b) >= 3 {
|
||||
l := int(binary.BigEndian.Uint16(b))
|
||||
if len(b) < 2+l+1 {
|
||||
return n
|
||||
}
|
||||
b = b[2+l+1:]
|
||||
n++
|
||||
}
|
||||
return n
|
||||
}
|
||||
|
||||
// splitRawString peels an MQTT UTF-8 string off the front of b.
|
||||
func splitRawString(b []byte) (string, []byte, error) {
|
||||
if len(b) < 2 {
|
||||
return "", nil, io.ErrUnexpectedEOF
|
||||
}
|
||||
n := int(binary.BigEndian.Uint16(b))
|
||||
if len(b) < 2+n {
|
||||
return "", nil, io.ErrUnexpectedEOF
|
||||
}
|
||||
return string(b[2 : 2+n]), b[2+n:], nil
|
||||
}
|
||||
|
||||
func encRawString(s string) []byte {
|
||||
out := binary.BigEndian.AppendUint16(nil, uint16(len(s)))
|
||||
return append(out, s...)
|
||||
}
|
||||
|
||||
func frame(typ, flags byte, body []byte) []byte {
|
||||
out := []byte{typ<<4 | flags}
|
||||
n := len(body)
|
||||
for {
|
||||
digit := byte(n % 128)
|
||||
n /= 128
|
||||
if n > 0 {
|
||||
digit |= 0x80
|
||||
}
|
||||
out = append(out, digit)
|
||||
if n == 0 {
|
||||
break
|
||||
}
|
||||
}
|
||||
return append(out, body...)
|
||||
}
|
||||
|
||||
func readRaw(br *bufio.Reader) (typ, flags byte, body []byte, err error) {
|
||||
head, err := br.ReadByte()
|
||||
if err != nil {
|
||||
return 0, 0, nil, err
|
||||
}
|
||||
var (
|
||||
length int
|
||||
multiplier = 1
|
||||
)
|
||||
for i := 0; i < 4; i++ {
|
||||
digit, err := br.ReadByte()
|
||||
if err != nil {
|
||||
return 0, 0, nil, err
|
||||
}
|
||||
length += int(digit&0x7F) * multiplier
|
||||
if digit&0x80 == 0 {
|
||||
break
|
||||
}
|
||||
multiplier *= 128
|
||||
}
|
||||
body = make([]byte, length)
|
||||
if _, err := io.ReadFull(br, body); err != nil {
|
||||
return 0, 0, nil, err
|
||||
}
|
||||
return head >> 4, head & 0x0F, body, nil
|
||||
}
|
||||
|
||||
// ---- the operator-supplied QUERY topic ---------------------------------------
|
||||
|
||||
func TestCommandTopicFor(t *testing.T) {
|
||||
var p Plugin
|
||||
_ = p.Init(context.Background(), map[string]string{
|
||||
"host": "h", "commandTopic": "/greencell/evse/" + snPlaceholder + "/command",
|
||||
})
|
||||
if got := p.commandTopicFor("SN1"); got != "/greencell/evse/SN1/command" {
|
||||
t.Errorf("commandTopicFor = %q", got)
|
||||
}
|
||||
|
||||
// A topic without the placeholder is used as given.
|
||||
_ = p.Init(context.Background(), map[string]string{"host": "h", "commandTopic": "/fixed/topic"})
|
||||
if got := p.commandTopicFor("SN1"); got != "/fixed/topic" {
|
||||
t.Errorf("commandTopicFor = %q, want the literal topic", got)
|
||||
}
|
||||
|
||||
// Unset means listen-only; nothing is ever published.
|
||||
_ = p.Init(context.Background(), map[string]string{"host": "h"})
|
||||
if got := p.commandTopicFor("SN1"); got != "" {
|
||||
t.Errorf("commandTopicFor = %q, want empty when unconfigured", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestChargerStateSendsQueryWhenATopicIsConfigured(t *testing.T) {
|
||||
sn := "SN1"
|
||||
seen := make(chan brokerMsg, 4)
|
||||
addr := startBrokerRecording(t, seen, []brokerMsg{
|
||||
{evsePrefix + sn + "/status", `{"state":"IDLE"}`},
|
||||
})
|
||||
|
||||
p := pluginAt(t, addr, map[string]string{
|
||||
"timeout": "1", "commandTopic": "/greencell/evse/" + snPlaceholder + "/command",
|
||||
})
|
||||
if _, err := p.chargerState(context.Background(), sn); err != nil {
|
||||
t.Fatalf("chargerState: %v", err)
|
||||
}
|
||||
|
||||
var got []brokerMsg
|
||||
close(seen)
|
||||
for m := range seen {
|
||||
got = append(got, m)
|
||||
}
|
||||
// The discovery broadcast, then the QUERY on the configured topic.
|
||||
var query *brokerMsg
|
||||
for i := range got {
|
||||
if got[i].topic == "/greencell/evse/SN1/command" {
|
||||
query = &got[i]
|
||||
}
|
||||
}
|
||||
if query == nil {
|
||||
t.Fatalf("no QUERY was published; broker saw %+v", got)
|
||||
}
|
||||
if query.payload != queryCommand {
|
||||
t.Errorf("QUERY payload = %q, want %q", query.payload, queryCommand)
|
||||
}
|
||||
}
|
||||
|
||||
func TestChargerStateSendsNothingWithoutACommandTopic(t *testing.T) {
|
||||
sn := "SN1"
|
||||
seen := make(chan brokerMsg, 4)
|
||||
addr := startBrokerRecording(t, seen, []brokerMsg{
|
||||
{evsePrefix + sn + "/status", `{"state":"IDLE"}`},
|
||||
})
|
||||
|
||||
p := pluginAt(t, addr, map[string]string{"timeout": "1"})
|
||||
if _, err := p.chargerState(context.Background(), sn); err != nil {
|
||||
t.Fatalf("chargerState: %v", err)
|
||||
}
|
||||
|
||||
close(seen)
|
||||
for m := range seen {
|
||||
if m.topic != broadcastTopic {
|
||||
t.Errorf("unconfigured, the plugin must publish only the discovery broadcast; it also sent %+v", m)
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user