Explorar el Código

fix: adding graceful shutdown of the api server

master
Blaz Smehov hace 2 días
padre
commit
faa941e4bd
Se han modificado 4 ficheros con 83 adiciones y 9 borrados
  1. +56
    -0
      build/docker-compose.yaml
  2. +18
    -5
      cmd/server/main.go
  3. +7
    -1
      internal/pkg/controller/beacons_controller.go
  4. +2
    -3
      scripts/testAPI.sh

+ 56
- 0
build/docker-compose.yaml Ver fichero

@@ -54,4 +54,60 @@ services:
ports: ports:
- "127.0.0.1:6379:6379" - "127.0.0.1:6379:6379"


presense-decoder:
build:
context: ../
dockerfile: build/package/Dockerfile.decoder
image: presense-decoder
container_name: presense-decoder
environment:
- KAFKA_URL=kafka:29092
depends_on:
- kafka-init
restart: always
presense-server:
build:
context: ../
dockerfile: build/package/Dockerfile.server
image: presense-server
container_name: presense-server
environment:
- VALKEY_URL=valkey:6379
- KAFKA_URL=kafka:29092
ports:
- "127.0.0.1:1902:1902"
depends_on:
- kafka-init
- valkey
restart: always

presense-bridge:
build:
context: ../
dockerfile: build/package/Dockerfile.bridge
image: presense-bridge
container_name: presense-bridge
environment:
- KAFKA_URL=kafka:29092
- MQTT_HOST=192.168.1.101:1883
- MQTT_USERNAME=user
- MQTT_PASSWORD=pass
depends_on:
- kafka-init
restart: always

presense-location:
build:
context: ../
dockerfile: build/package/Dockerfile.location
image: presense-location
container_name: presense-location
environment:
- KAFKA_URL=kafka:29092
depends_on:
- kafka-init
restart: always




+ 18
- 5
cmd/server/main.go Ver fichero

@@ -48,7 +48,7 @@ func main() {
alertsReader := appState.AddKafkaReader(cfg.KafkaURL, "alertbeacons", "gid-alert-serv") alertsReader := appState.AddKafkaReader(cfg.KafkaURL, "alertbeacons", "gid-alert-serv")


client := appState.AddValkeyClient(cfg.ValkeyURL) client := appState.AddValkeyClient(cfg.ValkeyURL)
// Need Lua script to pull all of the beacons in one go on init
fmt.Println("Init of kafka writers and readers done")


chLoc := make(chan model.HTTPLocation, 200) chLoc := make(chan model.HTTPLocation, 200)
chEvents := make(chan model.BeaconEvent, 500) chEvents := make(chan model.BeaconEvent, 500)
@@ -59,6 +59,8 @@ func main() {


r := mux.NewRouter() r := mux.NewRouter()


fmt.Println("new print")
fmt.Println("new print")
// For now just add beacon DELETE / GET / POST / PUT methods // For now just add beacon DELETE / GET / POST / PUT methods
r.HandleFunc("/api/beacons/{beacon_id}", controller.BeaconsDeleteController(writer, ctx, appState)).Methods("DELETE") r.HandleFunc("/api/beacons/{beacon_id}", controller.BeaconsDeleteController(writer, ctx, appState)).Methods("DELETE")
r.HandleFunc("/api/beacons", controller.BeaconsListController(appState)).Methods("GET") r.HandleFunc("/api/beacons", controller.BeaconsListController(appState)).Methods("GET")
@@ -69,9 +71,14 @@ func main() {
r.HandleFunc("/api/settings", controller.SettingsListController(appState, client, ctx)).Methods("GET") r.HandleFunc("/api/settings", controller.SettingsListController(appState, client, ctx)).Methods("GET")
r.HandleFunc("/api/settings", controller.SettingsEditController(settingsWriter, appState, client, ctx)).Methods("POST") r.HandleFunc("/api/settings", controller.SettingsEditController(settingsWriter, appState, client, ctx)).Methods("POST")


r.HandleFunc("/api/beacons/ws", serveWs(appState, ctx))
// r.HandleFunc("/api/beacons/ws", serveWs(appState, ctx))


http.ListenAndServe(cfg.HTTPAddr, handlers.CORS(originsOk, headersOk, methodsOk)(r))
server := http.Server{
Addr: cfg.HTTPAddr,
Handler: handlers.CORS(originsOk, headersOk, methodsOk)(r),
}

go server.ListenAndServe()


eventLoop: eventLoop:
for { for {
@@ -89,7 +96,11 @@ eventLoop:
} }
} }


fmt.Println("broken out of the main event loop")
if err := server.Shutdown(context.Background()); err != nil {
fmt.Printf("could not shutdown: %v\n", err)
}

fmt.Println("broken out of the main event loop and HTTP server shutdown")
wg.Wait() wg.Wait()


fmt.Println("All go routines have stopped, Beggining to close Kafka connections") fmt.Println("All go routines have stopped, Beggining to close Kafka connections")
@@ -151,7 +162,9 @@ func writer(ws *websocket.Conn, appstate *appcontext.AppState, ctx context.Conte
} }


func reader(ws *websocket.Conn) { func reader(ws *websocket.Conn) {
defer ws.Close()
defer func() {
ws.Close()
}()
ws.SetReadLimit(512) ws.SetReadLimit(512)
ws.SetReadDeadline(time.Now().Add((60 * 9) / 10 * time.Second)) ws.SetReadDeadline(time.Now().Add((60 * 9) / 10 * time.Second))
ws.SetPongHandler(func(string) error { ws.SetReadDeadline(time.Now().Add((60 * 9) / 10 * time.Second)); return nil }) ws.SetPongHandler(func(string) error { ws.SetReadDeadline(time.Now().Add((60 * 9) / 10 * time.Second)); return nil })


+ 7
- 1
internal/pkg/controller/beacons_controller.go Ver fichero

@@ -87,24 +87,30 @@ func BeaconsAddController(writer *kafka.Writer, ctx context.Context) http.Handle
decoder := json.NewDecoder(r.Body) decoder := json.NewDecoder(r.Body)
var inBeacon model.Beacon var inBeacon model.Beacon
err := decoder.Decode(&inBeacon) err := decoder.Decode(&inBeacon)
fmt.Printf("hello world\n")


if err != nil { if err != nil {
http.Error(w, err.Error(), 400) http.Error(w, err.Error(), 400)
return return
} }
fmt.Printf("hello world\n")
fmt.Printf("in beacon: %+v\n", inBeacon)


if (len(strings.TrimSpace(inBeacon.Name)) == 0) || (len(strings.TrimSpace(inBeacon.ID)) == 0) { if (len(strings.TrimSpace(inBeacon.Name)) == 0) || (len(strings.TrimSpace(inBeacon.ID)) == 0) {
http.Error(w, "name and beacon_id cannot be blank", 400) http.Error(w, "name and beacon_id cannot be blank", 400)
return return
} }


fmt.Printf("sending POST beacon id: %s message\n", inBeacon.ID)
fmt.Printf("Adding new print here also\n")
// fmt.Printf("sending POST beacon id: %s message\n", inBeacon.ID)


apiUpdate := model.ApiUpdate{ apiUpdate := model.ApiUpdate{
Method: "POST", Method: "POST",
Beacon: inBeacon, Beacon: inBeacon,
} }


fmt.Printf("message: %+v\n", apiUpdate)

if err := sendKafkaMessage(writer, &apiUpdate, ctx); err != nil { if err := sendKafkaMessage(writer, &apiUpdate, ctx); err != nil {
fmt.Println("error in sending Kafka POST message") fmt.Println("error in sending Kafka POST message")
http.Error(w, "Error in sending kafka message", 500) http.Error(w, "Error in sending kafka message", 500)


+ 2
- 3
scripts/testAPI.sh Ver fichero

@@ -8,7 +8,6 @@ curl -s -X POST $URL \
-d '{"Beacon_id":"'"$BEACON_ID"'","Name":"Beacon1","tx_power":-59,"rssi":-70}' -d '{"Beacon_id":"'"$BEACON_ID"'","Name":"Beacon1","tx_power":-59,"rssi":-70}'
echo -e "\n" echo -e "\n"


sleep 1
curl -X GET $URL


echo "GET beacon ID: $BEACON_ID"
curl -X GET $URL/$BEACON_ID
sleep 1

Cargando…
Cancelar
Guardar