platform/rabbit/conn_test.go

package rabbit_test

import (
	"os"
	"sync"
	"testing"
	"time"

	"git.k6n.net/mats/platform/rabbit"
)

// rabbitURL returns the RabbitMQ URL for integration tests, or empty if unset.
// Tests check this and skip when no broker is available.
func rabbitURL() string {
	if u := os.Getenv("RABBIT_URL"); u != "" {
		return u
	}
	if u := os.Getenv("RABBIT_HOST"); u != "" {
		return "amqp://" + u
	}
	// Default local dev URL used by docker-compose.
	return "amqp://guest:guest@localhost:5672"
}

// skipWithoutRabbit skips the test if no RabbitMQ broker is reachable.
func skipWithoutRabbit(t *testing.T) string {
	t.Helper()
	url := rabbitURL()
	// Quick probe: try connecting and immediately close.
	conn, err := rabbit.Connect(url, "test-probe")
	if err != nil {
		t.Skipf("RabbitMQ not available at %s: %v", url, err)
	}
	_ = conn.Close()
	return url
}

func TestDial_ConnectAndClose(t *testing.T) {
	url := skipWithoutRabbit(t)

	conn, err := rabbit.Dial(url, "test-conn")
	if err != nil {
		t.Fatalf("Dial failed: %v", err)
	}
	defer conn.Close()

	// The underlying connection should be usable.
	if c := conn.Connection(); c == nil {
		t.Fatal("Connection() returned nil")
	}
	ch, err := conn.Channel()
	if err != nil {
		t.Fatalf("Channel() failed: %v", err)
	}
	if err := ch.Close(); err != nil {
		t.Fatalf("channel close: %v", err)
	}
}

func TestDial_Done(t *testing.T) {
	url := skipWithoutRabbit(t)

	conn, err := rabbit.Dial(url, "test-done")
	if err != nil {
		t.Fatalf("Dial failed: %v", err)
	}

	// Done should NOT be closed initially.
	select {
	case <-conn.Done():
		t.Fatal("Done channel closed before Close()")
	default:
	}

	// After Close, Done should be closed.
	_ = conn.Close()
	select {
	case <-conn.Done():
		// expected
	case <-time.After(time.Second):
		t.Fatal("Done channel not closed after Close()")
	}

	// Double close is safe.
	_ = conn.Close()
}

func TestDial_NotifyOnReconnect(t *testing.T) {
	url := skipWithoutRabbit(t)

	conn, err := rabbit.Dial(url, "test-notify")
	if err != nil {
		t.Fatalf("Dial failed: %v", err)
	}
	defer conn.Close()

	// NotifyOnReconnect should call the function immediately.
	called := make(chan struct{}, 1)
	conn.NotifyOnReconnect(func() {
		called <- struct{}{}
	})

	select {
	case <-called:
		// expected — immediate call
	case <-time.After(time.Second):
		t.Fatal("NotifyOnReconnect was not called immediately")
	}

	// Multiple registrations should each fire immediately.
	var mu sync.Mutex
	count := 0
	conn.NotifyOnReconnect(func() {
		mu.Lock()
		count++
		mu.Unlock()
	})
	conn.NotifyOnReconnect(func() {
		mu.Lock()
		count++
		mu.Unlock()
	})

	// Wait a moment for the callbacks to run.
	time.Sleep(50 * time.Millisecond)
	mu.Lock()
	if count != 2 {
		t.Fatalf("expected 2 immediate NotifyOnReconnect calls, got %d", count)
	}
	mu.Unlock()
}

func TestDial_NotifyReconnect_NotImmediate(t *testing.T) {
	url := skipWithoutRabbit(t)

	conn, err := rabbit.Dial(url, "test-notify-reconnect")
	if err != nil {
		t.Fatalf("Dial failed: %v", err)
	}
	defer conn.Close()

	called := make(chan struct{}, 1)
	conn.NotifyReconnect(func() {
		called <- struct{}{}
	})

	select {
	case <-called:
		t.Fatal("NotifyReconnect fired immediately")
	case <-time.After(100 * time.Millisecond):
		// expected
	}
}

func TestDial_OnReconnectChannel(t *testing.T) {
	url := skipWithoutRabbit(t)

	conn, err := rabbit.Dial(url, "test-onreconn")
	if err != nil {
		t.Fatalf("Dial failed: %v", err)
	}
	defer conn.Close()

	// OnReconnect returns an open channel immediately after Dial.
	// It should NOT fire (only triggers on actual reconnects).
	select {
	case <-conn.OnReconnect():
		t.Log("OnReconnect fired immediately (race with initial connect — allowed)")
	default:
	}

	// Verify OnReconnect returns a non-nil channel.
	ch := conn.OnReconnect()
	if ch == nil {
		t.Fatal("OnReconnect returned nil channel")
	}
}

func TestDial_ConcurrentUse(t *testing.T) {
	url := skipWithoutRabbit(t)

	conn, err := rabbit.Dial(url, "test-concurrent")
	if err != nil {
		t.Fatalf("Dial failed: %v", err)
	}
	defer conn.Close()

	// Channel() and Connection() should be safe from multiple goroutines.
	var wg sync.WaitGroup
	for i := 0; i < 10; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			ch, err := conn.Channel()
			if err != nil {
				t.Logf("concurrent Channel(): %v", err)
				return
			}
			_ = ch.Close()
		}()
	}
	wg.Wait()
}

// TestOnReconnect_RegisteredFnsFiresOnReconnect is a structural test: it
// verifies that the registered notification list is called on reconnect by
// checking that functions registered after initial connect are not called
// immediately (they only fire on the NEXT reconnect). Full reconnect testing
// requires controlling the broker — see the integration test harness.
func TestOnReconnect_NotImmediate(t *testing.T) {
	url := skipWithoutRabbit(t)

	conn, err := rabbit.Dial(url, "test-reconn-struct")
	if err != nil {
		t.Fatalf("Dial failed: %v", err)
	}
	defer conn.Close()

	// Register a function with NotifyOnReconnect AFTER Dial. It fires
	// immediately (that's the contract).
	immediate := make(chan struct{}, 1)
	conn.NotifyOnReconnect(func() {
		immediate <- struct{}{}
	})

	select {
	case <-immediate:
		// expected
	case <-time.After(time.Second):
		t.Fatal("NotifyOnReconnect did not fire immediately")
	}

	// OnReconnect should NOT have fired (only fires on actual reconnect).
	select {
	case <-conn.OnReconnect():
		t.Log("OnReconnect fired — race goroutine from initial connect")
	default:
	}
}

func TestConnect_BackwardCompat(t *testing.T) {
	url := skipWithoutRabbit(t)

	// Connect should still work and return a usable *amqp.Connection.
	conn, err := rabbit.Connect(url, "test-compat")
	if err != nil {
		t.Fatalf("Connect failed: %v", err)
	}
	defer conn.Close()

	ch, err := conn.Channel()
	if err != nil {
		t.Fatalf("Channel from Connect failed: %v", err)
	}
	_ = ch.Close()
}