From 00e7dca3c78bb9f17cd0d66d016cad40ba7c9fd1 Mon Sep 17 00:00:00 2001 From: Michael McGuinness Date: Mon, 3 Feb 2025 18:51:44 +0000 Subject: [PATCH] Merged in feature/docinit (pull request #43) Move queue to config * movequeue * passtests --- .mockery.yml | 2 +- internal/document/init/create.go | 15 +++--- internal/objectstorage/get.go | 8 ++++ internal/server/queue/config.go | 6 --- internal/server/queue/delete.go | 21 --------- internal/server/queue/receive.go | 17 ------- internal/server/queue/send.go | 30 ------------ internal/server/runner/listener.go | 6 +-- internal/server/runner/poll.go | 23 +++++---- internal/server/runner/poll_test.go | 37 ++++++++------- .../{server => serviceconfig}/queue/client.go | 0 internal/serviceconfig/queue/config.go | 10 ++-- internal/serviceconfig/queue/delete.go | 25 ++++++++++ .../queue/delete_test.go | 21 ++++----- internal/serviceconfig/queue/receive.go | 22 +++++++++ .../queue/receive_test.go | 13 ++--- internal/serviceconfig/queue/send.go | 36 ++++++++++++++ .../queue/send_test.go | 18 ++++--- internal/test/queue.go | 19 +++++--- internal/test/queue_test.go | 47 +++++++++---------- test/docInitRunner/docinitrunner_test.go | 20 ++++---- test/queryRunner/queryrunner_test.go | 12 ++--- 22 files changed, 209 insertions(+), 199 deletions(-) create mode 100644 internal/objectstorage/get.go delete mode 100644 internal/server/queue/config.go delete mode 100644 internal/server/queue/delete.go delete mode 100644 internal/server/queue/receive.go delete mode 100644 internal/server/queue/send.go rename internal/{server => serviceconfig}/queue/client.go (100%) create mode 100644 internal/serviceconfig/queue/delete.go rename internal/{server => serviceconfig}/queue/delete_test.go (66%) create mode 100644 internal/serviceconfig/queue/receive.go rename internal/{server => serviceconfig}/queue/receive_test.go (72%) create mode 100644 internal/serviceconfig/queue/send.go rename internal/{server => serviceconfig}/queue/send_test.go (61%) diff --git a/.mockery.yml b/.mockery.yml index eca44192..1d8786fc 100644 --- a/.mockery.yml +++ b/.mockery.yml @@ -3,7 +3,7 @@ with-expecter: true outpkg: "{{.PackageName}}mock" dir: "mocks/{{.PackageName}}" packages: - queryorchestration/internal/server/queue: + queryorchestration/internal/serviceconfig/queue: interfaces: SQSClient: queryorchestration/internal/server/runner: diff --git a/internal/document/init/create.go b/internal/document/init/create.go index 88c44d97..fd43799b 100644 --- a/internal/document/init/create.go +++ b/internal/document/init/create.go @@ -7,7 +7,7 @@ import ( "queryorchestration/internal/document" documentclean "queryorchestration/internal/document/clean" "queryorchestration/internal/job" - "queryorchestration/internal/server/queue" + "queryorchestration/internal/serviceconfig/queue" "github.com/aws/aws-sdk-go-v2/service/sqs/types" "github.com/google/uuid" @@ -79,12 +79,13 @@ func (s *Service) informCreate(ctx context.Context, id uuid.UUID, j *job.Job) er return nil } - err := queue.Send(ctx, &queue.Config{ - Client: s.cfg.GetQueueClient(), - URL: s.cfg.GetDocumentCleanURL(), - }, documentclean.Create{ - ID: id, - }, map[string]types.MessageAttributeValue{}) + err := s.cfg.SendToQueue(ctx, &queue.SendParams{ + QueueURL: s.cfg.GetDocumentCleanURL(), + Body: documentclean.Create{ + ID: id, + }, + Attributes: map[string]types.MessageAttributeValue{}, + }) if err != nil { return err } diff --git a/internal/objectstorage/get.go b/internal/objectstorage/get.go new file mode 100644 index 00000000..9f55cdd2 --- /dev/null +++ b/internal/objectstorage/get.go @@ -0,0 +1,8 @@ +package objectstorage + +import ( + "context" +) + +func Get(ctx context.Context) { +} diff --git a/internal/server/queue/config.go b/internal/server/queue/config.go deleted file mode 100644 index e629cfb6..00000000 --- a/internal/server/queue/config.go +++ /dev/null @@ -1,6 +0,0 @@ -package queue - -type Config struct { - URL string - Client SQSClient -} diff --git a/internal/server/queue/delete.go b/internal/server/queue/delete.go deleted file mode 100644 index aa91287e..00000000 --- a/internal/server/queue/delete.go +++ /dev/null @@ -1,21 +0,0 @@ -package queue - -import ( - "context" - - "github.com/aws/aws-sdk-go-v2/aws" - "github.com/aws/aws-sdk-go-v2/service/sqs" - "github.com/aws/aws-sdk-go-v2/service/sqs/types" -) - -func Delete(ctx context.Context, config *Config, msg *types.Message) error { - _, err := config.Client.DeleteMessage(ctx, &sqs.DeleteMessageInput{ - QueueUrl: aws.String(config.URL), - ReceiptHandle: msg.ReceiptHandle, - }) - if err != nil { - return err - } - - return nil -} diff --git a/internal/server/queue/receive.go b/internal/server/queue/receive.go deleted file mode 100644 index 133cf01a..00000000 --- a/internal/server/queue/receive.go +++ /dev/null @@ -1,17 +0,0 @@ -package queue - -import ( - "context" - - "github.com/aws/aws-sdk-go-v2/service/sqs" -) - -func Receive(ctx context.Context, config *Config, attributes []string) (*sqs.ReceiveMessageOutput, error) { - return config.Client.ReceiveMessage(ctx, &sqs.ReceiveMessageInput{ - QueueUrl: &config.URL, - MaxNumberOfMessages: 1, - WaitTimeSeconds: 2, - VisibilityTimeout: 2, - MessageAttributeNames: attributes, - }) -} diff --git a/internal/server/queue/send.go b/internal/server/queue/send.go deleted file mode 100644 index 0a300238..00000000 --- a/internal/server/queue/send.go +++ /dev/null @@ -1,30 +0,0 @@ -package queue - -import ( - "context" - "encoding/json" - - "github.com/aws/aws-sdk-go-v2/aws" - "github.com/aws/aws-sdk-go-v2/service/sqs" - "github.com/aws/aws-sdk-go-v2/service/sqs/types" -) - -func Send(ctx context.Context, config *Config, body interface{}, attributes map[string]types.MessageAttributeValue) error { - jsonBytes, err := json.Marshal(body) - if err != nil { - return err - } - - strBody := string(jsonBytes) - - _, err = config.Client.SendMessage(ctx, &sqs.SendMessageInput{ - MessageAttributes: attributes, - QueueUrl: aws.String(config.URL), - MessageBody: aws.String(strBody), - }) - if err != nil { - return err - } - - return nil -} diff --git a/internal/server/runner/listener.go b/internal/server/runner/listener.go index e7d9d939..05c5f3ee 100644 --- a/internal/server/runner/listener.go +++ b/internal/server/runner/listener.go @@ -73,9 +73,7 @@ func New(ctx context.Context, cfg ListenerConfig) (*Server, error) { } return &Server{ - controller: cfg.GetController(), - queueURL: cfg.GetQueueURL(), - client: cfg.GetQueueClient(), - cleanup: cleanup, + cfg: cfg, + cleanup: cleanup, }, nil } diff --git a/internal/server/runner/poll.go b/internal/server/runner/poll.go index 2046d0ec..a08512a9 100644 --- a/internal/server/runner/poll.go +++ b/internal/server/runner/poll.go @@ -4,14 +4,12 @@ import ( "context" "fmt" "log/slog" - "queryorchestration/internal/server/queue" + "queryorchestration/internal/serviceconfig/queue" ) type Server struct { - controller Controller - queueURL string - client queue.SQSClient - cleanup func() error + cleanup func() error + cfg ListenerConfig } func (c *Server) Listen(ctx context.Context) { @@ -63,11 +61,9 @@ func (c *Server) Listen(ctx context.Context) { //} func (c *Server) pollMessage(ctx context.Context) error { - cfg := &queue.Config{ - URL: c.queueURL, - Client: c.client, - } - result, err := queue.Receive(ctx, cfg, []string{}) + result, err := c.cfg.ReceiveFromQueue(ctx, &queue.ReceiveParams{ + QueueURL: c.cfg.GetQueueURL(), + }) if err != nil { return fmt.Errorf("message fetch fail: %v", err) } @@ -78,12 +74,15 @@ func (c *Server) pollMessage(ctx context.Context) error { // simple case. // if process takes too long to complete then we need to use a thread pool. go func() { - err := c.controller.Process(ctx, &message) + err := c.cfg.GetController().Process(ctx, &message) if err != nil { slog.Error("message process fail", "err", err) } - err = queue.Delete(ctx, cfg, &message) + err = c.cfg.DeleteFromQueue(ctx, &queue.DeleteParams{ + QueueURL: c.cfg.GetQueueURL(), + ReceiptHandle: message.ReceiptHandle, + }) if err != nil { slog.Error("message delete fail", "err", err) } diff --git a/internal/server/runner/poll_test.go b/internal/server/runner/poll_test.go index 89c1e610..b5b243fe 100644 --- a/internal/server/runner/poll_test.go +++ b/internal/server/runner/poll_test.go @@ -2,7 +2,6 @@ package runner import ( "context" - "queryorchestration/internal/server/queue" queuemock "queryorchestration/mocks/queue" runnermock "queryorchestration/mocks/runner" "testing" @@ -17,12 +16,13 @@ import ( func TestPollMessages(t *testing.T) { ctx := context.Background() - controller := runnermock.NewMockController(t) mockSQS := queuemock.NewMockSQSClient(t) - cfg := &queue.Config{ - URL: "/i/am/here", - Client: mockSQS, + cfg := &BaseConfig{} + cfg.QueueClient = mockSQS + cfg.ControllerFunc = func() Controller { + return runnermock.NewMockController(t) } + cfg.QueueURL = "/i/am/here" res := sqs.ReceiveMessageOutput{Messages: []types.Message{}} @@ -30,7 +30,7 @@ func TestPollMessages(t *testing.T) { ReceiveMessage( mock.Anything, mock.MatchedBy(func(in *sqs.ReceiveMessageInput) bool { - return *in.QueueUrl == cfg.URL + return *in.QueueUrl == cfg.QueueURL }), mock.Anything, ). @@ -40,10 +40,8 @@ func TestPollMessages(t *testing.T) { defer cancel() scfg := &Server{ - controller: controller, - queueURL: cfg.URL, - client: cfg.Client, - cleanup: func() error { return nil }, + cfg: cfg, + cleanup: func() error { return nil }, } scfg.Listen(ctx) } @@ -51,26 +49,29 @@ func TestPollMessages(t *testing.T) { func TestPollMessage(t *testing.T) { ctx := context.Background() - controller := runnermock.NewMockController(t) mockSQS := queuemock.NewMockSQSClient(t) + cfg := &BaseConfig{} + cfg.QueueClient = mockSQS + cfg.ControllerFunc = func() Controller { + return runnermock.NewMockController(t) + } + cfg.QueueURL = "/i/am/here" - cfg := &Server{ - controller: controller, - queueURL: "/i/am/here", - client: mockSQS, - cleanup: func() error { return nil }, + ser := &Server{ + cfg: cfg, + cleanup: func() error { return nil }, } mockSQS.EXPECT(). ReceiveMessage( mock.Anything, mock.MatchedBy(func(in *sqs.ReceiveMessageInput) bool { - return *in.QueueUrl == cfg.queueURL + return *in.QueueUrl == cfg.QueueURL }), mock.Anything, ). Return(&sqs.ReceiveMessageOutput{}, nil) - err := cfg.pollMessage(ctx) + err := ser.pollMessage(ctx) assert.NoError(t, err) } diff --git a/internal/server/queue/client.go b/internal/serviceconfig/queue/client.go similarity index 100% rename from internal/server/queue/client.go rename to internal/serviceconfig/queue/client.go diff --git a/internal/serviceconfig/queue/config.go b/internal/serviceconfig/queue/config.go index 4cdd6560..d1463ed6 100644 --- a/internal/serviceconfig/queue/config.go +++ b/internal/serviceconfig/queue/config.go @@ -3,7 +3,6 @@ package queue import ( "context" "fmt" - "queryorchestration/internal/server/queue" "queryorchestration/internal/serviceconfig/aws" "github.com/aws/aws-sdk-go-v2/config" @@ -15,17 +14,20 @@ type QueueConfig struct { aws.AWSConfig EnableOtel bool `env:"ENABLE_OTEL" envDefault:"false"` AWSEndpointUrlSQS string `env:"AWS_ENDPOINT_URL_SQS"` - QueueClient queue.SQSClient + QueueClient SQSClient } type ConfigProvider interface { aws.ConfigProvider SetQueueClient(context.Context) error - GetQueueClient() queue.SQSClient + GetQueueClient() SQSClient PingQueueByURL(context.Context, string) error + SendToQueue(context.Context, *SendParams) error + ReceiveFromQueue(context.Context, *ReceiveParams) (*sqs.ReceiveMessageOutput, error) + DeleteFromQueue(context.Context, *DeleteParams) error } -func (c *QueueConfig) GetQueueClient() queue.SQSClient { +func (c *QueueConfig) GetQueueClient() SQSClient { return c.QueueClient } diff --git a/internal/serviceconfig/queue/delete.go b/internal/serviceconfig/queue/delete.go new file mode 100644 index 00000000..8366e265 --- /dev/null +++ b/internal/serviceconfig/queue/delete.go @@ -0,0 +1,25 @@ +package queue + +import ( + "context" + + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/aws/aws-sdk-go-v2/service/sqs" +) + +type DeleteParams struct { + QueueURL string + ReceiptHandle *string +} + +func (c *QueueConfig) DeleteFromQueue(ctx context.Context, params *DeleteParams) error { + _, err := c.QueueClient.DeleteMessage(ctx, &sqs.DeleteMessageInput{ + QueueUrl: aws.String(params.QueueURL), + ReceiptHandle: params.ReceiptHandle, + }) + if err != nil { + return err + } + + return nil +} diff --git a/internal/server/queue/delete_test.go b/internal/serviceconfig/queue/delete_test.go similarity index 66% rename from internal/server/queue/delete_test.go rename to internal/serviceconfig/queue/delete_test.go index 38215d94..4508d1ec 100644 --- a/internal/server/queue/delete_test.go +++ b/internal/serviceconfig/queue/delete_test.go @@ -2,12 +2,11 @@ package queue_test import ( "context" - "queryorchestration/internal/server/queue" + "queryorchestration/internal/serviceconfig/queue" queuemock "queryorchestration/mocks/queue" "testing" "github.com/aws/aws-sdk-go-v2/service/sqs" - "github.com/aws/aws-sdk-go-v2/service/sqs/types" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/mock" ) @@ -15,26 +14,24 @@ import ( func TestDelete(t *testing.T) { ctx := context.Background() mockSQS := queuemock.NewMockSQSClient(t) - cfg := &queue.Config{ - URL: "/i/am/here", - Client: mockSQS, - } + cfg := queue.QueueConfig{} + cfg.QueueClient = mockSQS handle := "iamahandler" + params := &queue.DeleteParams{ + ReceiptHandle: &handle, + QueueURL: "/i/am/here", + } mockSQS.EXPECT(). DeleteMessage( mock.Anything, mock.MatchedBy(func(in *sqs.DeleteMessageInput) bool { - return *in.QueueUrl == cfg.URL && *in.ReceiptHandle == handle + return *in.QueueUrl == params.QueueURL && *in.ReceiptHandle == handle }), mock.Anything, ). Return(&sqs.DeleteMessageOutput{}, nil) - message := &types.Message{ - ReceiptHandle: &handle, - } - - err := queue.Delete(ctx, cfg, message) + err := cfg.DeleteFromQueue(ctx, params) assert.NoError(t, err) } diff --git a/internal/serviceconfig/queue/receive.go b/internal/serviceconfig/queue/receive.go new file mode 100644 index 00000000..d643196f --- /dev/null +++ b/internal/serviceconfig/queue/receive.go @@ -0,0 +1,22 @@ +package queue + +import ( + "context" + + "github.com/aws/aws-sdk-go-v2/service/sqs" +) + +type ReceiveParams struct { + QueueURL string + Attributes []string +} + +func (c *QueueConfig) ReceiveFromQueue(ctx context.Context, params *ReceiveParams) (*sqs.ReceiveMessageOutput, error) { + return c.QueueClient.ReceiveMessage(ctx, &sqs.ReceiveMessageInput{ + QueueUrl: ¶ms.QueueURL, + MaxNumberOfMessages: 1, + WaitTimeSeconds: 2, + VisibilityTimeout: 2, + MessageAttributeNames: params.Attributes, + }) +} diff --git a/internal/server/queue/receive_test.go b/internal/serviceconfig/queue/receive_test.go similarity index 72% rename from internal/server/queue/receive_test.go rename to internal/serviceconfig/queue/receive_test.go index 452ba2f3..a801c392 100644 --- a/internal/server/queue/receive_test.go +++ b/internal/serviceconfig/queue/receive_test.go @@ -2,7 +2,7 @@ package queue_test import ( "context" - "queryorchestration/internal/server/queue" + "queryorchestration/internal/serviceconfig/queue" queuemock "queryorchestration/mocks/queue" "testing" @@ -15,9 +15,10 @@ import ( func TestReceive(t *testing.T) { ctx := context.Background() mockSQS := queuemock.NewMockSQSClient(t) - cfg := &queue.Config{ - URL: "/i/am/here", - Client: mockSQS, + cfg := queue.QueueConfig{} + cfg.QueueClient = mockSQS + params := &queue.ReceiveParams{ + QueueURL: "/i/am/here", } res := sqs.ReceiveMessageOutput{Messages: []types.Message{}} @@ -26,13 +27,13 @@ func TestReceive(t *testing.T) { ReceiveMessage( mock.Anything, mock.MatchedBy(func(in *sqs.ReceiveMessageInput) bool { - return *in.QueueUrl == cfg.URL + return *in.QueueUrl == params.QueueURL }), mock.Anything, ). Return(&res, nil) - ares, err := queue.Receive(ctx, cfg, []string{}) + ares, err := cfg.ReceiveFromQueue(ctx, params) assert.NoError(t, err) assert.EqualExportedValues(t, res, *ares) } diff --git a/internal/serviceconfig/queue/send.go b/internal/serviceconfig/queue/send.go new file mode 100644 index 00000000..b27c5712 --- /dev/null +++ b/internal/serviceconfig/queue/send.go @@ -0,0 +1,36 @@ +package queue + +import ( + "context" + "encoding/json" + + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/aws/aws-sdk-go-v2/service/sqs" + "github.com/aws/aws-sdk-go-v2/service/sqs/types" +) + +type SendParams struct { + QueueURL string + Body interface{} + Attributes map[string]types.MessageAttributeValue +} + +func (c *QueueConfig) SendToQueue(ctx context.Context, params *SendParams) error { + jsonBytes, err := json.Marshal(params.Body) + if err != nil { + return err + } + + strBody := string(jsonBytes) + + _, err = c.QueueClient.SendMessage(ctx, &sqs.SendMessageInput{ + MessageAttributes: params.Attributes, + QueueUrl: aws.String(params.QueueURL), + MessageBody: aws.String(strBody), + }) + if err != nil { + return err + } + + return nil +} diff --git a/internal/server/queue/send_test.go b/internal/serviceconfig/queue/send_test.go similarity index 61% rename from internal/server/queue/send_test.go rename to internal/serviceconfig/queue/send_test.go index 20c6661d..c86eca17 100644 --- a/internal/server/queue/send_test.go +++ b/internal/serviceconfig/queue/send_test.go @@ -2,12 +2,11 @@ package queue_test import ( "context" - "queryorchestration/internal/server/queue" + "queryorchestration/internal/serviceconfig/queue" queuemock "queryorchestration/mocks/queue" "testing" "github.com/aws/aws-sdk-go-v2/service/sqs" - "github.com/aws/aws-sdk-go-v2/service/sqs/types" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/mock" ) @@ -15,24 +14,23 @@ import ( func TestSend(t *testing.T) { ctx := context.Background() mockSQS := queuemock.NewMockSQSClient(t) - cfg := &queue.Config{ - URL: "/i/am/here", - Client: mockSQS, + cfg := queue.QueueConfig{} + cfg.QueueClient = mockSQS + params := &queue.SendParams{ + QueueURL: "/i/am/here", + Body: struct{}{}, } mockSQS.EXPECT(). SendMessage( mock.Anything, mock.MatchedBy(func(in *sqs.SendMessageInput) bool { - return *in.QueueUrl == cfg.URL && *in.MessageBody == "{}" + return *in.QueueUrl == params.QueueURL && *in.MessageBody == "{}" }), mock.Anything, ). Return(&sqs.SendMessageOutput{}, nil) - body := struct{}{} - attrs := map[string]types.MessageAttributeValue{} - - err := queue.Send(ctx, cfg, body, attrs) + err := cfg.SendToQueue(ctx, params) assert.NoError(t, err) } diff --git a/internal/test/queue.go b/internal/test/queue.go index 5abc877b..78002150 100644 --- a/internal/test/queue.go +++ b/internal/test/queue.go @@ -3,8 +3,8 @@ package test import ( "context" "fmt" - "queryorchestration/internal/server/queue" "queryorchestration/internal/serviceconfig" + "queryorchestration/internal/serviceconfig/queue" "testing" "time" @@ -131,10 +131,10 @@ func CreateQueue(t *testing.T, ctx context.Context, cfg *serviceconfig.BaseConfi return *queueM.QueueUrl } -func AssertMessageWait(t *testing.T, ctx context.Context, cfg *queue.Config, attrs []string) types.Message { +func AssertMessageWait(t *testing.T, ctx context.Context, cfg serviceconfig.ConfigProvider, params *queue.ReceiveParams) types.Message { time.Sleep(1 * time.Second) - result, err := queue.Receive(ctx, cfg, attrs) + result, err := cfg.ReceiveFromQueue(ctx, params) assert.NoError(t, err) assert.NotNil(t, result.Messages) assert.Len(t, result.Messages, 1) @@ -143,14 +143,19 @@ func AssertMessageWait(t *testing.T, ctx context.Context, cfg *queue.Config, att return result.Messages[0] } -func AssertMessageBodyWait(t *testing.T, ctx context.Context, cfg *queue.Config, body string) { - message := AssertMessageWait(t, ctx, cfg, []string{}) +func AssertMessageBodyWait(t *testing.T, ctx context.Context, cfg serviceconfig.ConfigProvider, url string, body string) { + message := AssertMessageWait(t, ctx, cfg, &queue.ReceiveParams{ + QueueURL: url, + }) assert.Equal(t, body, *message.Body) } -func AssertMessageAttrWait(t *testing.T, ctx context.Context, cfg *queue.Config, name string, value string) { - message := AssertMessageWait(t, ctx, cfg, []string{name}) +func AssertMessageAttrWait(t *testing.T, ctx context.Context, cfg serviceconfig.ConfigProvider, url string, name string, value string) { + message := AssertMessageWait(t, ctx, cfg, &queue.ReceiveParams{ + QueueURL: url, + Attributes: []string{name}, + }) assert.NotNil(t, message.MessageAttributes[name]) assert.Equal(t, value, *(message.MessageAttributes[name]).StringValue) diff --git a/internal/test/queue_test.go b/internal/test/queue_test.go index 32d1de27..a1ff3298 100644 --- a/internal/test/queue_test.go +++ b/internal/test/queue_test.go @@ -2,7 +2,7 @@ package test import ( "context" - "queryorchestration/internal/server/queue" + "queryorchestration/internal/serviceconfig/queue" "testing" "github.com/aws/aws-sdk-go-v2/aws" @@ -73,15 +73,15 @@ func TestAssertMessageWait(t *testing.T) { defer cleanup() url := CreateQueue(t, ctx, cfg, "myname") - qucfg := &queue.Config{ - URL: url, - Client: cfg.QueueClient, - } - - err := queue.Send(ctx, qucfg, "body", map[string]types.MessageAttributeValue{}) + err := cfg.SendToQueue(ctx, &queue.SendParams{ + QueueURL: url, + Body: "body", + }) assert.NoError(t, err) - msg := AssertMessageWait(t, ctx, qucfg, []string{}) + msg := AssertMessageWait(t, ctx, cfg, &queue.ReceiveParams{ + QueueURL: url, + }) assert.NotNil(t, msg) } @@ -99,15 +99,13 @@ func TestAssertMessageBodyWait(t *testing.T) { defer cleanup() url := CreateQueue(t, ctx, cfg, "myname") - qucfg := &queue.Config{ - URL: url, - Client: cfg.QueueClient, - } - - err := queue.Send(ctx, qucfg, "body", map[string]types.MessageAttributeValue{}) + err := cfg.SendToQueue(ctx, &queue.SendParams{ + QueueURL: url, + Body: "body", + }) assert.NoError(t, err) - AssertMessageBodyWait(t, ctx, qucfg, "\"body\"") + AssertMessageBodyWait(t, ctx, cfg, url, "\"body\"") } func TestAssertMessageAttrWait(t *testing.T) { @@ -124,21 +122,20 @@ func TestAssertMessageAttrWait(t *testing.T) { defer cleanup() url := CreateQueue(t, ctx, cfg, "myname") - qucfg := &queue.Config{ - URL: url, - Client: cfg.QueueClient, - } - name := "name" value := "value" - err := queue.Send(ctx, qucfg, "body", map[string]types.MessageAttributeValue{ - name: { - DataType: aws.String("String"), - StringValue: aws.String(value), + err := cfg.SendToQueue(ctx, &queue.SendParams{ + QueueURL: url, + Body: "body", + Attributes: map[string]types.MessageAttributeValue{ + name: { + DataType: aws.String("String"), + StringValue: aws.String(value), + }, }, }) assert.NoError(t, err) - AssertMessageAttrWait(t, ctx, qucfg, name, value) + AssertMessageAttrWait(t, ctx, cfg, url, name, value) } diff --git a/test/docInitRunner/docinitrunner_test.go b/test/docInitRunner/docinitrunner_test.go index f56d6088..b9f25261 100644 --- a/test/docInitRunner/docinitrunner_test.go +++ b/test/docInitRunner/docinitrunner_test.go @@ -5,12 +5,11 @@ import ( "os" "path" documentinit "queryorchestration/internal/document/init" - "queryorchestration/internal/server/queue" + "queryorchestration/internal/serviceconfig/queue" "queryorchestration/internal/test" queryservice "queryorchestration/pkg/queryService" "testing" - "github.com/aws/aws-sdk-go-v2/service/sqs/types" "github.com/stretchr/testify/assert" ) @@ -76,16 +75,13 @@ func TestDocInitRunner(t *testing.T) { Location: "/I/am/here", // TODO } - qcfg := &queue.Config{ - URL: net.Runners[test.DocInitRunner].URI, - Client: cfg.QueueClient, - } - - err = queue.Send(ctx, qcfg, document, map[string]types.MessageAttributeValue{}) + err = cfg.SendToQueue(ctx, &queue.SendParams{ + QueueURL: net.Runners[test.DocInitRunner].URI, + Body: document, + }) assert.NoError(t, err) - _ = test.AssertMessageWait(t, ctx, &queue.Config{ - URL: doccleanurl, - Client: cfg.GetQueueClient(), - }, []string{}) + _ = test.AssertMessageWait(t, ctx, cfg, &queue.ReceiveParams{ + QueueURL: doccleanurl, + }) } diff --git a/test/queryRunner/queryrunner_test.go b/test/queryRunner/queryrunner_test.go index f9debf47..41b621a7 100644 --- a/test/queryRunner/queryrunner_test.go +++ b/test/queryRunner/queryrunner_test.go @@ -3,11 +3,10 @@ package integration_test import ( "context" "queryorchestration/internal/query" - "queryorchestration/internal/server/queue" + "queryorchestration/internal/serviceconfig/queue" "queryorchestration/internal/test" "testing" - "github.com/aws/aws-sdk-go-v2/service/sqs/types" "github.com/google/uuid" "github.com/stretchr/testify/assert" ) @@ -29,12 +28,11 @@ func TestQueryRunner(t *testing.T) { CleanVersion: int32(1), TextVersion: int32(1), } - qcfg := &queue.Config{ - URL: c.URI, - Client: cfg.QueueClient, - } - err := queue.Send(ctx, qcfg, document, map[string]types.MessageAttributeValue{}) + err := cfg.SendToQueue(ctx, &queue.SendParams{ + QueueURL: c.URI, + Body: document, + }) assert.NoError(t, err) // TODO - check document output