jacob-mathison/query-orchestration
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Activity

Merged in feature/serviceconfig-integration (pull request #38)

Browse Source 
DRAFT PR : WIP working through ideas for integration

* movearound

* attempttwo

* openapi

* further sanding

* fix

* start on tests

* runthroughsingleconfig

* somechanges

* reflectissue

* removeerrs

* mostlyremovepanic

* removeenv

* noncfgtests

* go

* repo

* fix service config test

* add PWD to all

* test fix

* fix lint

* todo for later

* passingunittests

* alltests

* testlogger

* testloggername

* clean
This commit is contained in:
Jay Brown
2025-01-31 13:43:55 +00:00
parent 058f8dba7f
commit 15adaebfcd
138 changed files with 1864 additions and 1695 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/server/queue/config.go
-7
View File
@@ -1,17 +1,10 @@
package queue
import (
"context"
"github.com/aws/aws-sdk-go-v2/service/sqs"
"github.com/aws/aws-sdk-go-v2/service/sqs/types"
)
type Config struct {
URL string
Client *sqs.Client
}
type Controller interface {
Process(ctx context.Context, message *types.Message) error
}
internal/server/queue/listener.go
-54
View File
@@ -1,54 +0,0 @@
package queue
import (
"context"
"log"
"queryorchestration/internal/server"
"queryorchestration/internal/server/env"
"github.com/aws/aws-sdk-go-v2/config"
"github.com/aws/aws-sdk-go-v2/service/sqs"
)
type ListenerConfig struct {
Controller func(*server.Config) Controller
BasePath string
}
type Server struct {
controller Controller
queueConnection *Config
}
func NewServer(ctx context.Context, lConfig *ListenerConfig) *Server {
serverCfg := server.New(ctx, &server.NewConfig{
BasePath: lConfig.BasePath,
})
cfg, err := config.LoadDefaultConfig(ctx)
if err != nil {
log.Panicf("Unable to load SDK config: %v", err)
}
queueURL := env.GetPanic("QUEUE_URL")
sqsClient := sqs.NewFromConfig(cfg)
return &Server{
controller: lConfig.Controller(serverCfg),
queueConnection: &Config{
URL: queueURL,
Client: sqsClient,
},
}
}
func (s *Server) Listen(ctx context.Context) {
config := &PollConfig{
Controller: s.controller,
Config: s.queueConnection,
}
log.Print("Listening to queue")
PollMessages(ctx, config)
}
internal/server/queue/listener_test.go
-47
View File
@@ -1,47 +0,0 @@
package queue_test
import (
"context"
"os"
"path"
"queryorchestration/internal/server"
"queryorchestration/internal/server/queue"
"queryorchestration/internal/test"
"testing"
"time"
"github.com/stretchr/testify/assert"
)
func TestNew(t *testing.T) {
if testing.Short() {
t.Skip("Skipping long test in short mode")
}
ctx := context.Background()
_, cleanup := test.CreateQueue(t, ctx, &test.CreateQueueConfig{})
defer cleanup()
_, cleanup = test.CreateDB(t, ctx, &test.CreateDatabaseConfig{})
defer cleanup()
ctx, cancel := context.WithTimeout(ctx, time.Second)
defer cancel()
controller := func(cfg *server.Config) queue.Controller {
return MockController{}
}
t.Setenv("QUEUE_URL", "example_url")
queue.NewServer(ctx, &queue.ListenerConfig{
Controller: controller,
BasePath: path.Join(os.Getenv("PWD"), "../../.."),
})
}
func TestListen(t *testing.T) {
ctx := context.Background()
queue := queue.Server{}
assert.Panics(t, func() { queue.Listen(ctx) })
}
internal/server/queue/poll.go
-51
View File
@@ -1,51 +0,0 @@
package queue
import (
"context"
"fmt"
"log"
)
type PollConfig struct {
Config *Config
Controller Controller
}
func PollMessages(ctx context.Context, queueConfig *PollConfig) {
for {
select {
case <-ctx.Done():
return
default:
err := PollMessage(ctx, queueConfig)
if err != nil {
log.Print(err)
}
}
}
}
func PollMessage(ctx context.Context, queueConfig *PollConfig) error {
result, err := Receive(ctx, queueConfig.Config, []string{
"type",
})
if err != nil {
return fmt.Errorf("message fetch fail: %v", err)
}
for _, message := range result.Messages {
go func() {
err := queueConfig.Controller.Process(ctx, &message)
if err != nil {
log.Printf("message process fail: %v", err)
}
err = Delete(ctx, queueConfig.Config, &message)
if err != nil {
log.Printf("message delete fail: %v", err)
}
}()
}
return nil
}
internal/server/queue/poll_test.go
-62
View File
@@ -1,62 +0,0 @@
package queue_test
import (
"context"
"queryorchestration/internal/server/queue"
"queryorchestration/internal/test"
"testing"
"time"
"github.com/aws/aws-sdk-go-v2/service/sqs/types"
"github.com/stretchr/testify/assert"
)
type MockController struct{}
func (s MockController) Process(ctx context.Context, req *types.Message) error {
return nil
}
func TestPollMessages(t *testing.T) {
if testing.Short() {
t.Skip("Skipping long test in short mode")
}
ctx := context.Background()
queueConfig, cleanup := test.CreateQueue(t, ctx, &test.CreateQueueConfig{})
defer cleanup()
controller := MockController{}
ctx, cancel := context.WithTimeout(ctx, time.Second)
defer cancel()
cfg := &queue.Config{
URL: queueConfig.URL,
Client: queueConfig.Client,
}
queue.PollMessages(ctx, &queue.PollConfig{
Config: cfg,
Controller: controller,
})
}
func TestPollMessage(t *testing.T) {
if testing.Short() {
t.Skip("Skipping long test in short mode")
}
ctx := context.Background()
queueConfig, cleanup := test.CreateQueue(t, ctx, &test.CreateQueueConfig{})
defer cleanup()
controller := MockController{}
cfg := &queue.Config{
URL: queueConfig.URL,
Client: queueConfig.Client,
}
err := queue.PollMessage(ctx, &queue.PollConfig{
Config: cfg,
Controller: controller,
})
assert.Nil(t, err)
}