mirror of
https://github.com/hanzoai/amqp.git
synced 2026-10-30 14:57:07 +00:00
zeekayandGitHub
9485cff055
* feat: cloud Mount() per HIP-0106
Adds pkg/amqp.Mount(*zip.App, cloud.Deps) error so the unified Hanzo
Cloud binary can blank-import and register the AMQP -> NATS
JetStream bridge alongside every other Hanzo subsystem.
AMQP is broker-only — the data plane is raw AMQP 0-9-1 TCP and has
no HTTP surface. Mount() exposes /v1/amqp/health (always 200) and
/v1/amqp/readyz (200 only when the listener has bound + NATS is
reachable) on the parent zip.App and spawns the proxy.Proxy
listener in a background goroutine. Failure to reach NATS at boot
is logged as a warning — the binary continues serving health so
the operator can fix the upstream without restart-looping pods.
- pkg/amqp/mount.go: Mount + MountWithConfig + Shutdown
- pkg/amqp/env.go: env getter indirection (lets tests stub)
- pkg/amqp/mount_test.go: health/readyz contract under
DisableListener
- main.go: rewritten as a HIP-0106 thin shim — cloud.LoadConfig +
zip.New + amqp.Mount + app.Listen. Existing proxy package
(proxy.New, proxy.Start, proxy.Shutdown) unchanged.
- schema/amqp.zap: minimal Health ZAP interface. AMQP-protocol
proper stays outside ZAP.
init() registers with cloud.Register("amqp", 30, …).
* refactor: collapse pkg/amqp/ → root for clean import path
Subsystem code lives in package amqp at repo root. Importers now use
`github.com/hanzoai/amqp` (no /pkg/amqp/ nesting). Standalone shim moved
to cmd/amqp/ where applicable.
Per Hanzo Go-stdlib pattern: repo IS the package.
* refactor: flatten import path — github.com/hanzoai/cloud (drop /pkg/cloud)
52 lines
1.3 KiB
Go
52 lines
1.3 KiB
Go
package amqp_test
|
|
|
|
import (
|
|
"encoding/json"
|
|
"io"
|
|
"net/http/httptest"
|
|
"testing"
|
|
|
|
luxlog "github.com/luxfi/log"
|
|
|
|
"github.com/hanzoai/amqp"
|
|
"github.com/hanzoai/cloud"
|
|
"github.com/hanzoai/zip"
|
|
)
|
|
|
|
func TestMount_HealthReadyz(t *testing.T) {
|
|
app := zip.New(zip.Config{Logger: luxlog.New("test")})
|
|
deps := cloud.Deps{Logger: luxlog.New("test")}
|
|
|
|
// DisableListener=true keeps the test from trying to bind 5672 or
|
|
// reach a NATS server. Health stays 200, readyz stays 503.
|
|
if err := amqp.MountWithConfig(app, deps, amqp.MountConfig{
|
|
DisableListener: true,
|
|
}); err != nil {
|
|
t.Fatalf("Mount: %v", err)
|
|
}
|
|
|
|
req := httptest.NewRequest("GET", "/v1/amqp/health", nil)
|
|
resp, err := app.Fiber().Test(req)
|
|
if err != nil {
|
|
t.Fatalf("health test: %v", err)
|
|
}
|
|
if resp.StatusCode != 200 {
|
|
t.Fatalf("health status = %d, want 200", resp.StatusCode)
|
|
}
|
|
body, _ := io.ReadAll(resp.Body)
|
|
var hb map[string]any
|
|
_ = json.Unmarshal(body, &hb)
|
|
if hb["service"] != "amqp" {
|
|
t.Fatalf("health body service = %v, want amqp", hb["service"])
|
|
}
|
|
|
|
req = httptest.NewRequest("GET", "/v1/amqp/readyz", nil)
|
|
resp, err = app.Fiber().Test(req)
|
|
if err != nil {
|
|
t.Fatalf("readyz test: %v", err)
|
|
}
|
|
if resp.StatusCode != 503 {
|
|
t.Fatalf("readyz (no listener) status = %d, want 503", resp.StatusCode)
|
|
}
|
|
}
|