From 530b2f932bc499e020d1ea13854acc54fd190851 Mon Sep 17 00:00:00 2001
From: Laurent Broudoux <laurent.broudoux@gmail.com>
Date: Mon, 6 Jan 2025 09:34:24 +0100
Subject: [PATCH] feat: #88 Adding MQTT support for mocking Signed-off-by:
 Laurent Broudoux <laurent.broudoux@gmail.com>

---
 ensemble/async/async.go                      | 16 ++++++++++++
 ensemble/async/connection/generic/generic.go | 26 ++++++++++++++++++++
 ensemble/ensemble.go                         |  9 +++++++
 3 files changed, 51 insertions(+)
 create mode 100644 ensemble/async/connection/generic/generic.go

diff --git a/ensemble/async/async.go b/ensemble/async/async.go
index 74598a3..d3284ee 100644
--- a/ensemble/async/async.go
+++ b/ensemble/async/async.go
@@ -22,6 +22,7 @@ import (
 
 	"github.com/testcontainers/testcontainers-go"
 	"github.com/testcontainers/testcontainers-go/wait"
+	"microcks.io/testcontainers-go/ensemble/async/connection/generic"
 	"microcks.io/testcontainers-go/ensemble/async/connection/kafka"
 )
 
@@ -136,6 +137,21 @@ func WithKafkaConnection(connection kafka.Connection) testcontainers.CustomizeRe
 	}
 }
 
+// WithKafkaConnection connects the MicrocksAsyncMinionContainer to a MQTT broker to allow MQTT messages mocking.
+func WithMQTTConnection(connection generic.Connection) testcontainers.CustomizeRequestOption {
+	return func(req *testcontainers.GenericContainerRequest) error {
+		if req.Env == nil {
+			req.Env = make(map[string]string)
+		}
+		req.Env["MQTT_SERVER"] = connection.Server
+		req.Env["MQTT_USERNAME"] = connection.Username
+		req.Env["MQTT_PASSWORD"] = connection.Password
+		addProtocol(req, "MQTT")
+
+		return nil
+	}
+}
+
 // WSMockEndpoint gets the exposed mock endpoints for a WebSocket Service.
 func (container *MicrocksAsyncMinionContainer) WSMockEndpoint(ctx context.Context, service, version, operationName string) (string, error) {
 	// Get the container host.
diff --git a/ensemble/async/connection/generic/generic.go b/ensemble/async/connection/generic/generic.go
new file mode 100644
index 0000000..68f3d9e
--- /dev/null
+++ b/ensemble/async/connection/generic/generic.go
@@ -0,0 +1,26 @@
+/*
+ * Copyright The Microcks Authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package generic
+
+// Connection represents generic message broker connection
+type Connection struct {
+	// Server represents the hostname + port address.
+	Server string
+	// Username for connecting to remote broker.
+	Username string
+	// Password for connecting to remote broker.
+	Password string
+}
diff --git a/ensemble/ensemble.go b/ensemble/ensemble.go
index 34c07f4..1423b73 100644
--- a/ensemble/ensemble.go
+++ b/ensemble/ensemble.go
@@ -24,6 +24,7 @@ import (
 	"microcks.io/go-client"
 	microcks "microcks.io/testcontainers-go"
 	"microcks.io/testcontainers-go/ensemble/async"
+	"microcks.io/testcontainers-go/ensemble/async/connection/generic"
 	"microcks.io/testcontainers-go/ensemble/async/connection/kafka"
 	"microcks.io/testcontainers-go/ensemble/postman"
 )
@@ -284,6 +285,14 @@ func WithKafkaConnection(connection kafka.Connection) Option {
 	}
 }
 
+// WithMQTTConnection configures a connection to a MQTT Broker.
+func WithMQTTConnection(connection generic.Connection) Option {
+	return func(e *MicrocksContainersEnsemble) error {
+		e.asyncMinionContainerOptions.Add(async.WithMQTTConnection(connection))
+		return nil
+	}
+}
+
 // WithSecret creates a new secret.
 func WithSecret(s client.Secret) Option {
 	return func(e *MicrocksContainersEnsemble) error {