feat: make Connect method idempotent across multiple extensions

Author: juicycleff

extensions/grpc/server.go

@@ -52,12 +52,13 @@ func (s *grpcServer) RegisterService(desc *grpc.ServiceDesc, impl interface{}) e
 	return nil
 }
 
+// Start is idempotent — calling it on an already-running server is a no-op.
 func (s *grpcServer) Start(ctx context.Context, addr string) error {
 	// Check if already running
 	s.mu.RLock()
 	if s.running {
 		s.mu.RUnlock()
-		return ErrAlreadyStarted
+		return nil
 	}
 	s.mu.RUnlock()
 
@@ -70,7 +71,7 @@ func (s *grpcServer) Start(ctx context.Context, addr string) error {
 
 	// Double-check running status after acquiring write lock
 	if s.running {
-		return ErrAlreadyStarted
+		return nil
 	}
 
 	// Create server with options

extensions/mqtt/client.go

@@ -46,7 +46,7 @@ func (c *mqttClient) Connect(ctx context.Context) error {
 	defer c.mu.Unlock()
 
 	if c.client != nil && c.client.IsConnected() {
-		return fmt.Errorf("already connected")
+		return nil
 	}
 
 	// Create client options

extensions/mqtt/extension_test.go

@@ -2,13 +2,25 @@ package mqtt
 
 import (
 	"context"
+	"net"
 	"testing"
+	"time"
 
 	"github.com/xraph/confy"
 	"github.com/xraph/forge"
 	"github.com/xraph/forge/internal/logger"
 )
 
+// requireBroker skips the test if no MQTT broker is reachable at localhost:1883.
+func requireBroker(t *testing.T) {
+	t.Helper()
+	conn, err := net.DialTimeout("tcp", "localhost:1883", 500*time.Millisecond)
+	if err != nil {
+		t.Skip("skipping: MQTT broker not available at localhost:1883")
+	}
+	conn.Close()
+}
+
 func TestNewExtension(t *testing.T) {
 	ext := NewExtension()
 	if ext == nil {
@@ -64,6 +76,8 @@ func TestNewExtensionWithConfig(t *testing.T) {
 }
 
 func TestExtensionRegister(t *testing.T) {
+	requireBroker(t)
+
 	testApp := createTestApp(t)
 	ext := NewExtension(
 		WithBroker("tcp://localhost:1883"),
@@ -112,6 +126,8 @@ func TestExtensionRegisterRequireConfigFails(t *testing.T) {
 }
 
 func TestExtensionClient(t *testing.T) {
+	requireBroker(t)
+
 	testApp := createTestApp(t)
 	ext := NewExtension(
 		WithBroker("tcp://localhost:1883"),

extensions/queue/backends_test.go

@@ -2,13 +2,25 @@ package queue
 
 import (
 	"context"
+	"net"
 	"testing"
+	"time"
 
 	"github.com/xraph/forge"
 	"github.com/xraph/forge/errors"
 	"github.com/xraph/forge/internal/logger"
 )
 
+// requireService skips the test if the given TCP address is not reachable.
+func requireService(t *testing.T, addr, name string) {
+	t.Helper()
+	conn, err := net.DialTimeout("tcp", addr, 500*time.Millisecond)
+	if err != nil {
+		t.Skipf("skipping: %s not available at %s", name, addr)
+	}
+	conn.Close()
+}
+
 // Test all backend constructors and basic error paths
 
 func TestRedisQueue_Constructor(t *testing.T) {
@@ -363,6 +375,8 @@ func TestNATSQueue_NotConnectedErrors(t *testing.T) {
 // Test extension with different drivers
 
 func TestExtension_RedisDriver(t *testing.T) {
+	requireService(t, "localhost:6379", "Redis")
+
 	app := forge.New(
 		forge.WithAppName("test-app"),
 		forge.WithAppVersion("1.0.0"),
@@ -392,6 +406,8 @@ func TestExtension_RedisDriver(t *testing.T) {
 }
 
 func TestExtension_RabbitMQDriver(t *testing.T) {
+	requireService(t, "localhost:5672", "RabbitMQ")
+
 	app := forge.New(
 		forge.WithAppName("test-app"),
 		forge.WithAppVersion("1.0.0"),
@@ -421,6 +437,8 @@ func TestExtension_RabbitMQDriver(t *testing.T) {
 }
 
 func TestExtension_NATSDriver(t *testing.T) {
+	requireService(t, "localhost:4222", "NATS")
+
 	app := forge.New(
 		forge.WithAppName("test-app"),
 		forge.WithAppVersion("1.0.0"),

extensions/queue/coverage_test.go

@@ -27,10 +27,10 @@ func TestInMemoryQueue_DoubleConnect(t *testing.T) {
 		t.Fatalf("Connect() error = %v", err)
 	}
 
-	// Second connect should return error
+	// Connect is idempotent — second call should succeed
 	err = q.Connect(ctx)
-	if !errors.Is(err, ErrAlreadyConnected) {
-		t.Errorf("Connect() second time error = %v, want ErrAlreadyConnected", err)
+	if err != nil {
+		t.Errorf("Connect() second time error = %v, want nil", err)
 	}
 }
 

extensions/queue/inmemory.go

@@ -51,12 +51,13 @@ func NewInMemoryQueue(config Config, logger forge.Logger, metrics forge.Metrics)
 	}
 }
 
+// Connect is idempotent — calling it on an already-connected instance is a no-op.
 func (q *InMemoryQueue) Connect(ctx context.Context) error {
 	q.mu.Lock()
 	defer q.mu.Unlock()
 
 	if q.connected {
-		return ErrAlreadyConnected
+		return nil
 	}
 
 	q.connected = true

extensions/queue/inmemory_test.go

@@ -28,10 +28,10 @@ func TestInMemoryQueue_Connect(t *testing.T) {
 		t.Fatalf("failed to connect: %v", err)
 	}
 
-	// Try connecting again
+	// Connect is idempotent — second call should succeed
 	err = q.Connect(ctx)
-	if !errors.Is(err, ErrAlreadyConnected) {
-		t.Errorf("expected ErrAlreadyConnected, got %v", err)
+	if err != nil {
+		t.Errorf("expected nil on idempotent connect, got %v", err)
 	}
 }
 

extensions/queue/nats.go

@@ -55,7 +55,7 @@ func (q *NATSQueue) Connect(ctx context.Context) error {
 	defer q.mu.Unlock()
 
 	if q.connected {
-		return ErrAlreadyConnected
+		return nil
 	}
 
 	// Build connection URL

extensions/queue/nats_test.go

@@ -80,10 +80,10 @@ func TestNATSQueue_ConnectDisconnect(t *testing.T) {
 		t.Skipf("Cannot connect to NATS: %v", err)
 	}
 
-	// Test double connect
+	// Connect is idempotent — second call should succeed
 	err = queue.Connect(ctx)
-	if !errors.Is(err, ErrAlreadyConnected) {
-		t.Errorf("Connect() second time should return ErrAlreadyConnected, got %v", err)
+	if err != nil {
+		t.Errorf("Connect() second time should be idempotent, got %v", err)
 	}
 
 	// Test ping

extensions/queue/rabbitmq.go

@@ -53,7 +53,7 @@ func (q *RabbitMQQueue) Connect(ctx context.Context) error {
 	defer q.mu.Unlock()
 
 	if q.connected {
-		return ErrAlreadyConnected
+		return nil
 	}
 
 	// Build connection URL