add brokers api endpoint

This commit is contained in:
Enot (ded) Skelly
2026-05-27 10:09:08 -07:00
parent 1b16bdc297
commit 67bbe57e26
4 changed files with 62 additions and 10 deletions
+1 -2
View File
@@ -124,9 +124,8 @@ func main() {
go broker1.Start(ctx)
go broker2.Start(ctx)
// ── HTTP server ──────────────────────────────────────────────────────────
r := router.New(h, store)
r := router.New(h, store, []*ingest.Worker{broker1, broker2})
srv := &http.Server{
Addr: addr,
+38
View File
@@ -0,0 +1,38 @@
package handlers
import (
"net/http"
"github.com/MeshCore-Tower/tower-server/internal/ingest"
"github.com/go-chi/chi/v5"
)
type BrokerStatus struct {
Name string `json:"name"`
Connected bool `json:"connected"`
}
// BrokersRouter mounts all /brokers routes onto a subrouter.
//
// GET /brokers → ListBrokers
//
// Note: broker configuration is managed via the server config
// file, not the API (v1). These endpoints are read-only.
func BrokersRouter(workers []*ingest.Worker) http.Handler {
r := chi.NewRouter()
// GET /brokers → ListBrokers
//
// Returns all configured brokers and their connection status
r.Get("/", func(w http.ResponseWriter, r *http.Request) {
brokers := make([]BrokerStatus, len(workers))
for i, v := range workers {
brokers[i] = BrokerStatus{
Name: v.BrokerName(),
Connected: v.IsConnected(),
}
}
respond(w, http.StatusOK, brokers)
})
return r
}
+4 -1
View File
@@ -10,6 +10,7 @@ import (
"github.com/MeshCore-Tower/tower-server/internal/api/handlers"
mw "github.com/MeshCore-Tower/tower-server/internal/api/middleware"
"github.com/MeshCore-Tower/tower-server/internal/hub"
"github.com/MeshCore-Tower/tower-server/internal/ingest"
"github.com/MeshCore-Tower/tower-server/internal/ws"
)
@@ -21,6 +22,7 @@ import (
// /api/v1/ → public group
// /packets → packets subrouter
// /nodes → nodes subrouter
// /brokers → brokers subrouter
// /observers → observers subrouter
// /channels → channels subrouter
// /iatas → iatas subrouter
@@ -29,7 +31,7 @@ import (
//
// The private group is stubbed and ready for the auth middleware drop-in
// described in Future Features → Admin authentication.
func New(h *hub.Hub, reader api.Reader) http.Handler {
func New(h *hub.Hub, reader api.Reader, workers []*ingest.Worker) http.Handler {
r := chi.NewRouter()
// ── Global middleware ────────────────────────────────────────────────────
@@ -49,6 +51,7 @@ func New(h *hub.Hub, reader api.Reader) http.Handler {
r.Group(func(r chi.Router) {
r.Mount("/packets", handlers.PacketsRouter())
r.Mount("/nodes", handlers.NodesRouter())
r.Mount("/brokers", handlers.BrokersRouter(workers))
r.Mount("/observers", handlers.ObserversRouter(reader))
r.Mount("/channels", handlers.ChannelsRouter(reader))
r.Mount("/messages", handlers.MessagesRouter(reader))
+19 -7
View File
@@ -249,10 +249,11 @@ type ChannelKeyStore interface {
// Worker holds the dependencies for one broker's ingest loop.
type Worker struct {
cfg Config
db DB
hub *hub.Hub
keys ChannelKeyStore
cfg Config
db DB
hub *hub.Hub
keys ChannelKeyStore
client mqtt.Client
}
// New creates an ingest Worker. Call Start() to connect and begin processing.
@@ -281,17 +282,28 @@ func (w *Worker) Start(ctx context.Context) {
log.Printf("ingest[%s]: connection lost: %v", w.cfg.BrokerName, err)
})
client := mqtt.NewClient(opts)
if tok := client.Connect(); tok.Wait() && tok.Error() != nil {
w.client = mqtt.NewClient(opts)
if tok := w.client.Connect(); tok.Wait() && tok.Error() != nil {
log.Printf("ingest[%s]: initial connect failed: %v", w.cfg.BrokerName, tok.Error())
// paho will retry; we fall through and wait for ctx
}
<-ctx.Done()
client.Disconnect(500)
w.client.Disconnect(500)
log.Printf("ingest[%s]: stopped", w.cfg.BrokerName)
}
func (w *Worker) BrokerName() string {
return w.cfg.BrokerName
}
func (w *Worker) IsConnected() bool {
if w.client == nil {
return false
}
return w.client.IsConnected()
}
// subscribe registers the wildcard topic handler after (re)connect.
func (w *Worker) subscribe(client mqtt.Client) {
// meshcore/{IATA}/{pubkey}/packets