develop #1
@ -18,6 +18,7 @@ waitForService "postgres-db:5432"
|
||||
waitForService "api-eventbus:5672"
|
||||
waitForService "api-logger:24224"
|
||||
waitForService "api-registry:8500"
|
||||
waitForService "basket-svc:80"
|
||||
|
||||
# run migrations
|
||||
migrate.sh
|
||||
|
||||
@ -14,7 +14,7 @@ import (
|
||||
|
||||
const (
|
||||
defAppName = "order-svc-migrations"
|
||||
defMigrationsTableName = `"order".migrations`
|
||||
defMigrationsTableName = "ordering.migrations"
|
||||
defLoggerAddr = "api-logger:24224"
|
||||
// defKVNmspc = "dev.egommerce/service/order-migration"
|
||||
)
|
||||
|
||||
@ -10,15 +10,19 @@ import (
|
||||
"git.pbiernat.dev/egommerce/order-service/internal/app/config"
|
||||
"git.pbiernat.dev/egommerce/order-service/internal/app/database"
|
||||
"git.pbiernat.dev/egommerce/order-service/internal/app/server"
|
||||
"github.com/go-redis/redis/v8"
|
||||
)
|
||||
|
||||
const (
|
||||
defAppName = "order-svc"
|
||||
defAppDomain = "order-svc"
|
||||
defPathPrefix = "/order"
|
||||
defNetAddr = ":80"
|
||||
defLoggerAddr = "api-logger:24224"
|
||||
defRegistryAddr = "api-registry:8500"
|
||||
defDbURL = "postgres://postgres:12345678@postgres-db:5432/egommerce"
|
||||
defCacheAddr = "api-cache:6379"
|
||||
defCachePassword = "12345678"
|
||||
defMongoDbURL = "mongodb://mongodb:12345678@mongo-db:27017"
|
||||
defEventBusURL = "amqp://guest:guest@api-eventbus:5672"
|
||||
ebEventsExchange = "api-events"
|
||||
@ -35,11 +39,14 @@ func main() {
|
||||
c.AppID, _ = os.Hostname()
|
||||
c.AppName = config.GetEnv("APP_NAME", defAppName)
|
||||
c.AppDomain = config.GetEnv("APP_DOMAIN", defAppDomain)
|
||||
c.PathPrefix = config.GetEnv("APP_PATH_PREFIX", defPathPrefix)
|
||||
c.NetAddr = config.GetEnv("SERVER_ADDR", defNetAddr)
|
||||
c.Port, _ = strconv.Atoi(c.NetAddr[1:])
|
||||
c.LoggerAddr = config.GetEnv("LOGGER_ADDR", defLoggerAddr)
|
||||
c.RegistryAddr = config.GetEnv("REGISTRY_ADDR", defRegistryAddr)
|
||||
c.DbURL = config.GetEnv("DATABASE_URL", defDbURL)
|
||||
c.CacheAddr = config.GetEnv("CACHE_ADDR", defCacheAddr)
|
||||
c.CachePassword = config.GetEnv("CACHE_PASSWORD", defCachePassword)
|
||||
c.EventBusURL = config.GetEnv("EVENTBUS_URL", defEventBusURL)
|
||||
c.EventBusExchange = ebEventsExchange
|
||||
c.KVNamespace = config.GetEnv("APP_KV_NAMESPACE", defKVNmspc)
|
||||
@ -56,6 +63,14 @@ func main() {
|
||||
}
|
||||
defer dbConn.Close()
|
||||
|
||||
// redis conn
|
||||
redis := redis.NewClient(&redis.Options{
|
||||
Addr: c.CacheAddr,
|
||||
Password: c.CachePassword,
|
||||
DB: 0,
|
||||
})
|
||||
defer redis.Close()
|
||||
|
||||
// eventbus conn
|
||||
ebConn, ebCh, err := amqp.Open(c.EventBusURL)
|
||||
if err != nil {
|
||||
@ -72,7 +87,7 @@ func main() {
|
||||
}
|
||||
|
||||
// start server
|
||||
srv := server.NewServer(c, logger, dbConn, ebCh)
|
||||
srv := server.NewServer(c, logger, dbConn, redis, ebCh)
|
||||
|
||||
forever := make(chan struct{})
|
||||
srv.StartWithGracefulShutdown(forever)
|
||||
|
||||
@ -14,11 +14,15 @@ import (
|
||||
|
||||
discovery "git.pbiernat.dev/egommerce/go-api-pkg/consul"
|
||||
"git.pbiernat.dev/egommerce/go-api-pkg/fluentd"
|
||||
amqp "git.pbiernat.dev/egommerce/go-api-pkg/rabbitmq"
|
||||
"git.pbiernat.dev/egommerce/go-api-pkg/rabbitmq"
|
||||
"git.pbiernat.dev/egommerce/order-service/internal/app/config"
|
||||
"git.pbiernat.dev/egommerce/order-service/internal/app/database"
|
||||
"git.pbiernat.dev/egommerce/order-service/internal/app/event"
|
||||
"git.pbiernat.dev/egommerce/order-service/internal/app/server"
|
||||
"git.pbiernat.dev/egommerce/order-service/internal/app/service"
|
||||
"git.pbiernat.dev/egommerce/order-service/internal/app/ui"
|
||||
"github.com/go-redis/redis/v8"
|
||||
"github.com/streadway/amqp"
|
||||
)
|
||||
|
||||
const (
|
||||
@ -26,6 +30,8 @@ const (
|
||||
defLoggerAddr = "api-logger:24224"
|
||||
defRegistryAddr = "api-registry:8500"
|
||||
defDbURL = "postgres://postgres:12345678@postgres-db:5432/egommerce"
|
||||
defCacheAddr = "api-cache:6379"
|
||||
defCachePassword = "12345678"
|
||||
defMongoDbURL = "mongodb://mongodb:12345678@mongo-db:27017"
|
||||
defEventBusURL = "amqp://guest:guest@api-eventbus:5672"
|
||||
ebEventsExchange = "api-events"
|
||||
@ -44,6 +50,8 @@ func main() {
|
||||
c.LoggerAddr = config.GetEnv("LOGGER_ADDR", defLoggerAddr)
|
||||
c.RegistryAddr = config.GetEnv("REGISTRY_ADDR", defRegistryAddr)
|
||||
c.DbURL = config.GetEnv("DATABASE_URL", defDbURL)
|
||||
c.CacheAddr = config.GetEnv("CACHE_ADDR", defCacheAddr)
|
||||
c.CachePassword = config.GetEnv("CACHE_PASSWORD", defCachePassword)
|
||||
c.EventBusURL = config.GetEnv("EVENTBUS_URL", defEventBusURL)
|
||||
c.EventBusExchange = ebEventsExchange
|
||||
c.EventBusQueue = ebEventsQueue
|
||||
@ -53,7 +61,7 @@ func main() {
|
||||
logger := fluentd.NewLogger(c.GetAppFullName(), logHost, logPort)
|
||||
defer logger.Close()
|
||||
|
||||
consul, err := discovery.NewService(c.RegistryAddr, c.AppID, c.AppName, c.AppID, "", 0)
|
||||
consul, err := discovery.NewService(c.RegistryAddr, c.AppID, c.AppName, c.AppID, "", "", 0)
|
||||
if err != nil {
|
||||
logger.Log("Error connecting to %s: %v", c.RegistryAddr, err)
|
||||
}
|
||||
@ -74,16 +82,24 @@ func main() {
|
||||
}
|
||||
defer dbConn.Close()
|
||||
|
||||
// redis conn
|
||||
redis := redis.NewClient(&redis.Options{
|
||||
Addr: c.CacheAddr,
|
||||
Password: c.CachePassword,
|
||||
DB: 0,
|
||||
})
|
||||
defer redis.Close()
|
||||
|
||||
// eventbus conn
|
||||
ebConn, ebCh, err := amqp.Open(c.EventBusURL)
|
||||
ebConn, ebCh, err := rabbitmq.Open(c.EventBusURL)
|
||||
if err != nil {
|
||||
logger.Log("Failed to connect to EventBus server: %v\n", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
defer ebCh.Close()
|
||||
defer amqp.Close(ebConn)
|
||||
defer rabbitmq.Close(ebConn)
|
||||
|
||||
err = amqp.NewExchange(ebCh, c.EventBusExchange)
|
||||
err = rabbitmq.NewExchange(ebCh, c.EventBusExchange)
|
||||
if err != nil {
|
||||
logger.Log("Failed to declare EventBus exchange: %v\n", err)
|
||||
os.Exit(1)
|
||||
@ -103,7 +119,7 @@ func main() {
|
||||
os.Exit(1)
|
||||
}
|
||||
|
||||
amqp.BindQueueToExchange(ebCh, c.EventBusQueue, c.EventBusExchange, "basket.order.basketCheckout")
|
||||
rabbitmq.BindQueueToExchange(ebCh, c.EventBusQueue, c.EventBusExchange, "basket.order.basketCheckout")
|
||||
// if err != nil {
|
||||
// logger.Log("Failed to prepare EventBus queue: %v\n", err)
|
||||
// os.Exit(1)
|
||||
@ -136,26 +152,39 @@ func main() {
|
||||
}()
|
||||
|
||||
go func() {
|
||||
orderSrv := service.NewOrderService(dbConn, redis, ebCh, logger)
|
||||
|
||||
for d := range msgs {
|
||||
msg, err := amqp.Deserialize(d.Body)
|
||||
go func(d amqp.Delivery) {
|
||||
msg, err := rabbitmq.Deserialize(d.Body)
|
||||
if err != nil {
|
||||
logger.Log("json error: %v\n", err)
|
||||
logger.Log("deserialize error: %v\n", err)
|
||||
d.Reject(false) // FIXME: how to handle erros in queue...????
|
||||
continue
|
||||
return
|
||||
}
|
||||
|
||||
eName := fmt.Sprintf("%s", msg["event"])
|
||||
data := (msg["data"]).(map[string]interface{})
|
||||
logger.Log("Message<%s>: %s\n", eName, data)
|
||||
|
||||
reqID := data["request_id"].(string) // FIXME Check input params!
|
||||
|
||||
switch true {
|
||||
case strings.Contains(eName, event.EVENT_BASKET_CHECKOUT):
|
||||
basketID := data["basket_id"].(string) // FIXME Check input params!
|
||||
// create new order based on basket
|
||||
logger.Log("Event: %s", event.EVENT_BASKET_CHECKOUT)
|
||||
order, err := ui.CreateOrder(orderSrv, basketID, reqID)
|
||||
if err != nil {
|
||||
logger.Log("%s error: %v", event.EVENT_BASKET_CHECKOUT, err)
|
||||
d.Reject(false) // FIXME: or Nack? how to handle erros in queue...
|
||||
break
|
||||
}
|
||||
logger.Log("Event: %s. Created Order: %v", event.EVENT_BASKET_CHECKOUT, order)
|
||||
}
|
||||
|
||||
logger.Log("ACK: %s", eName)
|
||||
d.Ack(false)
|
||||
}(d)
|
||||
}
|
||||
}()
|
||||
|
||||
|
||||
@ -3,9 +3,11 @@ module git.pbiernat.dev/egommerce/order-service
|
||||
go 1.18
|
||||
|
||||
require (
|
||||
git.pbiernat.dev/egommerce/go-api-pkg v0.0.113
|
||||
git.pbiernat.dev/egommerce/api-entities v0.0.26
|
||||
git.pbiernat.dev/egommerce/go-api-pkg v0.0.136
|
||||
github.com/go-pg/migrations/v8 v8.1.0
|
||||
github.com/go-pg/pg/v10 v10.10.7
|
||||
github.com/go-redis/redis/v8 v8.11.5
|
||||
github.com/gofiber/fiber/v2 v2.40.1
|
||||
github.com/jackc/pgx/v4 v4.17.2
|
||||
github.com/joho/godotenv v1.4.0
|
||||
@ -15,6 +17,8 @@ require (
|
||||
require (
|
||||
github.com/andybalholm/brotli v1.0.4 // indirect
|
||||
github.com/armon/go-metrics v0.4.1 // indirect
|
||||
github.com/cespare/xxhash/v2 v2.1.2 // indirect
|
||||
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
|
||||
github.com/fatih/color v1.13.0 // indirect
|
||||
github.com/fluent/fluent-logger-golang v1.9.0 // indirect
|
||||
github.com/go-pg/zerochecker v0.2.0 // indirect
|
||||
@ -31,7 +35,7 @@ require (
|
||||
github.com/jackc/pgpassfile v1.0.0 // indirect
|
||||
github.com/jackc/pgproto3/v2 v2.3.1 // indirect
|
||||
github.com/jackc/pgservicefile v0.0.0-20200714003250-2b9c44734f2b // indirect
|
||||
github.com/jackc/pgtype v1.12.0 // indirect
|
||||
github.com/jackc/pgtype v1.13.0 // indirect
|
||||
github.com/jackc/puddle v1.3.0 // indirect
|
||||
github.com/jinzhu/inflection v1.0.0 // indirect
|
||||
github.com/klauspost/compress v1.15.9 // indirect
|
||||
|
||||
25
src/go.sum
25
src/go.sum
@ -1,6 +1,10 @@
|
||||
cloud.google.com/go v0.26.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=
|
||||
git.pbiernat.dev/egommerce/go-api-pkg v0.0.113 h1:kf7HesezhXIAMNYgLCm8x6YVqyLqJRqaPKIFEXf4xSs=
|
||||
git.pbiernat.dev/egommerce/go-api-pkg v0.0.113/go.mod h1:nAwcw2MZtn/54YKq8VQK6RJAsiuoLUtPuazXg8JcqK8=
|
||||
git.pbiernat.dev/egommerce/api-entities v0.0.26 h1:Avz02GINwuYWOjw1fmZIJ3QgGEIz3a5vRQZNaxxUQIk=
|
||||
git.pbiernat.dev/egommerce/api-entities v0.0.26/go.mod h1:+BXvUcr6Cr6QNpJsW8BUfe1vVILdWDADNE0e3u0lNvU=
|
||||
git.pbiernat.dev/egommerce/go-api-pkg v0.0.135 h1:qOa6MB6d2/lr0t9c3WWP84rf/T57PNYgizTmuNCDws8=
|
||||
git.pbiernat.dev/egommerce/go-api-pkg v0.0.135/go.mod h1:w2N79aoumjrrcrGPJLkCwxAHtrLd7G4Uj8VOxvPooa0=
|
||||
git.pbiernat.dev/egommerce/go-api-pkg v0.0.136 h1:SzJRAkqJKdng/3d0V7o/R0yGh7QaZynPBn/P++on9RA=
|
||||
git.pbiernat.dev/egommerce/go-api-pkg v0.0.136/go.mod h1:w2N79aoumjrrcrGPJLkCwxAHtrLd7G4Uj8VOxvPooa0=
|
||||
github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
|
||||
github.com/DataDog/datadog-go v3.2.0+incompatible/go.mod h1:LButxg5PwREeZtORoXG3tL4fMGNddJ+vMq1mwgfaqoQ=
|
||||
github.com/Masterminds/semver/v3 v3.1.1/go.mod h1:VPu/7SZ7ePZ3QOrcuXROw5FAcLl4a0cBrbBpGY/8hQs=
|
||||
@ -23,6 +27,8 @@ github.com/bgentry/speakeasy v0.1.0/go.mod h1:+zsyZBPWlz7T6j88CTgSN5bM796AkVf0kB
|
||||
github.com/bmizerany/assert v0.0.0-20160611221934-b7ed37b82869 h1:DDGfHa7BWjL4YnC6+E63dPcxHo2sUxDIu8g3QgEJdRY=
|
||||
github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
|
||||
github.com/cespare/xxhash/v2 v2.1.1/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
|
||||
github.com/cespare/xxhash/v2 v2.1.2 h1:YRXhKfTDauu4ajMg1TPgFO5jnlC2HCbmLXMcTG5cbYE=
|
||||
github.com/cespare/xxhash/v2 v2.1.2/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
|
||||
github.com/circonus-labs/circonus-gometrics v2.3.1+incompatible/go.mod h1:nmEj6Dob7S7YxXgwXpfOuvO54S+tGdZdw9fuRZt25Ag=
|
||||
github.com/circonus-labs/circonusllhist v0.1.3/go.mod h1:kMXHVDlOchFAehlya5ePtbp5jckzBHf4XRpQvBOLI+I=
|
||||
github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw=
|
||||
@ -35,6 +41,8 @@ github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ3
|
||||
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
||||
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
|
||||
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
||||
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78=
|
||||
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc=
|
||||
github.com/envoyproxy/go-control-plane v0.9.1-0.20191026205805-5f8ba28d4473/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
|
||||
github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c=
|
||||
github.com/fatih/color v1.7.0/go.mod h1:Zm6kSWBoL9eyXnKyktHP6abPY2pDugNf5KwzbycvMj4=
|
||||
@ -59,6 +67,8 @@ github.com/go-pg/pg/v10 v10.10.7 h1:Q7Bs45kP9MIg03v/ejwdqsPd1T0cecgeDoTJVg/UJuQ=
|
||||
github.com/go-pg/pg/v10 v10.10.7/go.mod h1:GLmFXufrElQHf5uzM3BQlcfwV3nsgnHue5uzjQ6Nqxg=
|
||||
github.com/go-pg/zerochecker v0.2.0 h1:pp7f72c3DobMWOb2ErtZsnrPaSvHd2W4o9//8HtF4mU=
|
||||
github.com/go-pg/zerochecker v0.2.0/go.mod h1:NJZ4wKL0NmTtz0GKCoJ8kym6Xn/EQzXRl2OnAe7MmDo=
|
||||
github.com/go-redis/redis/v8 v8.11.5 h1:AcZZR7igkdvfVmQTPnu9WE37LRrO/YrBH5zWyjDC0oI=
|
||||
github.com/go-redis/redis/v8 v8.11.5/go.mod h1:gREzHqY1hg6oD9ngVRbLStwAWKhA0FEgq8Jd4h5lpwo=
|
||||
github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY=
|
||||
github.com/gofiber/fiber/v2 v2.40.1 h1:pc7n9VVpGIqNsvg9IPLQhyFEMJL8gCs1kneH5D1pIl4=
|
||||
github.com/gofiber/fiber/v2 v2.40.1/go.mod h1:Gko04sLksnHbzLSRBFWPFdzM9Ws9pRxvvIaohJK1dsk=
|
||||
@ -171,8 +181,9 @@ github.com/jackc/pgtype v0.0.0-20190421001408-4ed0de4755e0/go.mod h1:hdSHsc1V01C
|
||||
github.com/jackc/pgtype v0.0.0-20190824184912-ab885b375b90/go.mod h1:KcahbBH1nCMSo2DXpzsoWOAfFkdEtEJpPbVLq8eE+mc=
|
||||
github.com/jackc/pgtype v0.0.0-20190828014616-a8802b16cc59/go.mod h1:MWlu30kVJrUS8lot6TQqcg7mtthZ9T0EoIBFiJcmcyw=
|
||||
github.com/jackc/pgtype v1.8.1-0.20210724151600-32e20a603178/go.mod h1:C516IlIV9NKqfsMCXTdChteoXmwgUceqaLfjg2e3NlM=
|
||||
github.com/jackc/pgtype v1.12.0 h1:Dlq8Qvcch7kiehm8wPGIW0W3KsCCHJnRacKW0UM8n5w=
|
||||
github.com/jackc/pgtype v1.12.0/go.mod h1:LUMuVrfsFfdKGLw+AFFVv6KtHOFMwRgDDzBt76IqCA4=
|
||||
github.com/jackc/pgtype v1.13.0 h1:XkIc7A+1BmZD19bB2NxrtjJweHxQ9agqvM+9URc68Cg=
|
||||
github.com/jackc/pgtype v1.13.0/go.mod h1:LUMuVrfsFfdKGLw+AFFVv6KtHOFMwRgDDzBt76IqCA4=
|
||||
github.com/jackc/pgx/v4 v4.0.0-20190420224344-cc3461e65d96/go.mod h1:mdxmSJJuR08CZQyj1PVQBHy9XOp5p8/SHH6a0psbY9Y=
|
||||
github.com/jackc/pgx/v4 v4.0.0-20190421002000-1b8f0016e912/go.mod h1:no/Y67Jkk/9WuGR0JG/JseM9irFbnEPbuWV2EELPNuM=
|
||||
github.com/jackc/pgx/v4 v4.0.0-pre1.0.20190824185557-6972a5742186/go.mod h1:X+GQnOEnf1dqHGpw7JmHqHc1NxDoalibchSk9/RWuDc=
|
||||
@ -251,16 +262,16 @@ github.com/modern-go/reflect2 v1.0.1/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3Rllmb
|
||||
github.com/mwitkow/go-conntrack v0.0.0-20161129095857-cc309e4a2223/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U=
|
||||
github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e h1:fD57ERR4JtEqsWbfPhv4DMiApHyliiK5xCTNVSPiaAs=
|
||||
github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e/go.mod h1:zD1mROLANZcx1PVRCS0qkT7pwLkGfwJo4zjcN/Tysno=
|
||||
github.com/nxadm/tail v1.4.4 h1:DQuhQpB1tVlglWS2hLQ5OV6B5r8aGxSrPc5Qo6uTN78=
|
||||
github.com/nxadm/tail v1.4.4/go.mod h1:kenIhsEOeOJmVchQTgglprH7qJGnHDVpk1VPCcaMI8A=
|
||||
github.com/nxadm/tail v1.4.8 h1:nPr65rt6Y5JFSKQO7qToXr7pePgD6Gwiw05lkbyAQTE=
|
||||
github.com/onsi/ginkgo v1.6.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
|
||||
github.com/onsi/ginkgo v1.12.1/go.mod h1:zj2OWP4+oCPe1qIXoGWkgMRwljMUYCdkwsT2108oapk=
|
||||
github.com/onsi/ginkgo v1.14.2 h1:8mVmC9kjFFmA8H4pKMUhcblgifdkOIXPvbhN1T36q1M=
|
||||
github.com/onsi/ginkgo v1.14.2/go.mod h1:iSB4RoI2tjJc9BBv4NKIKWKya62Rps+oPG/Lv9klQyY=
|
||||
github.com/onsi/ginkgo v1.16.5 h1:8xi0RTUf59SOSfEtZMvwTvXYMzG4gV23XVHOZiXNtnE=
|
||||
github.com/onsi/gomega v1.7.1/go.mod h1:XdKZgCCFLUoM/7CFJVPcG8C1xQ1AJ0vpAezJrB7JYyY=
|
||||
github.com/onsi/gomega v1.10.1/go.mod h1:iN09h71vgCQne3DLsj+A5owkum+a2tYe+TOCB1ybHNo=
|
||||
github.com/onsi/gomega v1.10.3 h1:gph6h/qe9GSUw1NhH1gp+qb+h8rXD8Cy60Z32Qw3ELA=
|
||||
github.com/onsi/gomega v1.10.3/go.mod h1:V9xEwhxec5O8UDM77eCW8vLymOMltsqPVYWrpDsH8xc=
|
||||
github.com/onsi/gomega v1.18.1 h1:M1GfJqGRrBrrGGsbxzV5dqM2U2ApXefZCQpkukxYRLE=
|
||||
github.com/pascaldekloe/goe v0.0.0-20180627143212-57f6aae5913c/go.mod h1:lzWF7FIEvWOWxwDKqyGYQf6ZUaNfKdP144TG7ZOy1lc=
|
||||
github.com/pascaldekloe/goe v0.1.0 h1:cBOtyMzM9HTpWjXfbbunk26uA6nG3a8n06Wieeh0MwY=
|
||||
github.com/pascaldekloe/goe v0.1.0/go.mod h1:lzWF7FIEvWOWxwDKqyGYQf6ZUaNfKdP144TG7ZOy1lc=
|
||||
@ -520,8 +531,8 @@ gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
|
||||
gopkg.in/yaml.v2 v2.2.4/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
|
||||
gopkg.in/yaml.v2 v2.2.5/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
|
||||
gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
|
||||
gopkg.in/yaml.v2 v2.3.0 h1:clyUAQHOM3G0M3f5vQj7LuJrETvjVot3Z5el9nffUtU=
|
||||
gopkg.in/yaml.v2 v2.3.0/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
|
||||
gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY=
|
||||
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
|
||||
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
|
||||
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
|
||||
|
||||
@ -1,9 +0,0 @@
|
||||
package definition
|
||||
|
||||
type ErrorResponse struct {
|
||||
Error string `json:"error"`
|
||||
}
|
||||
|
||||
func Error(err string) *ErrorResponse {
|
||||
return &ErrorResponse{err}
|
||||
}
|
||||
@ -1,5 +0,0 @@
|
||||
package definition
|
||||
|
||||
type HealthResponse struct {
|
||||
Status string `json:"status,omitempty"`
|
||||
}
|
||||
@ -1,8 +0,0 @@
|
||||
package definition
|
||||
|
||||
type UpdateOrderStatusRequest struct {
|
||||
Status string `json:"status"`
|
||||
}
|
||||
|
||||
type UpdateOrderStatusResponse struct {
|
||||
}
|
||||
@ -6,6 +6,7 @@ type Config struct {
|
||||
AppID string
|
||||
AppName string
|
||||
AppDomain string
|
||||
PathPrefix string
|
||||
NetAddr string
|
||||
Port int
|
||||
RegistryAddr string
|
||||
@ -13,6 +14,8 @@ type Config struct {
|
||||
|
||||
LoggerAddr string `json:"logger_addr"`
|
||||
DbURL string `json:"db_url"`
|
||||
CacheAddr string `json:"cache_addr"`
|
||||
CachePassword string `json:"cache_password"`
|
||||
MongoDbUrl string `json:"mongodb_url"`
|
||||
EventBusURL string `json:"eventbus_url"`
|
||||
EventBusExchange string `json:"eventbus_exchange"`
|
||||
|
||||
@ -3,7 +3,7 @@ package server
|
||||
import (
|
||||
"github.com/gofiber/fiber/v2"
|
||||
|
||||
def "git.pbiernat.dev/egommerce/order-service/internal/app/definition"
|
||||
def "git.pbiernat.dev/egommerce/api-entities/http"
|
||||
)
|
||||
|
||||
func (s *Server) HealthHandler(c *fiber.Ctx) error {
|
||||
|
||||
@ -5,26 +5,29 @@ import (
|
||||
|
||||
"github.com/gofiber/fiber/v2"
|
||||
|
||||
def "git.pbiernat.dev/egommerce/order-service/internal/app/definition"
|
||||
def "git.pbiernat.dev/egommerce/api-entities/http"
|
||||
"git.pbiernat.dev/egommerce/order-service/internal/app/service"
|
||||
"git.pbiernat.dev/egommerce/order-service/internal/app/ui"
|
||||
)
|
||||
|
||||
func (s *Server) UpdateOrderStatusHandler(c *fiber.Ctx) error {
|
||||
reqID, _ := s.GetRequestID(c)
|
||||
data := new(def.UpdateOrderStatusRequest)
|
||||
if err := c.BodyParser(data); err != nil {
|
||||
return err
|
||||
req := new(def.UpdateOrderStatusRequest)
|
||||
if err := c.BodyParser(req); err != nil {
|
||||
return s.Error400(c, err.Error())
|
||||
}
|
||||
|
||||
// check if order exists in DB service...
|
||||
orderId := c.Params("orderId")
|
||||
var err error
|
||||
if orderId == "" || err != nil {
|
||||
return c.SendStatus(http.StatusBadRequest)
|
||||
orderID := c.Params("orderId", "")
|
||||
if orderID == "" {
|
||||
return s.Error400(c, "")
|
||||
}
|
||||
|
||||
order := service.NewOrderService(s.ebCh, s.log)
|
||||
order.UpdateOrderStatus(reqID, orderId, data.Status)
|
||||
orderSrv := service.NewOrderService(s.db, s.cache, s.ebCh, s.log)
|
||||
_, err := ui.UpdateOrderStatus(orderSrv, req.Status, orderID, reqID)
|
||||
if err != nil {
|
||||
return s.Error400(c, "Failed to update order status")
|
||||
}
|
||||
|
||||
return c.SendStatus(http.StatusNoContent)
|
||||
}
|
||||
|
||||
@ -5,9 +5,21 @@ import (
|
||||
|
||||
"git.pbiernat.dev/egommerce/go-api-pkg/fluentd"
|
||||
"github.com/gofiber/fiber/v2"
|
||||
"github.com/gofiber/fiber/v2/middleware/cors"
|
||||
)
|
||||
|
||||
var (
|
||||
defaultCORS = cors.New(cors.Config{
|
||||
AllowOrigins: "*",
|
||||
AllowCredentials: true,
|
||||
AllowMethods: "GET, POST, PATCH, PUT, DELETE, OPTIONS",
|
||||
AllowHeaders: "Accept, Authorization, Content-Type, Vary, X-Request-Id",
|
||||
})
|
||||
)
|
||||
|
||||
func SetupRoutes(s *Server) {
|
||||
s.App.Options("*", defaultCORS)
|
||||
|
||||
s.App.Get("/health", s.HealthHandler)
|
||||
s.App.Get("/config", s.ConfigHandler)
|
||||
|
||||
@ -18,6 +30,7 @@ func SetupRoutes(s *Server) {
|
||||
}
|
||||
|
||||
func SetupMiddlewares(s *Server) {
|
||||
s.App.Use(defaultCORS)
|
||||
s.App.Use(LoggingMiddleware(s.log))
|
||||
}
|
||||
|
||||
|
||||
@ -2,18 +2,19 @@ package server
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"os/signal"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
"github.com/go-redis/redis/v8"
|
||||
"github.com/gofiber/fiber/v2"
|
||||
"github.com/jackc/pgx/v4/pgxpool"
|
||||
"github.com/streadway/amqp"
|
||||
|
||||
def "git.pbiernat.dev/egommerce/api-entities/http"
|
||||
discovery "git.pbiernat.dev/egommerce/go-api-pkg/consul"
|
||||
"git.pbiernat.dev/egommerce/go-api-pkg/fluentd"
|
||||
)
|
||||
@ -23,6 +24,7 @@ type Server struct {
|
||||
conf *Config
|
||||
log *fluentd.Logger
|
||||
db *pgxpool.Pool
|
||||
cache *redis.Client
|
||||
ebCh *amqp.Channel
|
||||
discovery *discovery.Service
|
||||
name string
|
||||
@ -34,9 +36,8 @@ type Headers struct {
|
||||
RequestID string `reqHeader:"x-request-id"`
|
||||
}
|
||||
|
||||
func NewServer(conf *Config, logger *fluentd.Logger, db *pgxpool.Pool, ebCh *amqp.Channel) *Server {
|
||||
logger.Log("API_ID: %s", conf.AppID)
|
||||
consul, err := discovery.NewService(conf.RegistryAddr, conf.AppID, conf.AppName, conf.AppID, conf.AppDomain, conf.Port)
|
||||
func NewServer(conf *Config, logger *fluentd.Logger, db *pgxpool.Pool, cache *redis.Client, ebCh *amqp.Channel) *Server {
|
||||
consul, err := discovery.NewService(conf.RegistryAddr, conf.AppID, conf.AppName, conf.AppID, conf.AppDomain, conf.PathPrefix, conf.Port)
|
||||
if err != nil {
|
||||
logger.Log("Error connecting to %s: %v", conf.RegistryAddr, err)
|
||||
}
|
||||
@ -59,6 +60,7 @@ func NewServer(conf *Config, logger *fluentd.Logger, db *pgxpool.Pool, ebCh *amq
|
||||
conf,
|
||||
logger,
|
||||
db,
|
||||
cache,
|
||||
ebCh,
|
||||
consul,
|
||||
conf.AppName,
|
||||
@ -67,13 +69,21 @@ func NewServer(conf *Config, logger *fluentd.Logger, db *pgxpool.Pool, ebCh *amq
|
||||
}
|
||||
|
||||
go func(s *Server) { // Consul KV config updater
|
||||
interval := time.Second * 30
|
||||
interval := time.Second * 15
|
||||
ticker := time.NewTicker(interval)
|
||||
for range ticker.C {
|
||||
s.updateKVConfig()
|
||||
}
|
||||
}(s)
|
||||
|
||||
go func(s *Server) { // Server metadata cache updater
|
||||
interval := time.Second * 5
|
||||
ticker := time.NewTicker(interval)
|
||||
for range ticker.C {
|
||||
s.cacheMetadata()
|
||||
}
|
||||
}(s)
|
||||
|
||||
SetupMiddlewares(s)
|
||||
SetupRoutes(s)
|
||||
|
||||
@ -114,31 +124,51 @@ func (s *Server) GetRequestID(c *fiber.Ctx) (string, error) {
|
||||
return hdr.RequestID, nil
|
||||
}
|
||||
|
||||
func (s *Server) updateKVConfig() error { // FIXME: duplicated in cmd/worker/main.go
|
||||
config, _, err := s.discovery.KV().Get(s.kvNmspc, nil)
|
||||
if err != nil {
|
||||
fmt.Println(err)
|
||||
|
||||
return err
|
||||
func (s *Server) Error400(c *fiber.Ctx, msg string) error {
|
||||
return c.Status(fiber.StatusBadRequest).JSON(&def.ErrorResponse{Error: msg})
|
||||
}
|
||||
|
||||
if config == nil {
|
||||
return errors.New("empty KV config data")
|
||||
func (s *Server) Error404(c *fiber.Ctx, msg string) error {
|
||||
return c.Status(fiber.StatusNotFound).JSON(&def.ErrorResponse{Error: msg})
|
||||
}
|
||||
|
||||
func (s *Server) updateKVConfig() { // FIXME: duplicated in cmd/worker/main.go
|
||||
config, _, err := s.discovery.KV().Get(s.kvNmspc, nil)
|
||||
if err != nil || config == nil {
|
||||
return
|
||||
}
|
||||
|
||||
kvCnf := bytes.NewBuffer(config.Value)
|
||||
decoder := json.NewDecoder(kvCnf)
|
||||
if err := decoder.Decode(&s.conf); err != nil {
|
||||
return err
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
func (s *Server) cacheMetadata() {
|
||||
ctx := context.Background()
|
||||
key, address := "internal__"+s.conf.AppName+"__ips", s.conf.AppID // FIXME: key name
|
||||
|
||||
pos := s.cache.LPos(ctx, key, address, redis.LPosArgs{}).Val()
|
||||
if pos >= 0 {
|
||||
s.cache.LRem(ctx, key, 0, address)
|
||||
}
|
||||
|
||||
s.cache.LPush(ctx, key, address).Err()
|
||||
}
|
||||
|
||||
func (s *Server) clearMetadataCache() {
|
||||
ctx := context.Background()
|
||||
key, address := "internal__"+s.conf.AppName+"__ips", s.conf.AppID // FIXME: key name
|
||||
|
||||
s.cache.LRem(ctx, key, 0, address)
|
||||
}
|
||||
|
||||
func (s *Server) gracefulShutdown() error {
|
||||
s.log.Log("Server is going down...")
|
||||
s.log.Log("Unregistering service: %s", s.discovery.GetID())
|
||||
s.discovery.Unregister()
|
||||
s.clearMetadataCache()
|
||||
|
||||
s.ebCh.Close()
|
||||
s.db.Close()
|
||||
|
||||
@ -1,27 +1,73 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"git.pbiernat.dev/egommerce/api-entities/model"
|
||||
"git.pbiernat.dev/egommerce/go-api-pkg/api"
|
||||
"git.pbiernat.dev/egommerce/go-api-pkg/fluentd"
|
||||
amqp "git.pbiernat.dev/egommerce/go-api-pkg/rabbitmq"
|
||||
"git.pbiernat.dev/egommerce/order-service/internal/app/event"
|
||||
"github.com/go-redis/redis/v8"
|
||||
"github.com/jackc/pgx/v4/pgxpool"
|
||||
base "github.com/streadway/amqp"
|
||||
)
|
||||
|
||||
const (
|
||||
SERVICE_USER_AGENT = "order-httpclient"
|
||||
)
|
||||
|
||||
type OrderService struct {
|
||||
dbConn *pgxpool.Pool
|
||||
redis *redis.Client
|
||||
ebCh *base.Channel
|
||||
log *fluentd.Logger
|
||||
}
|
||||
|
||||
func NewOrderService(chn *base.Channel, log *fluentd.Logger) *OrderService {
|
||||
return &OrderService{chn, log}
|
||||
func NewOrderService(dbConn *pgxpool.Pool, redis *redis.Client, chn *base.Channel, log *fluentd.Logger) *OrderService {
|
||||
return &OrderService{dbConn, redis, chn, log}
|
||||
}
|
||||
|
||||
func (s *OrderService) Log(format string, val ...any) {
|
||||
s.log.Log(format, val...)
|
||||
}
|
||||
|
||||
// FIXME: REFACTOR !!
|
||||
func (s *OrderService) CreateOrder(ctx context.Context, basketID string) (*model.OrderModel, error) {
|
||||
basketAPI := api.NewBasketAPI(SERVICE_USER_AGENT, s.redis)
|
||||
basket, err := basketAPI.GetBasket(basketID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
order := new(model.OrderModel)
|
||||
order.State = basket.State // FIXME: are the same status?
|
||||
|
||||
sql := `INSERT INTO ordering."order"(state) VALUES($1) RETURNING id`
|
||||
if err := s.dbConn.QueryRow(ctx, sql, order.State).Scan(&order.ID); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
items, err := basketAPI.GetBasketItems(basket.ID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
for _, item := range items {
|
||||
sql := `INSERT INTO ordering.order_item(order_id,product_id,price,quantity) VALUES($1,$2,$3,$4)`
|
||||
if _, err := s.dbConn.Exec(ctx, sql, order.ID, item.ProductID, item.Price, item.Quantity); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
return order, nil
|
||||
}
|
||||
func (s *OrderService) UpdateOrderStatus(reqID, orderID, status string) (string, error) {
|
||||
s.log.Log("Update order#%s status to %s", orderID, status)
|
||||
|
||||
msg := &event.StatusUpdateEvent{Event: event.NewEvent(reqID), OrderID: orderID, Status: status}
|
||||
amqp.Publish(s.ebCh, "api-events", "order.email.statusUpdate", msg)
|
||||
amqp.Publish(s.ebCh, "api-events", "order.warehouse.orderFinalized", msg) // test event...
|
||||
// amqp.Publish(s.ebCh, "api-events", "order.warehouse.orderFinalized", msg) // test event...
|
||||
|
||||
return orderID, nil
|
||||
}
|
||||
|
||||
31
src/internal/app/ui/order.go
Normal file
31
src/internal/app/ui/order.go
Normal file
@ -0,0 +1,31 @@
|
||||
package ui
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
def "git.pbiernat.dev/egommerce/api-entities/http"
|
||||
"git.pbiernat.dev/egommerce/api-entities/model"
|
||||
"git.pbiernat.dev/egommerce/order-service/internal/app/service"
|
||||
)
|
||||
|
||||
// FIXME: REFACTOR !!
|
||||
func CreateOrder(srv *service.OrderService, basketID, reqID string) (*model.OrderModel, error) { // FIXME: model.Order
|
||||
ctx := context.Background()
|
||||
order, err := srv.CreateOrder(ctx, basketID)
|
||||
if err != nil {
|
||||
srv.Log("UI CreateOrder error: %v\n", err)
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return order, nil
|
||||
}
|
||||
|
||||
func UpdateOrderStatus(srv *service.OrderService, orderID, status, reqID string) (*def.UpdateOrderStatusResponse, error) {
|
||||
res := &def.UpdateOrderStatusResponse{}
|
||||
_, err := srv.UpdateOrderStatus(reqID, orderID, status)
|
||||
if err != nil {
|
||||
return res, err
|
||||
}
|
||||
|
||||
return res, nil
|
||||
}
|
||||
Loading…
Reference in New Issue
Block a user