feature/abstract-queue-options

pkg/queue/queue.go

@@ -1,5 +1,40 @@
 package queue
 
+import (
+	"fmt"
+)
+
 type QueueInterface interface {
 	Connect() error
 }
+
+type QueueOptions interface {
+	Validate() error
+}
+
+// Queue represents a queue client instance
+type Queue struct {
+	Options QueueOptions
+	Client  QueueInterface
+}
+
+func New(opts QueueOptions, client ...QueueInterface) (*Queue, error) {
+	// If no client provided, create default production client
+	var err error
+	var q QueueInterface
+	if len(client) == 0 {
+		// Type assert to RabbitOptions for creating RabbitMQ client
+		if rabbitOpts, ok := opts.(*RabbitOptions); ok {
+			q, err = NewRabbitMQ(rabbitOpts)
+		} else {
+			return nil, fmt.Errorf("unsupported queue options type")
+		}
+	} else {
+		q, err = client[0], nil
+	}
+
+	return &Queue{
+		Options: opts,
+		Client:  q,
+	}, err
+}

pkg/queue/rabbitmq.go

@@ -28,6 +28,12 @@ type RabbitOptions struct {
 	Exchange        string
 }
 
+// Validate validates the RabbitOptions configuration
+func (r *RabbitOptions) Validate() error {
+	validate := validator.New()
+	return validate.Struct(r)
+}
+
 // RabbitOptionsBuilder provides a fluent interface for building Rabbit options
 type RabbitOptionsBuilder struct {
 	options *RabbitOptions
@@ -111,10 +117,9 @@ type RabbitMQ struct {
 
 // NewRabbitMQ creates a new RabbitMQ with the provided RabbitMQ settings
 func NewRabbitMQ(options *RabbitOptions) (*RabbitMQ, error) {
+
 	// Validate RabbitMQ configuration
-	validate := validator.New()
-	err := validate.Struct(options)
-	if err != nil {
+	if err := options.Validate(); err != nil {
 		return nil, err
 	}
 	// Extract protocol from host if present, otherwise default to amqp://

pkg/queue/rabbitmq_test.go

@@ -325,12 +325,13 @@ func TestRabbitConnectionStringGeneration(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			opts := tt.buildOpts()
-			rabbit, err := NewRabbitMQ(opts)
-
+			queueClient, err := New(opts)
 			if err != nil {
 				t.Fatalf("failed to create RabbitMQ instance: %v", err)
 			}
 
+			rabbit := queueClient.Client.(*RabbitMQ)
+
 			if rabbit.connectionString != tt.expectedConnStr {
 				t.Errorf("expected connection string '%s', got '%s'", tt.expectedConnStr, rabbit.connectionString)
 			}
@@ -373,11 +374,17 @@ func TestRabbitMQIntegration(t *testing.T) {
 			Build()
 
 		// Create RabbitMQ instance
-		rabbit, err := NewRabbitMQ(opts)
+		queueClient, err := New(opts)
 		if err != nil {
 			t.Fatalf("Failed to create RabbitMQ instance: %v", err)
 		}
 
+		// Get the underlying RabbitMQ client
+		rabbit, ok := queueClient.Client.(*RabbitMQ)
+		if !ok {
+			t.Fatal("Failed to assert RabbitMQ client type")
+		}
+
 		// Test connection
 		err = rabbit.Connect()
 		if err != nil {
@@ -422,11 +429,13 @@ func TestRabbitMQIntegration(t *testing.T) {
 			SetExchange(exchange).
 			Build()
 
-		rabbit, err := NewRabbitMQ(opts)
+		queueClient, err := New(opts)
 		if err != nil {
 			t.Fatalf("Failed to create RabbitMQ instance: %v", err)
 		}
 
+		rabbit := queueClient.Client.(*RabbitMQ)
+
 		err = rabbit.Connect()
 		if err != nil {
 			t.Fatalf("Failed to connect to RabbitMQ: %v", err)
@@ -463,11 +472,13 @@ func TestRabbitMQIntegration(t *testing.T) {
 		// Create multiple connections
 		connections := make([]*RabbitMQ, 3)
 		for i := 0; i < 3; i++ {
-			rabbit, err := NewRabbitMQ(opts)
+			queueClient, err := New(opts)
 			if err != nil {
 				t.Fatalf("Failed to create RabbitMQ instance %d: %v", i, err)
 			}
 
+			rabbit := queueClient.Client.(*RabbitMQ)
+
 			err = rabbit.Connect()
 			if err != nil {
 				t.Fatalf("Failed to connect RabbitMQ instance %d: %v", i, err)
@@ -550,11 +561,13 @@ func TestDisasterRecovery(t *testing.T) {
 			SetPassword("guest").
 			Build()
 
-		rabbit, err := NewRabbitMQ(opts)
+		queueClient, err := New(opts)
 		if err != nil {
 			t.Fatalf("failed to create RabbitMQ instance: %v", err)
 		}
 
+		rabbit := queueClient.Client.(*RabbitMQ)
+
 		// Track whether handler was called and with what payload
 		var handlerCalled bool
 		var receivedPayload []byte
@@ -600,11 +613,13 @@ func TestDisasterRecovery(t *testing.T) {
 			SetPassword("guest").
 			Build()
 
-		rabbit, err := NewRabbitMQ(opts)
+		queueClient, err := New(opts)
 		if err != nil {
 			t.Fatalf("failed to create RabbitMQ instance: %v", err)
 		}
 
+		rabbit := queueClient.Client.(*RabbitMQ)
+
 		// Set a handler that returns an error
 		expectedErr := fmt.Errorf("handler error")
 		rabbit.SetDisasterRecoveryHandler(func(payload []byte) error {
@@ -634,11 +649,13 @@ func TestDisasterRecovery(t *testing.T) {
 			SetPassword("guest").
 			Build()
 
-		rabbit, err := NewRabbitMQ(opts)
+		queueClient, err := New(opts)
 		if err != nil {
 			t.Fatalf("failed to create RabbitMQ instance: %v", err)
 		}
 
+		rabbit := queueClient.Client.(*RabbitMQ)
+
 		// Test payload
 		testPayload := []byte(`{"test": "data"}`)
 
@@ -662,11 +679,13 @@ func TestDisasterRecovery(t *testing.T) {
 			SetPassword("guest").
 			Build()
 
-		rabbit, err := NewRabbitMQ(opts)
+		queueClient, err := New(opts)
 		if err != nil {
 			t.Fatalf("failed to create RabbitMQ instance: %v", err)
 		}
 
+		rabbit := queueClient.Client.(*RabbitMQ)
+
 		// Track handler call
 		var handlerCalled bool
 		var receivedPayload []byte
@@ -710,11 +729,13 @@ func TestDisasterRecovery(t *testing.T) {
 			SetPassword("guest").
 			Build()
 
-		rabbit, err := NewRabbitMQ(opts)
+		queueClient, err := New(opts)
 		if err != nil {
 			t.Fatalf("failed to create RabbitMQ instance: %v", err)
 		}
 
+		rabbit := queueClient.Client.(*RabbitMQ)
+
 		// Track handler call
 		var handlerCalled bool
 		var receivedPayload []byte