From 34a517662797768207cbe9d6ad4b857113642afa Mon Sep 17 00:00:00 2001
From: thijsheijden <hi@thijsheijden.nl>
Date: Mon, 19 Apr 2021 19:25:04 +0200
Subject: [PATCH] Improved mock broker
Mock producer can now be made. Producer can publish messages. Messages
can be retrieved from the mock broker.
---
internal/drivers/brokerdriver/interface.go | 2 +-
internal/drivers/brokerdriver/mock/broker.go | 20 +++++++++++---
.../drivers/brokerdriver/mock/producer.go | 26 +++++++++++++++++++
internal/drivers/brokerdriver/producer.go | 8 +++---
4 files changed, 48 insertions(+), 8 deletions(-)
create mode 100644 internal/drivers/brokerdriver/mock/producer.go
diff --git a/internal/drivers/brokerdriver/interface.go b/internal/drivers/brokerdriver/interface.go
index 3aacf64..3dc54da 100644
--- a/internal/drivers/brokerdriver/interface.go
+++ b/internal/drivers/brokerdriver/interface.go
@@ -20,5 +20,5 @@ type Consumer interface {
// A Producer belongs to a broker and publishes messages to a queue
type Producer interface {
- PublishMessage(body *[]byte, queueID *string, headers *amqp.Table)
+ PublishMessage(body *[]byte, routingKey *string, headers *amqp.Table)
}
diff --git a/internal/drivers/brokerdriver/mock/broker.go b/internal/drivers/brokerdriver/mock/broker.go
index 8483d8f..80f1c1c 100644
--- a/internal/drivers/brokerdriver/mock/broker.go
+++ b/internal/drivers/brokerdriver/mock/broker.go
@@ -8,18 +8,32 @@ import (
// Driver is mock gateway
type Driver struct {
gateway brokeradapter.GatewayInterface
+
+ // Mock messages that are published by producers on this broker
+ // Key is the routing key
+ // Value is a slice of messages, in order of being sent 'first -> last'
+ Messages map[string][]brokeradapter.Message
}
-// CreateBroker is a creates a mock driver
+// CreateBroker creates a broker driver (mock)
func CreateBroker(gateway brokeradapter.GatewayInterface) *Driver {
return &Driver{
- gateway: gateway,
+ gateway: gateway,
+ Messages: make(map[string][]brokeradapter.Message),
}
}
-// CreateConsumer creates a mock consumer
+// CreateConsumer creates a consumer (mock)
func (d *Driver) CreateConsumer() brokerdriver.Consumer {
return &Consumer{
broker: d,
}
}
+
+// CreateProducer creates a producer (mock)
+func (d *Driver) CreateProducer() brokerdriver.Producer {
+ return &Producer{
+ broker: d,
+ exchange: "ui-direct-exchange", // This is the only exchange this service produces to
+ }
+}
diff --git a/internal/drivers/brokerdriver/mock/producer.go b/internal/drivers/brokerdriver/mock/producer.go
new file mode 100644
index 0000000..d92435b
--- /dev/null
+++ b/internal/drivers/brokerdriver/mock/producer.go
@@ -0,0 +1,26 @@
+package mockbrokerdriver
+
+import (
+ "query-service/internal/adapters/brokeradapter"
+
+ "github.com/streadway/amqp"
+)
+
+type Producer struct {
+ broker *Driver
+
+ // The exchange this producer is connected to
+ exchange string
+}
+
+// PublishMessage publishes a message to the given queue
+func (p *Producer) PublishMessage(body *[]byte, routingKey *string, headers *amqp.Table) {
+ // Create the message
+ msg := brokeradapter.Message{
+ Headers: *headers,
+ Body: *body,
+ }
+
+ // Append the message to the list
+ p.broker.Messages[*routingKey] = append(p.broker.Messages[*routingKey], msg)
+}
diff --git a/internal/drivers/brokerdriver/producer.go b/internal/drivers/brokerdriver/producer.go
index e037aac..513e43d 100644
--- a/internal/drivers/brokerdriver/producer.go
+++ b/internal/drivers/brokerdriver/producer.go
@@ -14,10 +14,10 @@ type AliceProducer struct {
producer alice.Producer
}
-// PublishMessage will publish a message to the specified queue id
-func (ap *AliceProducer) PublishMessage(body *[]byte, queueID *string, headers *amqp.Table) {
+// PublishMessage will publish a message to the specified queue id (mock)
+func (ap *AliceProducer) PublishMessage(body *[]byte, routingKey *string, headers *amqp.Table) {
sessionID := (*headers)["sessionID"]
- logger.Log(fmt.Sprintf("Publishing message to queue %v, for session %v", *queueID, sessionID))
+ logger.Log(fmt.Sprintf("Publishing message to queue %v, for session %v", *routingKey, sessionID))
- ap.producer.PublishMessage(*body, queueID, headers)
+ ap.producer.PublishMessage(*body, routingKey, headers)
}
--
GitLab