|
|
|
@@ -2,31 +2,22 @@ package main |
|
|
|
|
|
|
|
import ( |
|
|
|
"context" |
|
|
|
"encoding/json" |
|
|
|
"fmt" |
|
|
|
"log" |
|
|
|
"net/http" |
|
|
|
"os/signal" |
|
|
|
"strings" |
|
|
|
"sync" |
|
|
|
"syscall" |
|
|
|
"time" |
|
|
|
|
|
|
|
"github.com/AFASystems/presence/internal/pkg/common/appcontext" |
|
|
|
"github.com/AFASystems/presence/internal/pkg/config" |
|
|
|
"github.com/AFASystems/presence/internal/pkg/controller" |
|
|
|
"github.com/AFASystems/presence/internal/pkg/kafkaclient" |
|
|
|
"github.com/AFASystems/presence/internal/pkg/model" |
|
|
|
"github.com/AFASystems/presence/internal/pkg/service" |
|
|
|
"github.com/gorilla/handlers" |
|
|
|
"github.com/gorilla/mux" |
|
|
|
"github.com/gorilla/websocket" |
|
|
|
"github.com/redis/go-redis/v9" |
|
|
|
"github.com/segmentio/kafka-go" |
|
|
|
) |
|
|
|
|
|
|
|
var upgrader = websocket.Upgrader{ |
|
|
|
CheckOrigin: func(r *http.Request) bool { return true }, |
|
|
|
} |
|
|
|
|
|
|
|
var wg sync.WaitGroup |
|
|
|
|
|
|
|
func main() { |
|
|
|
@@ -48,8 +39,8 @@ func main() { |
|
|
|
alertsReader := appState.AddKafkaReader(cfg.KafkaURL, "alertbeacons", "gid-alert-serv") |
|
|
|
|
|
|
|
client := appState.AddValkeyClient(cfg.ValkeyURL) |
|
|
|
// Need Lua script to pull all of the beacons in one go on init |
|
|
|
|
|
|
|
// Separate channel for location change? |
|
|
|
chLoc := make(chan model.HTTPLocation, 200) |
|
|
|
chEvents := make(chan model.BeaconEvent, 500) |
|
|
|
|
|
|
|
@@ -60,16 +51,16 @@ func main() { |
|
|
|
r := mux.NewRouter() |
|
|
|
|
|
|
|
// For now just add beacon DELETE / GET / POST / PUT methods |
|
|
|
r.HandleFunc("/api/beacons/{beacon_id}", beaconsDeleteHandler(writer, ctx)).Methods("DELETE") |
|
|
|
r.HandleFunc("/api/beacons", beaconsListController(appState)).Methods("GET") |
|
|
|
r.HandleFunc("/api/beacons/{beacon_id}", beaconsListSingleController(appState)).Methods("GET") |
|
|
|
r.HandleFunc("/api/beacons", beaconsAddHandler(writer, ctx)).Methods("POST") |
|
|
|
r.HandleFunc("/api/beacons", beaconsAddHandler(writer, ctx)).Methods("PUT") |
|
|
|
r.HandleFunc("/api/beacons/{beacon_id}", controller.BeaconsDeleteController(writer, ctx)).Methods("DELETE") |
|
|
|
r.HandleFunc("/api/beacons", controller.BeaconsListController(appState)).Methods("GET") |
|
|
|
r.HandleFunc("/api/beacons/{beacon_id}", controller.BeaconsListSingleController(appState)).Methods("GET") |
|
|
|
r.HandleFunc("/api/beacons", controller.BeaconsAddController(writer, ctx)).Methods("POST") |
|
|
|
r.HandleFunc("/api/beacons", controller.BeaconsAddController(writer, ctx)).Methods("PUT") |
|
|
|
|
|
|
|
// r.HandleFunc("/api/settings", settingsListHandler(client)).Methods("GET") |
|
|
|
r.HandleFunc("/api/settings", settingsEditHandler(settingsWriter)).Methods("POST") |
|
|
|
r.HandleFunc("/api/settings", controller.SettingsListController(appState, client, ctx)).Methods("GET") |
|
|
|
r.HandleFunc("/api/settings", controller.SettingsEditController(settingsWriter, appState, client, ctx)).Methods("POST") |
|
|
|
|
|
|
|
http.ListenAndServe("0.0.0.0:1902", handlers.CORS(originsOk, headersOk, methodsOk)(r)) |
|
|
|
http.ListenAndServe(cfg.HTTPAddr, handlers.CORS(originsOk, headersOk, methodsOk)(r)) |
|
|
|
|
|
|
|
eventLoop: |
|
|
|
for { |
|
|
|
@@ -77,33 +68,12 @@ eventLoop: |
|
|
|
case <-ctx.Done(): |
|
|
|
break eventLoop |
|
|
|
case msg := <-chLoc: |
|
|
|
beacon, ok := appState.GetBeacon(msg.ID) |
|
|
|
if !ok { |
|
|
|
appState.UpdateBeacon(msg.ID, model.Beacon{ID: msg.ID, Location: msg.Location, Distance: msg.Distance, LastSeen: msg.LastSeen, PreviousConfidentLocation: msg.PreviousConfidentLocation}) |
|
|
|
} else { |
|
|
|
beacon.ID = msg.ID |
|
|
|
beacon.Location = msg.Location |
|
|
|
beacon.Distance = msg.Distance |
|
|
|
beacon.LastSeen = msg.LastSeen |
|
|
|
beacon.PreviousConfidentLocation = msg.PreviousConfidentLocation |
|
|
|
appState.UpdateBeacon(msg.ID, beacon) |
|
|
|
} |
|
|
|
if err := persistBeaconValkey(msg.ID, msg, client, ctx); err != nil { |
|
|
|
fmt.Printf("Error in persisting change location beacon: %v", err) |
|
|
|
if err := service.LocationToBeaconService(msg, appState, client, ctx); err != nil { |
|
|
|
fmt.Printf("Error in writing location change to beacon: %v\n", err) |
|
|
|
} |
|
|
|
case msg := <-chEvents: |
|
|
|
beacon, ok := appState.GetBeacon(msg.ID) |
|
|
|
if !ok { |
|
|
|
appState.UpdateBeacon(msg.ID, model.Beacon{ID: msg.ID, BeaconType: msg.Type, HSBattery: int64(msg.Battery), Event: msg.Event}) |
|
|
|
} else { |
|
|
|
beacon.ID = msg.ID |
|
|
|
beacon.BeaconType = msg.Type |
|
|
|
beacon.HSBattery = int64(msg.Battery) |
|
|
|
beacon.Event = msg.Event |
|
|
|
appState.UpdateBeacon(msg.ID, beacon) |
|
|
|
} |
|
|
|
if err := persistBeaconValkey(msg.ID, msg, client, ctx); err != nil { |
|
|
|
fmt.Printf("Error in persisting change event beacon: %v", err) |
|
|
|
if err := service.EventToBeaconService(msg, appState, client, ctx); err != nil { |
|
|
|
fmt.Printf("Error in writing event change to beacon: %v\n", err) |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
@@ -117,300 +87,3 @@ eventLoop: |
|
|
|
fmt.Println("All kafka clients shutdown, starting shutdown of valkey client") |
|
|
|
appState.CleanValkeyClient() |
|
|
|
} |
|
|
|
|
|
|
|
type RedisHashable interface { |
|
|
|
RedisHashable() (map[string]interface{}, error) |
|
|
|
model.BeaconEvent | model.HTTPLocation |
|
|
|
} |
|
|
|
|
|
|
|
func persistBeaconValkey[T RedisHashable](id string, msg T, client *redis.Client, ctx context.Context) error { |
|
|
|
key := fmt.Sprintf("beacon:%s", id) |
|
|
|
hashM, err := msg.RedisHashable() |
|
|
|
if err != nil { |
|
|
|
fmt.Println("Error in converting location into hashmap for Redis insert: ", err) |
|
|
|
return err |
|
|
|
} |
|
|
|
if err := client.HSet(ctx, key, hashM).Err(); err != nil { |
|
|
|
fmt.Println("Error in persisting set in Redis key: ", key) |
|
|
|
return err |
|
|
|
} |
|
|
|
if err := client.SAdd(ctx, "beacons", key).Err(); err != nil { |
|
|
|
fmt.Println("Error in adding beacon to the beacons list for get all operation: ", err) |
|
|
|
return err |
|
|
|
} |
|
|
|
return nil |
|
|
|
} |
|
|
|
|
|
|
|
// Makes no sense to define service as there is no actual business logic |
|
|
|
func beaconsListSingleController(appstate *appcontext.AppState) http.HandlerFunc { |
|
|
|
return func(w http.ResponseWriter, r *http.Request) { |
|
|
|
vars := mux.Vars(r) |
|
|
|
id := vars["beacon_id"] |
|
|
|
beacon, ok := appstate.GetBeacon(id) |
|
|
|
if !ok { |
|
|
|
w.Header().Set("Content-Type", "application/json") |
|
|
|
w.WriteHeader(http.StatusNotFound) |
|
|
|
json.NewEncoder(w).Encode(map[string]string{"error": "Beacon not found"}) |
|
|
|
return |
|
|
|
} |
|
|
|
|
|
|
|
w.Header().Set("Content-Type", "application/json") |
|
|
|
w.WriteHeader(http.StatusOK) |
|
|
|
json.NewEncoder(w).Encode(beacon) |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
func beaconsListController(appstate *appcontext.AppState) http.HandlerFunc { |
|
|
|
return func(w http.ResponseWriter, r *http.Request) { |
|
|
|
beacons := appstate.GetAllBeacons() |
|
|
|
w.Header().Set("Content-Type", "application/json") |
|
|
|
w.WriteHeader(http.StatusOK) |
|
|
|
json.NewEncoder(w).Encode(beacons) |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
// Probably define value as interface and then reuse this writer in all of the functions |
|
|
|
func sendKafkaMessage(writer *kafka.Writer, value *model.ApiUpdate, ctx context.Context) error { |
|
|
|
valueStr, err := json.Marshal(&value) |
|
|
|
if err != nil { |
|
|
|
fmt.Println("error in encoding: ", err) |
|
|
|
return err |
|
|
|
} |
|
|
|
msg := kafka.Message{ |
|
|
|
Value: valueStr, |
|
|
|
} |
|
|
|
|
|
|
|
if err := writer.WriteMessages(ctx, msg); err != nil { |
|
|
|
fmt.Println("Error in sending kafka message: ", err) |
|
|
|
return err |
|
|
|
} |
|
|
|
|
|
|
|
return nil |
|
|
|
} |
|
|
|
|
|
|
|
func beaconsDeleteHandler(writer *kafka.Writer, ctx context.Context) http.HandlerFunc { |
|
|
|
return func(w http.ResponseWriter, r *http.Request) { |
|
|
|
vars := mux.Vars(r) |
|
|
|
beaconId := vars["beacon_id"] |
|
|
|
apiUpdate := model.ApiUpdate{ |
|
|
|
Method: "DELETE", |
|
|
|
ID: beaconId, |
|
|
|
} |
|
|
|
|
|
|
|
fmt.Println("Sending DELETE message") |
|
|
|
|
|
|
|
if err := sendKafkaMessage(writer, &apiUpdate, ctx); err != nil { |
|
|
|
fmt.Println("error in sending Kafka DELETE message") |
|
|
|
http.Error(w, "Error in sending kafka message", 500) |
|
|
|
return |
|
|
|
} |
|
|
|
|
|
|
|
w.Write([]byte("ok")) |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
func beaconsAddHandler(writer *kafka.Writer, ctx context.Context) http.HandlerFunc { |
|
|
|
return func(w http.ResponseWriter, r *http.Request) { |
|
|
|
decoder := json.NewDecoder(r.Body) |
|
|
|
var inBeacon model.Beacon |
|
|
|
err := decoder.Decode(&inBeacon) |
|
|
|
|
|
|
|
if err != nil { |
|
|
|
http.Error(w, err.Error(), 400) |
|
|
|
return |
|
|
|
} |
|
|
|
|
|
|
|
fmt.Println("sending POST message") |
|
|
|
|
|
|
|
if (len(strings.TrimSpace(inBeacon.Name)) == 0) || (len(strings.TrimSpace(inBeacon.ID)) == 0) { |
|
|
|
http.Error(w, "name and beacon_id cannot be blank", 400) |
|
|
|
return |
|
|
|
} |
|
|
|
|
|
|
|
apiUpdate := model.ApiUpdate{ |
|
|
|
Method: "POST", |
|
|
|
Beacon: inBeacon, |
|
|
|
} |
|
|
|
|
|
|
|
if err := sendKafkaMessage(writer, &apiUpdate, ctx); err != nil { |
|
|
|
fmt.Println("error in sending Kafka POST message") |
|
|
|
http.Error(w, "Error in sending kafka message", 500) |
|
|
|
return |
|
|
|
} |
|
|
|
|
|
|
|
w.Write([]byte("ok")) |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
func settingsEditHandler(writer *kafka.Writer) http.HandlerFunc { |
|
|
|
return func(w http.ResponseWriter, r *http.Request) { |
|
|
|
decoder := json.NewDecoder(r.Body) |
|
|
|
var inSettings model.SettingsVal |
|
|
|
if err := decoder.Decode(&inSettings); err != nil { |
|
|
|
http.Error(w, err.Error(), 400) |
|
|
|
fmt.Println("Error in decoding Settings body: ", err) |
|
|
|
return |
|
|
|
} |
|
|
|
|
|
|
|
if !settingsCheck(inSettings) { |
|
|
|
http.Error(w, "values must be greater than 0", 400) |
|
|
|
fmt.Println("settings values must be greater than 0") |
|
|
|
return |
|
|
|
} |
|
|
|
|
|
|
|
valueStr, err := json.Marshal(&inSettings) |
|
|
|
if err != nil { |
|
|
|
http.Error(w, "Error in encoding settings", 500) |
|
|
|
fmt.Println("Error in encoding settings: ", err) |
|
|
|
return |
|
|
|
} |
|
|
|
|
|
|
|
msg := kafka.Message{ |
|
|
|
Value: valueStr, |
|
|
|
} |
|
|
|
|
|
|
|
if err := writer.WriteMessages(context.Background(), msg); err != nil { |
|
|
|
fmt.Println("error in sending Kafka message") |
|
|
|
http.Error(w, "Error in sending kafka message", 500) |
|
|
|
return |
|
|
|
} |
|
|
|
|
|
|
|
w.Write([]byte("ok")) |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
func settingsCheck(settings model.SettingsVal) bool { |
|
|
|
if settings.LocationConfidence <= 0 || settings.LastSeenThreshold <= 0 || settings.HASendInterval <= 0 { |
|
|
|
return false |
|
|
|
} |
|
|
|
|
|
|
|
return true |
|
|
|
} |
|
|
|
|
|
|
|
func serveWs(client *redis.Client) http.HandlerFunc { |
|
|
|
return func(w http.ResponseWriter, r *http.Request) { |
|
|
|
ws, err := upgrader.Upgrade(w, r, nil) |
|
|
|
if err != nil { |
|
|
|
if _, ok := err.(websocket.HandshakeError); !ok { |
|
|
|
log.Println(err) |
|
|
|
} |
|
|
|
return |
|
|
|
} |
|
|
|
|
|
|
|
go writer(ws, client) |
|
|
|
reader(ws) |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
func writer(ws *websocket.Conn, client *redis.Client) { |
|
|
|
pingTicker := time.NewTicker((60 * time.Second * 9) / 10) |
|
|
|
beaconTicker := time.NewTicker(2 * time.Second) |
|
|
|
defer func() { |
|
|
|
pingTicker.Stop() |
|
|
|
beaconTicker.Stop() |
|
|
|
ws.Close() |
|
|
|
}() |
|
|
|
for { |
|
|
|
select { |
|
|
|
case <-beaconTicker.C: |
|
|
|
httpresults, err := client.Get(context.Background(), "httpresults").Result() |
|
|
|
if err == redis.Nil { |
|
|
|
fmt.Println("no beacons list, starting empty") |
|
|
|
} else if err != nil { |
|
|
|
panic(err) |
|
|
|
} else { |
|
|
|
ws.SetWriteDeadline(time.Now().Add(10 * time.Second)) |
|
|
|
if err := ws.WriteMessage(websocket.TextMessage, []byte(httpresults)); err != nil { |
|
|
|
return |
|
|
|
} |
|
|
|
} |
|
|
|
case <-pingTicker.C: |
|
|
|
ws.SetWriteDeadline(time.Now().Add(10 * time.Second)) |
|
|
|
if err := ws.WriteMessage(websocket.PingMessage, []byte{}); err != nil { |
|
|
|
return |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
func serveLatestBeaconsWs(client *redis.Client) http.HandlerFunc { |
|
|
|
return func(w http.ResponseWriter, r *http.Request) { |
|
|
|
ws, err := upgrader.Upgrade(w, r, nil) |
|
|
|
if err != nil { |
|
|
|
if _, ok := err.(websocket.HandshakeError); !ok { |
|
|
|
log.Println(err) |
|
|
|
} |
|
|
|
return |
|
|
|
} |
|
|
|
|
|
|
|
go latestBeaconWriter(ws, client) |
|
|
|
reader(ws) |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
// This and writer can be refactored in one function |
|
|
|
func latestBeaconWriter(ws *websocket.Conn, client *redis.Client) { |
|
|
|
pingTicker := time.NewTicker((60 * time.Second * 9) / 10) |
|
|
|
beaconTicker := time.NewTicker(2 * time.Second) |
|
|
|
defer func() { |
|
|
|
pingTicker.Stop() |
|
|
|
beaconTicker.Stop() |
|
|
|
ws.Close() |
|
|
|
}() |
|
|
|
for { |
|
|
|
select { |
|
|
|
case <-beaconTicker.C: |
|
|
|
latestbeacons, err := client.Get(context.Background(), "latestbeacons").Result() |
|
|
|
if err == redis.Nil { |
|
|
|
fmt.Println("no beacons list, starting empty") |
|
|
|
} else if err != nil { |
|
|
|
panic(err) |
|
|
|
} else { |
|
|
|
ws.SetWriteDeadline(time.Now().Add(10 * time.Second)) |
|
|
|
if err := ws.WriteMessage(websocket.TextMessage, []byte(latestbeacons)); err != nil { |
|
|
|
return |
|
|
|
} |
|
|
|
} |
|
|
|
case <-pingTicker.C: |
|
|
|
ws.SetWriteDeadline(time.Now().Add(10 * time.Second)) |
|
|
|
if err := ws.WriteMessage(websocket.PingMessage, []byte{}); err != nil { |
|
|
|
return |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
func reader(ws *websocket.Conn) { |
|
|
|
defer ws.Close() |
|
|
|
ws.SetReadLimit(512) |
|
|
|
ws.SetReadDeadline(time.Now().Add(60 * time.Second)) |
|
|
|
ws.SetPongHandler(func(string) error { ws.SetReadDeadline(time.Now().Add(60 * time.Second)); return nil }) |
|
|
|
for { |
|
|
|
_, _, err := ws.ReadMessage() |
|
|
|
if err != nil { |
|
|
|
break |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
func handleConnections(clients map[*websocket.Conn]bool, broadcast chan model.Message) http.HandlerFunc { |
|
|
|
return func(w http.ResponseWriter, r *http.Request) { |
|
|
|
ws, err := upgrader.Upgrade(w, r, nil) |
|
|
|
if err != nil { |
|
|
|
log.Fatal(err) |
|
|
|
} |
|
|
|
defer ws.Close() |
|
|
|
clients[ws] = true |
|
|
|
|
|
|
|
for { |
|
|
|
var msg model.Message |
|
|
|
err := ws.ReadJSON(&msg) |
|
|
|
if err != nil { |
|
|
|
log.Printf("error: %v", err) |
|
|
|
delete(clients, ws) |
|
|
|
break |
|
|
|
} |
|
|
|
broadcast <- msg |
|
|
|
} |
|
|
|
} |
|
|
|
} |