diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 5f6188b9..746972a3 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -113,7 +113,7 @@ docker run -d --rm --name postgres -e POSTGRES_PASSWORD=password -p 5432:5432 po Run `webhook-gateway` integration tests using: ```shell -APP_PORT=8080 APP_WEBHOOK_SECRET_VALUE=fake RABBITMQ_CONNECT_STRING="amqp://guest:guest@localhost:5672/" go test -cover -v ./cmd/webhook-gateway -integration +APP_WEBHOOK_SECRET_VALUE=fake APP_PORT=8080 RABBITMQ_CONNECT_STRING="amqp://guest:guest@localhost:5672/" go test -v -cover -tags=integration -race ./cmd/webhook-gateway/... ``` It assumes you have access to a RabbitMQ server running reachable at $RABBITMQ_CONNECT_STRING. @@ -126,11 +126,10 @@ docker run -d --rm --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:4-manag Run `planner` integration tests using: ```shell -POSTGRESQL_DB_CONNECT_STRING="postgres://postgres:postgres@localhost:5432/gh_runner_operators?sslmode=disable" APP_PORT=8080 go test ./cmd/planner -v -count=1 +APP_PORT=8080 POSTGRESQL_DB_CONNECT_STRING="postgres://postgres:password@localhost:5432/postgres?sslmode=disable" RABBITMQ_CONNECT_STRING="amqp://guest:guest@localhost:5672/" go test -v -cover -tags=integration -race ./cmd/planner/... ``` -It assumes you are connected to a local PostgreSQL database "gh_runner_operators" as user "postgres" on host "localhost" at port "5432". - +It assumes you can reach postgres and rabbitmq servers as described above. ### Test design diff --git a/cmd/planner/main.go b/cmd/planner/main.go index bcfccb5c..e2142002 100644 --- a/cmd/planner/main.go +++ b/cmd/planner/main.go @@ -23,6 +23,7 @@ import ( "github.com/canonical/github-runner-operators/internal/database" "github.com/canonical/github-runner-operators/internal/planner" + "github.com/canonical/github-runner-operators/internal/queue" "github.com/canonical/github-runner-operators/internal/telemetry" "github.com/canonical/github-runner-operators/internal/version" "go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp" @@ -33,7 +34,6 @@ const ( portEnvVar = "APP_PORT" serviceName = "github-runner-planner" rabbitMQUriEnvVar = "RABBITMQ_CONNECT_STRING" - queueName = "webhook-queue" shutdownTimeout = 30 * time.Second ) @@ -71,7 +71,8 @@ func main() { metrics := planner.NewMetrics(db) - consumer := planner.NewJobConsumer(rabbitMQUri, queueName, db, metrics) + amqpConsumer := queue.NewAmqpConsumer(rabbitMQUri, queue.DefaultQueueConfig()) + consumer := planner.NewJobConsumer(amqpConsumer, db, metrics) var consumerWg sync.WaitGroup consumerWg.Add(1) diff --git a/cmd/planner/main_test.go b/cmd/planner/main_test.go index 6e645cc5..57b3d89b 100644 --- a/cmd/planner/main_test.go +++ b/cmd/planner/main_test.go @@ -19,6 +19,7 @@ import ( "testing" "time" + "github.com/canonical/github-runner-operators/internal/queue" amqp "github.com/rabbitmq/amqp091-go" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -74,7 +75,7 @@ func TestMain_FlavorPressure(t *testing.T) { assert.Eventually(t, func() bool { press := ctx.getFlavorPressure(flavor) return press[flavor] > pressure - }, 20*time.Minute, 500*time.Millisecond) + }, 2*time.Minute, 100*time.Millisecond, "expected flavor pressure to increase after webhook processing") // Trigger graceful shutdown and verify services stop cleanly ctx.shutdownMain() @@ -112,7 +113,7 @@ func (ctx *testContext) constructWebhookPayload(labels []string) []byte { return body } -// declareQueue declares a RabbitMQ queue for testing purposes. +// declareQueue declares the exchange, queue, and binding for testing. func (ctx *testContext) declareQueue() { ctx.t.Helper() @@ -124,6 +125,19 @@ func (ctx *testContext) declareQueue() { require.NoError(ctx.t, err, "open channel") defer ch.Close() + config := queue.DefaultQueueConfig() + + err = ch.ExchangeDeclare( + config.ExchangeName, + "direct", // type + true, // durable + false, // auto-delete + false, // internal + false, // no-wait + nil, + ) + require.NoError(ctx.t, err, "declare exchange") + _, err = ch.QueueDeclare( ctx.queueName, true, // durable @@ -133,9 +147,18 @@ func (ctx *testContext) declareQueue() { nil, ) require.NoError(ctx.t, err, "declare queue") + + err = ch.QueueBind( + ctx.queueName, + config.RoutingKey, + config.ExchangeName, + false, // no-wait + nil, + ) + require.NoError(ctx.t, err, "bind queue") } -// publishAndWaitAck publishes a message to the given RabbitMQ queue and waits for an ack. +// publishAndWaitAck publishes a message to the exchange with routing key and waits for an ack. func (ctx *testContext) publishAndWaitAck(body []byte) { ctx.t.Helper() @@ -152,8 +175,11 @@ func (ctx *testContext) publishAndWaitAck(body []byte) { confirms := ch.NotifyPublish(make(chan amqp.Confirmation, 1)) + config := queue.DefaultQueueConfig() err = ch.Publish( - "", ctx.queueName, false, false, + config.ExchangeName, // publish to exchange instead of directly to queue + config.RoutingKey, // use routing key + false, false, amqp.Publishing{ ContentType: "application/json", Body: body, diff --git a/cmd/webhook-gateway/main.go b/cmd/webhook-gateway/main.go index 50a6e23c..e378d35a 100644 --- a/cmd/webhook-gateway/main.go +++ b/cmd/webhook-gateway/main.go @@ -19,7 +19,6 @@ import ( "go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp" ) -const queueName = "webhook-queue" const webhookPath = "/webhook" const portEnvVar = "APP_PORT" const rabbitMQUriEnvVar = "RABBITMQ_CONNECT_STRING" @@ -51,7 +50,7 @@ func main() { log.Fatalln(webhookSecretEnvVar + " environment variable not set") } - p := queue.NewAmqpProducer(uri, queueName) + p := queue.NewAmqpProducer(uri, queue.DefaultQueueConfig()) handler := &webhook.Handler{ WebhookSecret: webhookSecret, Producer: p, diff --git a/cmd/webhook-gateway/webhook_test.go b/cmd/webhook-gateway/webhook_test.go index f3dbb9dd..18d462cd 100644 --- a/cmd/webhook-gateway/webhook_test.go +++ b/cmd/webhook-gateway/webhook_test.go @@ -19,19 +19,23 @@ import ( "testing" "time" + "github.com/canonical/github-runner-operators/internal/queue" + amqp "github.com/rabbitmq/amqp091-go" "github.com/stretchr/testify/assert" ) func TestHTTPRequestIsForwarded(t *testing.T) { const payload = `{"message":"Hello, Bob!"}` + amqpUri := getAmqpUriFromEnv(t) + + setupQueue(t, amqpUri) go main() secret := getSecretFromEnv(t) sendPayloadToHTTPServer(t, payload, secret) - amqpUri := getAmqpUriFromEnv(t) msg := consumeMessage(t, amqpUri) assert.Equal(t, payload, msg, "expected message body to match") @@ -98,32 +102,82 @@ func createSignature(message string, secret string) string { return "sha256=" + hex.EncodeToString(h.Sum(nil)) } -func consumeMessage(t *testing.T, amqpUri string) string { +// returns an AMQP channel and a cleanup function to close the connection and channel +func setupAMQPChannel(t *testing.T, amqpUri string) (*amqp.Channel, func()) { conn, err := amqp.Dial(amqpUri) if err != nil { t.Fatalf("Failed to connect to RabbitMQ: %v", err) } - defer conn.Close() + ch, err := conn.Channel() if err != nil { + conn.Close() t.Fatalf("Failed to open a channel: %v", err) } - defer ch.Close() + cleanup := func() { + ch.Close() + conn.Close() + } + + return ch, cleanup +} + +// setupQueue declares the exchange, queue, and binding for the test +// this is required because the webhook-gateway does not create the queue and +// published messages could be discarded if the queue does not exist +func setupQueue(t *testing.T, amqpUri string) { + ch, cleanup := setupAMQPChannel(t, amqpUri) + defer cleanup() + + config := queue.DefaultQueueConfig() + + err := ch.ExchangeDeclare( + config.ExchangeName, // name) + "direct", // type + true, // durable + false, // auto-deleted + false, // internal + false, // no-wait + nil, // arguments + ) + + if err != nil { + t.Fatalf("Failed to declare an exchange: %v", err) + } _, err = ch.QueueDeclare( - "webhook-queue", // name - true, // durable - false, // delete when unused - false, // exclusive - false, // no-wait - nil, // arguments + config.QueueName, // name + true, // durable + false, // delete when unused + false, // exclusive + false, // no-wait + nil, // arguments ) if err != nil { t.Fatalf("Failed to declare a queue: %v", err) } - deliveryChan, err := ch.Consume("webhook-queue", "", true, false, false, false, nil) + err = ch.QueueBind( + config.QueueName, // queue name + config.RoutingKey, // routing key + config.ExchangeName, // exchange + false, + nil, + ) + + if err != nil { + t.Fatalf("Failed to bind a queue: %v", err) + } +} + +func consumeMessage(t *testing.T, amqpUri string) string { + ch, cleanup := setupAMQPChannel(t, amqpUri) + defer cleanup() + + config := queue.DefaultQueueConfig() + + deliveryChan, err := ch.Consume(config.QueueName, "", true, false, false, false, nil) if err != nil { t.Fatalf("Failed to register a consumer: %v", err) diff --git a/internal/planner/consumer.go b/internal/planner/consumer.go index 24eda31f..7bb30dc5 100644 --- a/internal/planner/consumer.go +++ b/internal/planner/consumer.go @@ -32,9 +32,9 @@ const ( maxBackoff = 5 * time.Minute ) -func NewJobConsumer(amqpUri, queueName string, db JobDatabase, metrics *Metrics) *JobConsumer { +func NewJobConsumer(consumer queue.Consumer, db JobDatabase, metrics *Metrics) *JobConsumer { return &JobConsumer{ - consumer: queue.NewAmqpConsumer(amqpUri, queueName), + consumer: consumer, db: db, metrics: metrics, } diff --git a/internal/queue/config.go b/internal/queue/config.go new file mode 100644 index 00000000..da334189 --- /dev/null +++ b/internal/queue/config.go @@ -0,0 +1,29 @@ +/* + * Copyright 2026 Canonical Ltd. + * See LICENSE file for licensing details. + */ + +package queue + +// Default AMQP configuration values. +const ( + defaultExchangeName = "github-workflow-jobs" + defaultQueueName = "github-workflow-jobs" + defaultRoutingKey = "workflow-job" +) + +// QueueConfig holds AMQP exchange, queue, and routing configuration. +type QueueConfig struct { + ExchangeName string + QueueName string + RoutingKey string +} + +// DefaultQueueConfig returns a QueueConfig with default values. +func DefaultQueueConfig() QueueConfig { + return QueueConfig{ + ExchangeName: defaultExchangeName, + QueueName: defaultQueueName, + RoutingKey: defaultRoutingKey, + } +} diff --git a/internal/queue/consumer.go b/internal/queue/consumer.go index 5b160e4e..e96ab814 100644 --- a/internal/queue/consumer.go +++ b/internal/queue/consumer.go @@ -18,19 +18,19 @@ import ( // AmqpConsumer is an AMQP consumer for workflow job events. type AmqpConsumer struct { - client *Client - queueName string - channel <-chan amqp.Delivery + client *Client + config QueueConfig + channel <-chan amqp.Delivery } // NewAmqpConsumer creates a new AmqpConsumer with the given dependencies. -func NewAmqpConsumer(uri, queueName string) *AmqpConsumer { +func NewAmqpConsumer(uri string, config QueueConfig) *AmqpConsumer { return &AmqpConsumer{ - queueName: queueName, client: &Client{ uri: uri, connectFunc: amqpConnect, }, + config: config, } } @@ -81,13 +81,13 @@ func (c *AmqpConsumer) ensureChannel() error { } channel, err := c.client.amqpChannel.Consume( - c.queueName, // queue - "", // consumer tag, empty string means a unique random tag will be generated - false, // whether rabbitmq auto-acknowledges messages - true, // whether only this consumer can access the queue - false, // no-local - false, // false means wait for server confirmation that consumer is registered - nil, // args + c.config.QueueName, // queue + "", // consumer tag, empty string means a unique random tag will be generated + false, // whether rabbitmq auto-acknowledges messages + true, // whether only this consumer can access the queue + false, // no-local + false, // false means wait for server confirmation that consumer is registered + nil, // args ) if err != nil { return fmt.Errorf("cannot consume amqp messages: %w", err) @@ -106,7 +106,22 @@ func (c *AmqpConsumer) ensureAmqpChannel() error { } if c.client.amqpChannel == nil || c.client.amqpChannel.IsClosed() { - err := c.client.resetChannel(c.queueName, false) + err := c.client.resetChannel() + if err != nil { + return err + } + + err = c.client.declareExchange(c.config.ExchangeName) + if err != nil { + return err + } + + err = c.client.declareQueue(c.config.QueueName) + if err != nil { + return err + } + + err = c.client.bindQueue(c.config.QueueName, c.config.RoutingKey, c.config.ExchangeName) if err != nil { return err } diff --git a/internal/queue/consumer_test.go b/internal/queue/consumer_test.go index 627e678f..a5945c4e 100644 --- a/internal/queue/consumer_test.go +++ b/internal/queue/consumer_test.go @@ -1,5 +1,5 @@ /* - * Copyright 2025 Canonical Ltd. + * Copyright 2026 Canonical Ltd. * See LICENSE file for licensing details. */ @@ -14,9 +14,6 @@ import ( "github.com/stretchr/testify/assert" ) -const queueWithConsumeError = "queue-with-consume-error" -const queueWithQosError = "queue-with-qos-error" - type MockAmqpChannelConsumer struct { MockAmqpChannel msgChannel chan amqp.Delivery @@ -27,10 +24,12 @@ type MockAmqpChannelConsumer struct { qosCalls int consumerTag string consumerAutoAck bool + consumeError bool + qosError bool } func (ch *MockAmqpChannelConsumer) Consume(queue, consumer string, autoAck, exclusive, noLocal, noWait bool, args amqp.Table) (<-chan amqp.Delivery, error) { - if queue == queueWithConsumeError { + if ch.consumeError { return nil, errors.New("consume error") } ch.consumeCalls++ @@ -43,7 +42,7 @@ func (ch *MockAmqpChannelConsumer) Consume(queue, consumer string, autoAck, excl } func (ch *MockAmqpChannelConsumer) Qos(prefetchCount, prefetchSize int, global bool) error { - if ch.queueName == queueWithQosError { + if ch.qosError { return errors.New("qos error") } ch.qosCalls++ @@ -59,6 +58,9 @@ type MockAmqpConnectionConsumer struct { isclosed bool errMode bool confirmModeError bool + exchangeDeclareError bool + queueDeclareError bool + bindError bool returnChannelOnSetup bool } @@ -72,7 +74,10 @@ func (m *MockAmqpConnectionConsumer) Channel() (amqpChannel, error) { } m.amqpChannelConsumer = &MockAmqpChannelConsumer{ MockAmqpChannel: MockAmqpChannel{ - confirmModeError: m.confirmModeError, + confirmModeError: m.confirmModeError, + exchangeDeclareError: m.exchangeDeclareError, + queueDeclareError: m.queueDeclareError, + bindError: m.bindError, }, } return m.amqpChannelConsumer, nil @@ -112,8 +117,8 @@ func TestPull(t *testing.T) { amqpChannelConsumer: mockAmqpChannel, }, }, - queueName: queueName, - channel: msgChannel, + config: DefaultQueueConfig(), + channel: msgChannel, } msg, err := amqpConsumer.Pull(context.Background()) @@ -143,8 +148,8 @@ func TestPullContextCanceled(t *testing.T) { amqpChannelConsumer: mockAmqpChannel, }, }, - queueName: queueName, - channel: msgChannel, + config: DefaultQueueConfig(), + channel: msgChannel, } _, err := amqpConsumer.Pull(ctx) @@ -172,8 +177,8 @@ func TestPullChannelClosed(t *testing.T) { amqpChannelConsumer: mockAmqpChannel, }, }, - queueName: queueName, - channel: msgChannel, + config: DefaultQueueConfig(), + channel: msgChannel, } _, err := amqpConsumer.Pull(context.Background()) @@ -222,7 +227,7 @@ func TestPullNoChannel(t *testing.T) { amqpChannel: tt.channel, amqpConnection: mockAmqpConnection, }, - queueName: queueName, + config: DefaultQueueConfig(), } msg, err := amqpConsumer.Pull(context.Background()) @@ -252,7 +257,7 @@ func TestPullNoChannelFailure(t *testing.T) { amqpChannel: nil, amqpConnection: mockAmqpConnection, }, - queueName: queueName, + config: DefaultQueueConfig(), } _, err := amqpConsumer.Pull(context.Background()) @@ -301,7 +306,7 @@ func TestPullNoConnection(t *testing.T) { return mockAmqpConnection, nil }, }, - queueName: queueName, + config: DefaultQueueConfig(), } msg, err := amqpConsumer.Pull(context.Background()) @@ -327,7 +332,7 @@ func TestPullNoConnectionFailure(t *testing.T) { return nil, errors.New("connection error") }, }, - queueName: queueName, + config: DefaultQueueConfig(), } _, err := amqpConsumer.Pull(context.Background()) @@ -336,11 +341,11 @@ func TestPullNoConnectionFailure(t *testing.T) { assert.ErrorContains(t, err, "connection error") } -func TestPullQueueDeclare(t *testing.T) { +func TestPullQueueDeclareAndBind(t *testing.T) { /* arrange: create a consumer with no amqp channel act: pull a message from the queue - assert: channel is established and queue is declared + assert: channel is established, exchange and queue declared, and queue bound */ msgChannel := make(chan amqp.Delivery, 1) msgChannel <- amqp.Delivery{Body: []byte("TestMessage")} @@ -352,43 +357,55 @@ func TestPullQueueDeclare(t *testing.T) { amqpChannelConsumer: mockAmqpChannelConsumer, returnChannelOnSetup: true, } + config := DefaultQueueConfig() amqpConsumer := &AmqpConsumer{ client: &Client{ amqpChannel: nil, amqpConnection: mockAmqpConnection, }, - queueName: queueName, + config: config, } _, err := amqpConsumer.Pull(context.Background()) assert.NoError(t, err) assert.True(t, mockAmqpChannelConsumer.confirmMode, "expected channel to be in confirm mode") - assert.Equal(t, queueName, mockAmqpChannelConsumer.queueName, "expected queue name to be "+queueName) + assert.Equal(t, config.ExchangeName, mockAmqpChannelConsumer.exchangeName, "expected exchange name to be "+config.ExchangeName) + assert.Equal(t, config.QueueName, mockAmqpChannelConsumer.queueName, "expected queue name to be "+config.QueueName) assert.True(t, mockAmqpChannelConsumer.queueDurable, "expected queue to be durable") + assert.Equal(t, config.QueueName, mockAmqpChannelConsumer.boundQueue, "expected queue to be bound") + assert.Equal(t, config.ExchangeName, mockAmqpChannelConsumer.boundExchange, "expected queue to be bound to exchange") + assert.Equal(t, config.RoutingKey, mockAmqpChannelConsumer.boundRoutingKey, "expected routing key to be "+config.RoutingKey) } -func TestPullQueueDeclareFailure(t *testing.T) { +func TestPullDeclarationFailure(t *testing.T) { /* - arrange: create a consumer with no amqp channel where the queue declare fails + arrange: create a consumer with no amqp channel where a declaration fails act: pull a message from the queue assert: pull returns an error */ tests := []struct { name string - queueName string mockAmqpConnection *MockAmqpConnectionConsumer errMsg string }{ + { + name: "exchange declare error", + mockAmqpConnection: &MockAmqpConnectionConsumer{exchangeDeclareError: true}, + errMsg: "exchange declare error", + }, { name: "queue declare error", - queueName: queueWithDeclareError, - mockAmqpConnection: &MockAmqpConnectionConsumer{}, + mockAmqpConnection: &MockAmqpConnectionConsumer{queueDeclareError: true}, errMsg: "queue declare error", }, { - name: "confirm error", - queueName: queueName, + name: "queue bind error", + mockAmqpConnection: &MockAmqpConnectionConsumer{bindError: true}, + errMsg: "bind error", + }, + { + name: "confirm error", mockAmqpConnection: &MockAmqpConnectionConsumer{ confirmModeError: true, }, @@ -403,12 +420,12 @@ func TestPullQueueDeclareFailure(t *testing.T) { amqpChannel: nil, amqpConnection: tt.mockAmqpConnection, }, - queueName: tt.queueName, + config: DefaultQueueConfig(), } _, err := amqpConsumer.Pull(context.Background()) - assert.Error(t, err, "expected error when queue declare fails") + assert.Error(t, err, "expected error when declaration fails") assert.ErrorContains(t, err, tt.errMsg) }) } @@ -421,9 +438,7 @@ func TestPullQosFailure(t *testing.T) { assert: pull returns an error */ mockAmqpChannelConsumer := &MockAmqpChannelConsumer{ - MockAmqpChannel: MockAmqpChannel{ - queueName: queueWithQosError, - }, + qosError: true, } mockAmqpConnection := &MockAmqpConnectionConsumer{ amqpChannelConsumer: mockAmqpChannelConsumer, @@ -434,7 +449,7 @@ func TestPullQosFailure(t *testing.T) { amqpChannel: nil, amqpConnection: mockAmqpConnection, }, - queueName: queueWithQosError, + config: DefaultQueueConfig(), } _, err := amqpConsumer.Pull(context.Background()) @@ -449,16 +464,19 @@ func TestPullConsumeFailure(t *testing.T) { act: pull a message from the queue assert: pull returns an error */ + mockAmqpChannelConsumer := &MockAmqpChannelConsumer{ + consumeError: true, + } mockAmqpConnection := &MockAmqpConnectionConsumer{ returnChannelOnSetup: true, - amqpChannelConsumer: &MockAmqpChannelConsumer{}, + amqpChannelConsumer: mockAmqpChannelConsumer, } amqpConsumer := &AmqpConsumer{ client: &Client{ amqpChannel: nil, amqpConnection: mockAmqpConnection, }, - queueName: queueWithConsumeError, + config: DefaultQueueConfig(), } _, err := amqpConsumer.Pull(context.Background()) @@ -474,13 +492,13 @@ func TestNewAmqpConsumer(t *testing.T) { assert: consumer is properly initialized */ uri := "amqp://guest:guest@localhost:5672/" - queueName := "test-queue" + config := DefaultQueueConfig() - consumer := NewAmqpConsumer(uri, queueName) + consumer := NewAmqpConsumer(uri, config) assert.NotNil(t, consumer) - assert.Equal(t, queueName, consumer.queueName) assert.NotNil(t, consumer.client) assert.Equal(t, uri, consumer.client.uri) assert.NotNil(t, consumer.client.connectFunc) + assert.Equal(t, config, consumer.config) } diff --git a/internal/queue/mock_test.go b/internal/queue/mock_test.go new file mode 100644 index 00000000..6b24650e --- /dev/null +++ b/internal/queue/mock_test.go @@ -0,0 +1,190 @@ +/* + * Copyright 2026 Canonical Ltd. + * See LICENSE file for licensing details. + */ + +package queue + +import ( + "errors" + + amqp "github.com/rabbitmq/amqp091-go" +) + +// MockAmqpChannel implements amqpChannel for unit testing. +type MockAmqpChannel struct { + // Message tracking + msgs [][]byte + headers []map[string]interface{} + + // State + isclosed bool + isClosedRet bool + closeCalls int + closeErr error + confirmMode bool + + // Declaration tracking + queueName string + exchangeName string + queueDurable bool + + // Binding tracking + boundQueue string + boundExchange string + boundRoutingKey string + + // Publish behavior + publishError bool + publishNoAck bool + publishHangs bool + + // Error modes + confirmModeError bool + exchangeDeclareError bool + queueDeclareError bool + bindError bool + + // Consumer support + consumeCh <-chan amqp.Delivery + consumeErr error + qosErr error + + // Additional error modes for types_test + queueDeclarePassiveErr error +} + +func (ch *MockAmqpChannel) PublishWithDeferredConfirm(exchange string, key string, _, _ bool, msg amqp.Publishing) (confirmation, error) { + if ch.publishError { + return nil, errors.New("publish error") + } + ch.msgs = append(ch.msgs, msg.Body) + ch.headers = append(ch.headers, msg.Headers) + + doneCh := make(chan struct{}, 1) + + if !ch.publishHangs { + doneCh <- struct{}{} + } + + ack := !ch.publishNoAck + conf := &MockConfirmation{ + done: doneCh, + ack: ack, + } + return conf, nil +} + +func (ch *MockAmqpChannel) IsClosed() bool { + return ch.isclosed || ch.isClosedRet +} + +func (ch *MockAmqpChannel) Confirm(_ bool) error { + if ch.confirmModeError { + return errors.New("confirm error") + } + ch.confirmMode = true + return nil +} + +func (ch *MockAmqpChannel) ExchangeDeclare(name string, _ string, _, _, _, _ bool, _ amqp.Table) error { + if ch.exchangeDeclareError { + return errors.New("exchange declare error") + } + ch.exchangeName = name + return nil +} + +func (ch *MockAmqpChannel) QueueDeclare(name string, durable, _, _, _ bool, _ amqp.Table) (amqp.Queue, error) { + if ch.queueDeclareError { + return amqp.Queue{}, errors.New("queue declare error") + } + ch.queueName = name + ch.queueDurable = durable + return amqp.Queue{Name: name}, nil +} + +func (ch *MockAmqpChannel) QueueDeclarePassive(name string, durable, autoDelete, exclusive, noWait bool, args amqp.Table) (amqp.Queue, error) { + return amqp.Queue{Name: name}, ch.queueDeclarePassiveErr +} + +func (ch *MockAmqpChannel) QueueBind(name, key, exchange string, noWait bool, args amqp.Table) error { + if ch.bindError { + return errors.New("bind error") + } + ch.boundQueue = name + ch.boundRoutingKey = key + ch.boundExchange = exchange + return nil +} + +func (ch *MockAmqpChannel) Consume(queue, consumer string, autoAck, exclusive, noLocal, noWait bool, args amqp.Table) (<-chan amqp.Delivery, error) { + return ch.consumeCh, ch.consumeErr +} + +func (ch *MockAmqpChannel) Qos(prefetchCount, prefetchSize int, global bool) error { + return ch.qosErr +} + +func (ch *MockAmqpChannel) Close() error { + ch.closeCalls++ + ch.isclosed = true + return ch.closeErr +} + +// MockConfirmation implements confirmation for unit testing. +type MockConfirmation struct { + done <-chan struct{} + ack bool +} + +func (c *MockConfirmation) Done() <-chan struct{} { + return c.done +} + +func (c *MockConfirmation) Acked() bool { + return c.ack +} + +// MockAmqpConnection implements amqpConnection for unit testing. +type MockAmqpConnection struct { + channelCalls int + amqpChannel *MockAmqpChannel + channel amqpChannel + channelErr error + isclosed bool + isClosedRet bool + closeCalls int + closeErr error + errMode bool + confirmModeError bool + exchangeDeclareError bool +} + +func (m *MockAmqpConnection) Channel() (amqpChannel, error) { + if m.errMode { + return nil, errors.New("failed to open channel") + } + if m.channelErr != nil { + return nil, m.channelErr + } + if m.channel != nil { + return m.channel, nil + } + m.channelCalls++ + m.amqpChannel = &MockAmqpChannel{ + confirmModeError: m.confirmModeError, + exchangeDeclareError: m.exchangeDeclareError, + } + return m.amqpChannel, nil +} + +func (m *MockAmqpConnection) IsClosed() bool { + return m.isclosed || m.isClosedRet +} + +func (m *MockAmqpConnection) Close() error { + m.closeCalls++ + m.isclosed = true + return m.closeErr +} diff --git a/internal/queue/producer.go b/internal/queue/producer.go index 078d0058..cfdda8c8 100644 --- a/internal/queue/producer.go +++ b/internal/queue/producer.go @@ -40,7 +40,12 @@ func (p *AmqpProducer) resetConnectionOrChannelIfNecessary() error { } if p.client.amqpChannel == nil || p.client.amqpChannel.IsClosed() { - err := p.client.resetChannel(p.queueName, false) + err := p.client.resetChannel() + if err != nil { + return err + } + + err = p.client.declareExchange(p.config.ExchangeName) if err != nil { return err } @@ -63,10 +68,10 @@ func waitForMsgConfirmation(ctx context.Context, confirmation confirmation) erro func (p *AmqpProducer) publishMsg(msg []byte, headers map[string]interface{}) (confirmation, error) { confirmation, err := p.client.amqpChannel.PublishWithDeferredConfirm( - "", // exchange - p.queueName, // routing key - false, // mandatory - false, // immediate + p.config.ExchangeName, // exchange + p.config.RoutingKey, // routing key + false, // mandatory + false, // immediate amqp.Publishing{ ContentType: "application/json", Body: msg, @@ -80,13 +85,13 @@ func (p *AmqpProducer) publishMsg(msg []byte, headers map[string]interface{}) (c return confirmation, nil } -func NewAmqpProducer(uri string, queueName string) *AmqpProducer { +func NewAmqpProducer(uri string, config QueueConfig) *AmqpProducer { return &AmqpProducer{ client: &Client{ uri: uri, connectFunc: amqpConnect, }, - queueName: queueName, + config: config, } } diff --git a/internal/queue/producer_test.go b/internal/queue/producer_test.go index 4339bfb0..0f036c77 100644 --- a/internal/queue/producer_test.go +++ b/internal/queue/producer_test.go @@ -10,130 +10,12 @@ import ( "errors" "testing" - amqp "github.com/rabbitmq/amqp091-go" "github.com/stretchr/testify/assert" ) -const queueName = "test-queue" -const queueWithPublishNoAck = "queue-with-publish-no-ack" -const queueWithPublishError = "queue-with-publish-error" -const queueWithPublishHangs = "queue-with-publish-hangs" -const queueWithDeclareError = "queue-with-declare-error" - -type MockAmqpChannel struct { - msgs [][]byte - headers []map[string]interface{} - isclosed bool - confirmMode bool - queueName string - queueDurable bool - confirmModeError bool -} - -func (ch *MockAmqpChannel) PublishWithDeferredConfirm(_ string, key string, _, _ bool, msg amqp.Publishing) (confirmation, error) { - - if key == queueWithPublishError { - return nil, errors.New("publish error") - } - ch.msgs = append(ch.msgs, msg.Body) - ch.headers = append(ch.headers, msg.Headers) - - done_ch := make(chan struct{}, 1) - - if key != queueWithPublishHangs { - done_ch <- struct{}{} - } - - ack := key != queueWithPublishNoAck - confirmation := &MockConfirmation{ - done: done_ch, - ack: ack, - } - return confirmation, nil -} - -func (ch *MockAmqpChannel) IsClosed() bool { - return ch.isclosed -} - -func (ch *MockAmqpChannel) Confirm(_ bool) error { - if ch.confirmModeError { - return errors.New("confirm error") - } - ch.confirmMode = true - return nil -} - -func (ch *MockAmqpChannel) QueueDeclare(name string, durable, _, _, _ bool, _ amqp.Table) (amqp.Queue, error) { - if name == queueWithDeclareError { - return amqp.Queue{}, errors.New("queue declare error") - } - ch.queueName = name - ch.queueDurable = durable - return amqp.Queue{}, nil -} - -func (ch *MockAmqpChannel) QueueDeclarePassive(name string, durable, autoDelete, exclusive, noWait bool, args amqp.Table) (amqp.Queue, error) { - return amqp.Queue{}, nil -} - -func (ch *MockAmqpChannel) Consume(queue, consumer string, autoAck, exclusive, noLocal, noWait bool, args amqp.Table) (<-chan amqp.Delivery, error) { - return nil, nil -} - -func (ch *MockAmqpChannel) Qos(prefetchCount, prefetchSize int, global bool) error { - return nil -} - -func (ch *MockAmqpChannel) Close() error { - ch.isclosed = true - return nil -} - -type MockConfirmation struct { - done <-chan struct{} - ack bool -} - -func (c *MockConfirmation) Done() <-chan struct{} { - return c.done -} - -func (c *MockConfirmation) Acked() bool { - return c.ack -} - -type MockAmqpConnection struct { - channelCalls int - amqpChannel *MockAmqpChannel - isclosed bool - errMode bool - confirmModeError bool -} - -func (m *MockAmqpConnection) Channel() (amqpChannel, error) { - if m.errMode { - return nil, errors.New("failed to open channel") - } - m.channelCalls++ - m.amqpChannel = &MockAmqpChannel{ - confirmModeError: m.confirmModeError, - } - return m.amqpChannel, nil -} - -func (m *MockAmqpConnection) IsClosed() bool { - return m.isclosed -} - -func (m *MockAmqpConnection) Close() error { - m.isclosed = true - return nil -} - func TestPush(t *testing.T) { /* - arrange: create a queue with a fake amqp connection + arrange: create a producer with a fake amqp connection act: push a message to the queue assert: message was published to the amqp channel */ @@ -145,7 +27,7 @@ func TestPush(t *testing.T) { amqpChannel: mockAmqpChannel, }, }, - queueName: queueName, + config: DefaultQueueConfig(), } headers := map[string]interface{}{"header1": "value1"} @@ -157,21 +39,21 @@ func TestPush(t *testing.T) { func TestPushFailure(t *testing.T) { /* - arrange: create a queue with a fake confirm handler that always fails + arrange: create a producer with a fake confirm handler that always fails act: push a message to the queue that will fail to publish assert: the push return an error */ tests := []struct { - name string - queueName string - context context.Context - errMsg string + name string + mockChannel *MockAmqpChannel + context context.Context + errMsg string }{ { - name: "message is not acked", - queueName: queueWithPublishNoAck, - errMsg: "confirmation not acknowledged", - context: context.Background(), + name: "message is not acked", + mockChannel: &MockAmqpChannel{publishNoAck: true}, + errMsg: "confirmation not acknowledged", + context: context.Background(), }, { name: "context is done", @@ -180,28 +62,27 @@ func TestPushFailure(t *testing.T) { cancel() return ctx }(), - queueName: queueWithPublishHangs, - errMsg: "context canceled", + mockChannel: &MockAmqpChannel{publishHangs: true}, + errMsg: "context canceled", }, { - name: "publish returns error", - queueName: queueWithPublishError, - errMsg: "publish error", - context: context.Background(), + name: "publish returns error", + mockChannel: &MockAmqpChannel{publishError: true}, + errMsg: "publish error", + context: context.Background(), }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - mockAmqpChannel := &MockAmqpChannel{} amqpProducer := &AmqpProducer{ client: &Client{ - amqpChannel: mockAmqpChannel, + amqpChannel: tt.mockChannel, amqpConnection: &MockAmqpConnection{ - amqpChannel: mockAmqpChannel, + amqpChannel: tt.mockChannel, }, }, - queueName: tt.queueName, + config: DefaultQueueConfig(), } err := amqpProducer.Push(tt.context, nil, []byte("TestMessage")) @@ -241,6 +122,7 @@ func TestPushNoChannel(t *testing.T) { amqpChannel: tt.channel, amqpConnection: mockAmqpConnection, }, + config: DefaultQueueConfig(), } amqpProducer.Push(context.Background(), nil, []byte("TestMessage")) @@ -265,6 +147,7 @@ func TestPushNoChannelFailure(t *testing.T) { amqpChannel: nil, amqpConnection: mockAmqpConnection, }, + config: DefaultQueueConfig(), } err := amqpProducer.Push(context.Background(), nil, []byte("TestMessage")) assert.Error(t, err, "expected error when channel fails to open") @@ -305,6 +188,7 @@ func TestPushNoConnection(t *testing.T) { return mockAmqpConnection, nil }, }, + config: DefaultQueueConfig(), } amqpProducer.Push(context.Background(), nil, []byte("TestMessage")) @@ -328,6 +212,7 @@ func TestPushNoConnectionFailure(t *testing.T) { return nil, errors.New("connection error") }, }, + config: DefaultQueueConfig(), } err := amqpProducer.Push(context.Background(), nil, []byte("TestMessage")) @@ -335,49 +220,46 @@ func TestPushNoConnectionFailure(t *testing.T) { assert.ErrorContains(t, err, "connection error") } -func TestPushQueueDeclare(t *testing.T) { +func TestPushExchangeDeclare(t *testing.T) { /* - arrange: create a queue with no amqp channel + arrange: create a producer with no amqp channel act: push a message to the queue - assert: channel with confirm mode is established and queue is declared + assert: channel with confirm mode is established and exchange is declared with config name */ mockAmqpConnection := &MockAmqpConnection{} + config := DefaultQueueConfig() amqpProducer := &AmqpProducer{ client: &Client{ amqpChannel: nil, amqpConnection: mockAmqpConnection, }, - queueName: queueName, + config: config, } amqpProducer.Push(context.Background(), nil, []byte("TestMessage")) - assert.Equal(t, mockAmqpConnection.amqpChannel.confirmMode, true, "expected channel to be in confirm mode") - assert.Equal(t, mockAmqpConnection.amqpChannel.queueName, queueName, "expected queue name to be "+queueName) - assert.Equal(t, mockAmqpConnection.amqpChannel.queueDurable, true, "expected queue to be durable") + assert.True(t, mockAmqpConnection.amqpChannel.confirmMode, "expected channel to be in confirm mode") + assert.Equal(t, config.ExchangeName, mockAmqpConnection.amqpChannel.exchangeName, "expected exchange name to be "+config.ExchangeName) } -func TestPushQueueDeclareFailure(t *testing.T) { +func TestPushExchangeDeclareFailure(t *testing.T) { /* - arrange: create a queue with no amqp channel where the queue declare fails + arrange: create a producer with no amqp channel where the exchange declare fails act: push a message to the queue assert: push returns an error */ tests := []struct { name string - queueName string mockAmqpConnection *MockAmqpConnection errMsg string }{ { - name: "queue declare error", - queueName: queueWithDeclareError, - mockAmqpConnection: &MockAmqpConnection{}, - errMsg: "queue declare error", + name: "exchange declare error", + mockAmqpConnection: &MockAmqpConnection{exchangeDeclareError: true}, + errMsg: "exchange declare error", }, { - name: "confirm error", - queueName: queueName, + name: "confirm error", mockAmqpConnection: &MockAmqpConnection{ confirmModeError: true, }, @@ -392,12 +274,12 @@ func TestPushQueueDeclareFailure(t *testing.T) { amqpChannel: nil, amqpConnection: tt.mockAmqpConnection, }, - queueName: tt.queueName, + config: DefaultQueueConfig(), } err := amqpProducer.Push(context.Background(), nil, []byte("TestMessage")) - assert.Error(t, err, "expected error when queue declare fails") + assert.Error(t, err, "expected error when exchange declare fails") assert.ErrorContains(t, err, tt.errMsg) }) } diff --git a/internal/queue/types.go b/internal/queue/types.go index 63614750..6b52d0d1 100644 --- a/internal/queue/types.go +++ b/internal/queue/types.go @@ -34,8 +34,8 @@ func (c *Client) resetConnection() error { return nil } -// resetChannel opens a new AMQP channel and declares the queue. -func (c *Client) resetChannel(queue string, passive bool) error { +// resetChannel opens a new AMQP channel +func (c *Client) resetChannel() error { if c.amqpChannel != nil && !c.amqpChannel.IsClosed() { return nil } @@ -51,29 +51,51 @@ func (c *Client) resetChannel(queue string, passive bool) error { return fmt.Errorf("failed to put channel in confirm mode: %w", err) } - if passive { - _, err = ch.QueueDeclarePassive( - queue, // queueName - true, // durable - false, // delete when unused - false, // exclusive - false, // no-wait - nil, // arguments - ) - } else { - _, err = ch.QueueDeclare( - queue, // queueName - true, // durable - false, // delete when unused - false, // exclusive - false, // no-wait - nil, // arguments - ) + return nil +} + +func (c *Client) declareExchange(exchangeName string) error { + err := c.amqpChannel.ExchangeDeclare( + exchangeName, + "direct", // use direct exchange + true, // durable - don't delete on broker restart + false, // autoDelete - don't delete when no longer in use + false, // internal set to false - be suitable for publishing + false, // wait for confirmation + nil, // no additional arguments + ) + if err != nil { + return fmt.Errorf("failed to declare AMQP exchange: %w", err) } + return nil +} + +func (c *Client) declareQueue(queueName string) error { + _, err := c.amqpChannel.QueueDeclare( + queueName, + true, // durable + false, // delete when unused + false, // exclusive + false, // no-wait + nil, // arguments + ) if err != nil { return fmt.Errorf("failed to declare AMQP queue: %w", err) } + return nil +} +func (c *Client) bindQueue(queueName, routingKey, exchangeName string) error { + err := c.amqpChannel.QueueBind( + queueName, + routingKey, + exchangeName, + false, // no-wait + nil, // arguments + ) + if err != nil { + return fmt.Errorf("failed to bind AMQP queue: %w", err) + } return nil } @@ -100,8 +122,8 @@ func (c *Client) Close() error { } type AmqpProducer struct { - client *Client - queueName string + client *Client + config QueueConfig } type Producer interface { @@ -113,8 +135,10 @@ type amqpChannel interface { IsClosed() bool Close() error Confirm(noWait bool) error + ExchangeDeclare(name string, kind string, durable, autoDelete, internal, noWait bool, args amqp.Table) error QueueDeclare(name string, durable, autoDelete, exclusive, noWait bool, args amqp.Table) (amqp.Queue, error) QueueDeclarePassive(name string, durable, autoDelete, exclusive, noWait bool, args amqp.Table) (amqp.Queue, error) + QueueBind(name, key, exchange string, noWait bool, args amqp.Table) error Consume(queue, consumer string, autoAck, exclusive, noLocal, noWait bool, args amqp.Table) (<-chan amqp.Delivery, error) Qos(prefetchCount, prefetchSize int, global bool) error } @@ -155,6 +179,10 @@ func (ch *amqpChannelWrapper) Qos(prefetchCount, prefetchSize int, global bool) return ch.Channel.Qos(prefetchCount, prefetchSize, global) } +func (ch *amqpChannelWrapper) QueueBind(name, key, exchange string, noWait bool, args amqp.Table) error { + return ch.Channel.QueueBind(name, key, exchange, noWait, args) +} + type Consumer interface { Pull(ctx context.Context) (amqp.Delivery, error) Close() error diff --git a/internal/queue/types_test.go b/internal/queue/types_test.go index c515bf71..0cc8abfb 100644 --- a/internal/queue/types_test.go +++ b/internal/queue/types_test.go @@ -11,63 +11,6 @@ import ( "github.com/stretchr/testify/assert" ) -// fakeChannel implements amqpChannel for unit testing without external deps. -type fakeChannel struct { - closed bool - closeCalls int - closeErr error - isClosedRet bool - - publishErr error - confirmErr error - qosErr error - consumeErr error - declareErr error - declarePassErr error - - consumeCh <-chan amqp.Delivery -} - -func (f *fakeChannel) PublishWithDeferredConfirm(exchange string, key string, mandatory, immediate bool, msg amqp.Publishing) (confirmation, error) { - return nil, f.publishErr -} -func (f *fakeChannel) IsClosed() bool { return f.closed || f.isClosedRet } -func (f *fakeChannel) Close() error { - f.closeCalls++ - f.closed = true - return f.closeErr -} -func (f *fakeChannel) Confirm(noWait bool) error { return f.confirmErr } -func (f *fakeChannel) QueueDeclare(name string, durable, autoDelete, exclusive, noWait bool, args amqp.Table) (amqp.Queue, error) { - return amqp.Queue{Name: name}, f.declareErr -} -func (f *fakeChannel) QueueDeclarePassive(name string, durable, autoDelete, exclusive, noWait bool, args amqp.Table) (amqp.Queue, error) { - return amqp.Queue{Name: name}, f.declarePassErr -} -func (f *fakeChannel) Consume(queue, consumer string, autoAck, exclusive, noLocal, noWait bool, args amqp.Table) (<-chan amqp.Delivery, error) { - return f.consumeCh, f.consumeErr -} -func (f *fakeChannel) Qos(prefetchCount, prefetchSize int, global bool) error { return f.qosErr } - -// fakeConnection implements amqpConnection for unit testing. -type fakeConnection struct { - closed bool - closeCalls int - closeErr error - isClosedRet bool - - channel amqpChannel - chErr error -} - -func (f *fakeConnection) Channel() (amqpChannel, error) { return f.channel, f.chErr } -func (f *fakeConnection) IsClosed() bool { return f.closed || f.isClosedRet } -func (f *fakeConnection) Close() error { - f.closeCalls++ - f.closed = true - return f.closeErr -} - // fakeJobDB implements JobDatabase with no-op methods for consumer tests. type fakeJobDB struct{} @@ -100,8 +43,8 @@ func TestClient_Close_Table(t *testing.T) { { name: "both open ok", client: &Client{ - amqpChannel: &fakeChannel{isClosedRet: false}, - amqpConnection: &fakeConnection{isClosedRet: false}, + amqpChannel: &MockAmqpChannel{isClosedRet: false}, + amqpConnection: &MockAmqpConnection{isClosedRet: false}, }, expectErr: nil, expectChCalls: 1, @@ -112,8 +55,8 @@ func TestClient_Close_Table(t *testing.T) { { name: "channel error first", client: &Client{ - amqpChannel: &fakeChannel{isClosedRet: false, closeErr: chErr}, - amqpConnection: &fakeConnection{isClosedRet: false, closeErr: connErr}, + amqpChannel: &MockAmqpChannel{isClosedRet: false, closeErr: chErr}, + amqpConnection: &MockAmqpConnection{isClosedRet: false, closeErr: connErr}, }, expectErr: chErr, expectChCalls: 1, @@ -122,8 +65,8 @@ func TestClient_Close_Table(t *testing.T) { { name: "connection error only", client: &Client{ - amqpChannel: &fakeChannel{isClosedRet: false}, - amqpConnection: &fakeConnection{isClosedRet: false, closeErr: connErr}, + amqpChannel: &MockAmqpChannel{isClosedRet: false}, + amqpConnection: &MockAmqpConnection{isClosedRet: false, closeErr: connErr}, }, expectErr: connErr, expectChCalls: 1, @@ -141,8 +84,8 @@ func TestClient_Close_Table(t *testing.T) { { name: "already closed resources", client: &Client{ - amqpChannel: &fakeChannel{isClosedRet: true}, - amqpConnection: &fakeConnection{isClosedRet: true}, + amqpChannel: &MockAmqpChannel{isClosedRet: true}, + amqpConnection: &MockAmqpConnection{isClosedRet: true}, }, expectErr: nil, expectChCalls: 0, @@ -160,16 +103,16 @@ func TestClient_Close_Table(t *testing.T) { assert.NoError(t, err) } - if fc, ok := tt.client.amqpChannel.(*fakeChannel); ok { + if fc, ok := tt.client.amqpChannel.(*MockAmqpChannel); ok { assert.Equal(t, tt.expectChCalls, fc.closeCalls) if tt.expectChClosed { - assert.True(t, fc.closed) + assert.True(t, fc.isclosed) } } - if fconn, ok := tt.client.amqpConnection.(*fakeConnection); ok { + if fconn, ok := tt.client.amqpConnection.(*MockAmqpConnection); ok { assert.Equal(t, tt.expectConnCalls, fconn.closeCalls) if tt.expectConnClose { - assert.True(t, fconn.closed) + assert.True(t, fconn.isclosed) } } }) @@ -184,11 +127,10 @@ func TestAmqpConsumer_Table(t *testing.T) { act: Close the consumer. assert: Check no error and verify close calls. */ - ch := &fakeChannel{isClosedRet: false} - conn := &fakeConnection{isClosedRet: false} + ch := &MockAmqpChannel{isClosedRet: false} + conn := &MockAmqpConnection{isClosedRet: false} cons := &AmqpConsumer{ - queueName: "q", client: &Client{ amqpChannel: ch, amqpConnection: conn, @@ -209,8 +151,8 @@ func TestClient_Close_Idempotent(t *testing.T) { act: Close the client twice. assert: First close should close both, second close should be a no-op. */ - ch := &fakeChannel{isClosedRet: false} - conn := &fakeConnection{isClosedRet: false} + ch := &MockAmqpChannel{isClosedRet: false} + conn := &MockAmqpConnection{isClosedRet: false} c := &Client{ amqpChannel: ch,