1 parent
5237c8ab85
commit
4941840504
3 files changed
+42
-9
No files matched your search
@@ -2,4 +2,9 @@ module git.k6n.net/mats/go-order-manager
|
|||||||
|
|
||||||
go 1.26.2
|
go 1.26.2
|
||||||
|
|
||||||
require github.com/rabbitmq/amqp091-go v1.10.0
|
require (
|
||||||
|
git.k6n.net/mats/platform v0.0.0
|
||||||
|
github.com/rabbitmq/amqp091-go v1.12.0
|
||||||
|
)
|
||||||
|
|
||||||
|
replace git.k6n.net/mats/platform => ../platform
|
||||||
@@ -1,4 +1,4 @@
|
|||||||
github.com/rabbitmq/amqp091-go v1.10.0 h1:STpn5XsHlHGcecLmMFCtg7mqq0RnD+zFr4uzukfVhBw=
|
github.com/rabbitmq/amqp091-go v1.12.0 h1:V0v14Iqfs+MwHWihJt/nGS5Ulu0vw572b2Co3mwunkI=
|
||||||
github.com/rabbitmq/amqp091-go v1.10.0/go.mod h1:Hy4jKW5kQART1u+JkDTF9YYOQUHXqMuhrgxOEeS7G4o=
|
github.com/rabbitmq/amqp091-go v1.12.0/go.mod h1:Hy4jKW5kQART1u+JkDTF9YYOQUHXqMuhrgxOEeS7G4o=
|
||||||
go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
|
go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
|
||||||
go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
|
go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
|
||||||
@@ -2,6 +2,7 @@ package main
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"compress/gzip"
|
"compress/gzip"
|
||||||
|
"context"
|
||||||
"encoding/gob"
|
"encoding/gob"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
"log"
|
"log"
|
||||||
@@ -9,6 +10,9 @@ import (
|
|||||||
"os"
|
"os"
|
||||||
"runtime"
|
"runtime"
|
||||||
"sync"
|
"sync"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"git.k6n.net/mats/platform/httpserver"
|
||||||
)
|
)
|
||||||
|
|
||||||
var amqpUrl = os.Getenv("AMQP_URL")
|
var amqpUrl = os.Getenv("AMQP_URL")
|
||||||
@@ -135,16 +139,17 @@ func main() {
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
log.Fatalf("Failed to connect to RabbitMQ: %v", err)
|
log.Fatalf("Failed to connect to RabbitMQ: %v", err)
|
||||||
}
|
}
|
||||||
defer client.Close()
|
// Note: client.Close() is called via a shutdown hook in
|
||||||
|
// httpserver.RunServerWithShutdown below — NOT via defer, so it runs
|
||||||
|
// BEFORE the HTTP server shuts down (stopping new AMQP deliveries
|
||||||
|
// before we stop accepting HTTP requests).
|
||||||
|
|
||||||
// Routes are served at their CANONICAL public paths under /api/orders, so
|
// Routes are served at their CANONICAL public paths under /api/orders, so
|
||||||
// ingress routes straight to this service with no rewrite/strip — you can
|
// ingress routes straight to this service with no rewrite/strip — you can
|
||||||
// curl the pod or the ingress and the path is identical (deploy/API-ROUTING.md).
|
// curl the pod or the ingress and the path is identical (deploy/API-ROUTING.md).
|
||||||
// /healthz stays pod-local for probes (never exposed at the edge).
|
// /healthz stays pod-local for probes (never exposed at the edge).
|
||||||
mux := http.NewServeMux()
|
mux := http.NewServeMux()
|
||||||
mux.HandleFunc("GET /healthz", func(w http.ResponseWriter, r *http.Request) {
|
httpserver.RegisterProbes(mux, nil)
|
||||||
w.WriteHeader(http.StatusOK)
|
|
||||||
})
|
|
||||||
mux.HandleFunc("GET /api/orders", func(w http.ResponseWriter, r *http.Request) {
|
mux.HandleFunc("GET /api/orders", func(w http.ResponseWriter, r *http.Request) {
|
||||||
w.Header().Set("Content-Type", "application/json")
|
w.Header().Set("Content-Type", "application/json")
|
||||||
w.WriteHeader(http.StatusOK)
|
w.WriteHeader(http.StatusOK)
|
||||||
@@ -169,7 +174,30 @@ func main() {
|
|||||||
w.WriteHeader(http.StatusNotImplemented)
|
w.WriteHeader(http.StatusNotImplemented)
|
||||||
w.Write([]byte(order_id))
|
w.Write([]byte(order_id))
|
||||||
})
|
})
|
||||||
if err := http.ListenAndServe(":8080", mux); err != nil {
|
|
||||||
log.Fatalf("Failed to start server: %v", err)
|
// Graceful shutdown: flush pending order saves + close the AMQP connection
|
||||||
|
// before the server stops. The previous code used a bare http.ListenAndServe
|
||||||
|
// with no signal handling — a SIGTERM killed the process mid-save, risking
|
||||||
|
// a corrupt .dbz file or a dropped AMQP ack. RunServerWithShutdown handles
|
||||||
|
// SIGINT/SIGTERM, runs the save hook, then shuts down the HTTP server.
|
||||||
|
srv := &http.Server{
|
||||||
|
Addr: ":8080",
|
||||||
|
Handler: mux,
|
||||||
|
ReadHeaderTimeout: 10 * time.Second,
|
||||||
|
}
|
||||||
|
if err := httpserver.RunServerWithShutdown(srv, "order-manager", 10*time.Second, 5*time.Second,
|
||||||
|
func(context.Context) error {
|
||||||
|
// Flush the in-memory order map to disk so a SIGTERM doesn't lose
|
||||||
|
// orders that were placed since the last async Save().
|
||||||
|
return handler.Save()
|
||||||
|
},
|
||||||
|
func(context.Context) error {
|
||||||
|
// Close the AMQP connection cleanly so in-flight deliveries are
|
||||||
|
// requeued, not lost.
|
||||||
|
client.Close()
|
||||||
|
return nil
|
||||||
|
},
|
||||||
|
); err != nil {
|
||||||
|
log.Fatalf("server: %v", err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
Reference in new issue
Block a user