144 lines
3.7 KiB
Go
144 lines
3.7 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"log"
|
|
"net/http"
|
|
"os"
|
|
"strings"
|
|
"sync"
|
|
|
|
"git.tornberg.me/mats/go-redis-inventory/pkg/inventory"
|
|
"github.com/matst80/slask-finder/pkg/index"
|
|
"github.com/matst80/slask-finder/pkg/messaging"
|
|
"github.com/redis/go-redis/v9"
|
|
"github.com/redis/go-redis/v9/maintnotifications"
|
|
|
|
amqp "github.com/rabbitmq/amqp091-go"
|
|
)
|
|
|
|
type Server struct {
|
|
service *inventory.RedisInventoryService
|
|
}
|
|
|
|
func (srv *Server) livezHandler(w http.ResponseWriter, r *http.Request) {
|
|
w.WriteHeader(http.StatusOK)
|
|
w.Write([]byte("OK"))
|
|
}
|
|
|
|
func (srv *Server) readyzHandler(w http.ResponseWriter, r *http.Request) {
|
|
w.WriteHeader(http.StatusOK)
|
|
w.Write([]byte("OK"))
|
|
}
|
|
|
|
func (srv *Server) getInventoryHandler(w http.ResponseWriter, r *http.Request) {
|
|
// Parse path: /inventory/{sku}/{location}
|
|
path := r.URL.Path
|
|
parts := strings.Split(strings.Trim(path, "/"), "/")
|
|
if len(parts) != 3 || parts[0] != "inventory" {
|
|
http.Error(w, "Invalid path", http.StatusBadRequest)
|
|
return
|
|
}
|
|
sku := inventory.SKU(parts[1])
|
|
locationID := inventory.LocationID(parts[2])
|
|
|
|
quantity, err := srv.service.GetInventory(sku, locationID)
|
|
if err != nil {
|
|
http.Error(w, err.Error(), http.StatusInternalServerError)
|
|
return
|
|
}
|
|
|
|
response := map[string]int64{"quantity": quantity}
|
|
w.Header().Set("Content-Type", "application/json")
|
|
json.NewEncoder(w).Encode(response)
|
|
}
|
|
|
|
var country = "se"
|
|
var redisAddress = "10.10.3.18:6379"
|
|
var redisPassword = "slaskredis"
|
|
|
|
func init() {
|
|
// Override redis config from environment variables if set
|
|
if addr, ok := os.LookupEnv("REDIS_ADDRESS"); ok {
|
|
redisAddress = addr
|
|
}
|
|
if password, ok := os.LookupEnv("REDIS_PASSWORD"); ok {
|
|
redisPassword = password
|
|
}
|
|
if ctry, ok := os.LookupEnv("COUNTRY"); ok {
|
|
country = ctry
|
|
}
|
|
}
|
|
|
|
func main() {
|
|
var ctx = context.Background()
|
|
rdb := redis.NewClient(&redis.Options{
|
|
Addr: redisAddress,
|
|
Password: redisPassword, // no password set
|
|
DB: 0, // use default DB
|
|
MaintNotificationsConfig: &maintnotifications.Config{
|
|
Mode: maintnotifications.ModeDisabled,
|
|
},
|
|
})
|
|
s, err := inventory.NewRedisInventoryService(rdb, ctx)
|
|
if err != nil {
|
|
log.Fatalf("Unable to connect to inventory redis: %v", err)
|
|
return
|
|
}
|
|
|
|
server := &Server{service: s}
|
|
|
|
// Set up HTTP routes
|
|
http.HandleFunc("/livez", server.livezHandler)
|
|
http.HandleFunc("/readyz", server.readyzHandler)
|
|
http.HandleFunc("/inventory/", server.getInventoryHandler)
|
|
|
|
stockhandler := &StockHandler{
|
|
MainStockLocationID: inventory.LocationID(country),
|
|
rdb: rdb,
|
|
ctx: ctx,
|
|
svc: *s,
|
|
}
|
|
|
|
amqpUrl, ok := os.LookupEnv("RABBIT_HOST")
|
|
if ok {
|
|
log.Printf("Connecting to rabbitmq")
|
|
conn, err := amqp.DialConfig(amqpUrl, amqp.Config{
|
|
Properties: amqp.NewConnectionProperties(),
|
|
})
|
|
//a.conn = conn
|
|
if err != nil {
|
|
log.Fatalf("Failed to connect to RabbitMQ: %v", err)
|
|
}
|
|
ch, err := conn.Channel()
|
|
if err != nil {
|
|
log.Fatalf("Failed to open a channel: %v", err)
|
|
}
|
|
// items listener
|
|
err = messaging.ListenToTopic(ch, country, "item_added", func(d amqp.Delivery) error {
|
|
var items []*index.DataItem
|
|
err := json.Unmarshal(d.Body, &items)
|
|
if err == nil {
|
|
log.Printf("Got upserts %d, message count %d", len(items), d.MessageCount)
|
|
wg := &sync.WaitGroup{}
|
|
for _, item := range items {
|
|
stockhandler.HandleItem(item, wg)
|
|
}
|
|
wg.Wait()
|
|
log.Print("Batch done...")
|
|
} else {
|
|
log.Printf("Failed to unmarshal upsert message %v", err)
|
|
}
|
|
return err
|
|
})
|
|
if err != nil {
|
|
log.Fatalf("Failed to listen to item_added topic: %v", err)
|
|
}
|
|
}
|
|
|
|
// Start HTTP server
|
|
log.Println("Starting HTTP server on :8080")
|
|
log.Fatal(http.ListenAndServe(":8080", nil))
|
|
}
|