graham/towerops-agent
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

cleanup

Browse source
This commit is contained in:
Graham McIntire 2026-02-11 10:07:27 -06:00
parent 769ac1f623
commit 37f411ec55
No known key found for this signature in database
2 changed files with 19 additions and 19 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

26
agent.go
View file

@ -17,15 +17,15 @@ import (
"google.golang.org/protobuf/proto" "google.golang.org/protobuf/proto"
) )
// phoenixMsg is the Phoenix channel message format (JSON wrapper around binary protobuf). // channelMsg is the WebSocket channel message format (JSON wrapper around binary protobuf).
type phoenixMsg struct { type channelMsg struct {
Topic string `json:"topic"` Topic string `json:"topic"`
Event string `json:"event"` Event string `json:"event"`
Payload json.RawMessage `json:"payload"` Payload json.RawMessage `json:"payload"`
Ref *string `json:"ref"` Ref *string `json:"ref"`
} }
// runAgent connects to the Phoenix server and runs the event loop with reconnect. // runAgent connects to the server and runs the event loop with reconnect.
func runAgent(ctx context.Context, wsURL, token string) { func runAgent(ctx context.Context, wsURL, token string) {
baseURL := strings.TrimRight(wsURL, "/") baseURL := strings.TrimRight(wsURL, "/")
retryDelay := time.Second retryDelay := time.Second
@ -91,7 +91,7 @@ func runSession(ctx context.Context, baseURL, token string) error {
} }
sendMsg := func(event string, payload json.RawMessage) { sendMsg := func(event string, payload json.RawMessage) {
msg := phoenixMsg{ msg := channelMsg{
Topic: topic, Topic: topic,
Event: event, Event: event,
Payload: payload, Payload: payload,
@ -120,7 +120,7 @@ func runSession(ctx context.Context, baseURL, token string) error {
// Join channel // Join channel
joinPayload, _ := json.Marshal(map[string]string{"token": token}) joinPayload, _ := json.Marshal(map[string]string{"token": token})
joinMsg := phoenixMsg{ joinMsg := channelMsg{
Topic: topic, Topic: topic,
Event: "phx_join", Event: "phx_join",
Payload: joinPayload, Payload: joinPayload,
@ -161,8 +161,8 @@ func runSession(ctx context.Context, baseURL, token string) error {
heartbeatTicker := time.NewTicker(60 * time.Second) heartbeatTicker := time.NewTicker(60 * time.Second)
defer heartbeatTicker.Stop() defer heartbeatTicker.Stop()
phxHeartbeatTicker := time.NewTicker(25 * time.Second) channelHeartbeatTicker := time.NewTicker(25 * time.Second)
defer phxHeartbeatTicker.Stop() defer channelHeartbeatTicker.Stop()
startTime := time.Now() startTime := time.Now()
defer func() { defer func() {
@ -180,7 +180,7 @@ func runSession(ctx context.Context, baseURL, token string) error {
return fmt.Errorf("read: %w", err) return fmt.Errorf("read: %w", err)
case data := <-msgCh: case data := <-msgCh:
var msg phoenixMsg var msg channelMsg
if err := json.Unmarshal(data, &msg); err != nil { if err := json.Unmarshal(data, &msg); err != nil {
slog.Warn("invalid message", "error", err) slog.Warn("invalid message", "error", err)
continue continue
@ -212,9 +212,9 @@ func runSession(ctx context.Context, baseURL, token string) error {
sendBinaryResult("heartbeat", hb) sendBinaryResult("heartbeat", hb)
slog.Debug("sent heartbeat") slog.Debug("sent heartbeat")
case <-phxHeartbeatTicker.C: case <-channelHeartbeatTicker.C:
ref := nextRef() ref := nextRef()
msg := phoenixMsg{ msg := channelMsg{
Topic: "phoenix", Topic: "phoenix",
Event: "heartbeat", Event: "heartbeat",
Payload: json.RawMessage(`{}`), Payload: json.RawMessage(`{}`),
@ -225,14 +225,14 @@ func runSession(ctx context.Context, baseURL, token string) error {
case writeCh <- data: case writeCh <- data:
default: default:
} }
slog.Debug("sent phoenix heartbeat", "ref", ref) slog.Debug("sent channel heartbeat", "ref", ref)
} }
} }
} }
// handleMessage dispatches incoming Phoenix channel messages. // handleMessage dispatches incoming channel messages.
func handleMessage( func handleMessage(
msg phoenixMsg, msg channelMsg,
snmpResultCh chan<- *pb.SnmpResult, snmpResultCh chan<- *pb.SnmpResult,
mikrotikResultCh chan<- *pb.MikrotikResult, mikrotikResultCh chan<- *pb.MikrotikResult,
credTestResultCh chan<- *pb.CredentialTestResult, credTestResultCh chan<- *pb.CredentialTestResult,

12
agent_test.go
View file

@ -5,8 +5,8 @@ import (
"testing" "testing"
) )
func TestPhoenixMsgSerialization(t *testing.T) { func TestChannelMsgSerialization(t *testing.T) {
msg := phoenixMsg{ msg := channelMsg{
Topic: "agent:123", Topic: "agent:123",
Event: "phx_join", Event: "phx_join",
Payload: json.RawMessage(`{"token":"test"}`), Payload: json.RawMessage(`{"token":"test"}`),
@ -27,9 +27,9 @@ func TestPhoenixMsgSerialization(t *testing.T) {
} }
} }
func TestPhoenixMsgDeserialization(t *testing.T) { func TestChannelMsgDeserialization(t *testing.T) {
raw := `{"topic":"agent:123","event":"phx_reply","payload":{"status":"ok"},"ref":"1"}` raw := `{"topic":"agent:123","event":"phx_reply","payload":{"status":"ok"},"ref":"1"}`
var msg phoenixMsg var msg channelMsg
if err := json.Unmarshal([]byte(raw), &msg); err != nil { if err := json.Unmarshal([]byte(raw), &msg); err != nil {
t.Fatal(err) t.Fatal(err)
} }
@ -44,9 +44,9 @@ func TestPhoenixMsgDeserialization(t *testing.T) {
} }
} }
func TestPhoenixMsgNullRef(t *testing.T) { func TestChannelMsgNullRef(t *testing.T) {
raw := `{"topic":"agent:123","event":"job","payload":{},"ref":null}` raw := `{"topic":"agent:123","event":"job","payload":{},"ref":null}`
var msg phoenixMsg var msg channelMsg
if err := json.Unmarshal([]byte(raw), &msg); err != nil { if err := json.Unmarshal([]byte(raw), &msg); err != nil {
t.Fatal(err) t.Fatal(err)
} }