lislgosms/gateway/internal/upstream/pool_test.go

package upstream

import (
	"context"
	cmpp "github.com/bigwhite/gocmpp"
	"testing"
	"time"

	"cmpp-platform/gateway/internal/queue"
)

func TestConnectionPoolAcquiresAcrossConnections(t *testing.T) {
	pool := &connectionPool{
		conns: []*connection{
			{client: &cmpp.Client{}, window: make(chan struct{}, 64), windowLimit: 1},
			{client: &cmpp.Client{}, window: make(chan struct{}, 64), windowLimit: 1},
		},
	}

	first, releaseFirst := pool.tryAcquireConnection()
	if first == nil {
		t.Fatalf("expected first connection")
	}
	second, releaseSecond := pool.tryAcquireConnection()
	if second == nil {
		t.Fatalf("expected second connection")
	}
	if first == second {
		t.Fatalf("expected pool to use another connection when the first window is full")
	}
	third, _ := pool.tryAcquireConnection()
	if third != nil {
		t.Fatalf("expected nil connection while all windows are full")
	}

	releaseFirst()
	reacquired, releaseReacquired := pool.tryAcquireConnection()
	if reacquired == nil {
		t.Fatalf("expected a connection after releasing a window")
	}
	releaseReacquired()
	releaseSecond()
}

func TestPoolContinuesOnHealthyConnectionWhenPeerIsDraining(t *testing.T) {
	draining := &connection{client: &cmpp.Client{}, window: make(chan struct{}, 64), windowLimit: 16, draining: true}
	healthy := &connection{client: &cmpp.Client{}, window: make(chan struct{}, 64), windowLimit: 16}
	pool := &connectionPool{conns: []*connection{draining, healthy}}
	selected, release := pool.tryAcquireConnection()
	if selected != healthy {
		t.Fatal("expected healthy peer connection")
	}
	release()
}

func TestRuntimeCapacityMatrixAndSmoothScaleDown(t *testing.T) {
	for _, connections := range []int{1, 2, 4, 8} {
		for _, window := range []int{1, 16, 32, 64} {
			config := normalizeUpstreamConfig(queue.UpstreamConfig{GatewayHost: "127.0.0.1", GatewayPort: 17890, Account: "a", PasswordCipher: "p", CMPPVersion: "3.0", DesiredConnections: connections, WindowSize: window})
			if config.DesiredConnections != connections || config.WindowSize != window {
				t.Fatalf("matrix normalized incorrectly: %+v", config)
			}
		}
	}
	pool := &connectionPool{channelID: "channel-1", connectionID: "primary", config: normalizeUpstreamConfig(queue.UpstreamConfig{DesiredConnections: 2, WindowSize: 16, ConnectionDrainSeconds: 1}), stopCh: make(chan struct{}), reconnectSignal: make(chan struct{}, 1)}
	pool.conns = []*connection{{pool: pool, window: make(chan struct{}, 64), windowLimit: 16}, {pool: pool, window: make(chan struct{}, 64), windowLimit: 16}}
	pool.reconfigure(queue.UpstreamConfig{DesiredConnections: 1, WindowSize: 32, ConnectionDrainSeconds: 1})
	deadline := time.Now().Add(time.Second)
	for {
		pool.mu.Lock()
		count := len(pool.conns)
		first := pool.conns[0]
		pool.mu.Unlock()
		if count == 1 {
			first.mu.Lock()
			limit := first.windowLimit
			first.mu.Unlock()
			if limit != 32 {
				t.Fatalf("window limit=%d want32", limit)
			}
			break
		}
		if time.Now().After(deadline) {
			t.Fatal("scale down did not complete")
		}
		time.Sleep(time.Millisecond)
	}
}

func TestManagerDisconnectChannelRemovesPoolAndStopsReconnects(t *testing.T) {
	pool := &connectionPool{
		channelID:       "channel-1",
		connectionID:    "channel-1:primary",
		config:          normalizeUpstreamConfig(queueUpstreamConfigForTest()),
		reconnectSignal: make(chan struct{}, 1),
		stopCh:          make(chan struct{}),
	}
	manager := &Manager{conns: map[string]*connectionPool{"channel-1": pool}}

	state, err := manager.DisconnectChannel(context.Background(), queue.DisconnectChannelCommand{
		MessageType:  queue.MessageTypeDisconnectChannel,
		ChannelID:    "channel-1",
		ConnectionID: "channel-1:primary",
	})

	if err != nil {
		t.Fatalf("disconnect channel: %v", err)
	}
	if state.Status != "disconnected" || state.CurrentConnections != 0 {
		t.Fatalf("unexpected state: %+v", state)
	}
	if _, exists := manager.conns["channel-1"]; exists {
		t.Fatal("expected channel pool to be removed")
	}
	select {
	case <-pool.stopCh:
	default:
		t.Fatal("expected reconnect supervisor to be stopped")
	}
}

func TestNormalizeUpstreamConfigDefaults(t *testing.T) {
	config := normalizeUpstreamConfig(queueUpstreamConfigForTest())
	if config.DesiredConnections != 1 {
		t.Fatalf("DesiredConnections = %d, want 1", config.DesiredConnections)
	}
	if config.WindowSize != defaultWindowSize {
		t.Fatalf("WindowSize = %d, want %d", config.WindowSize, defaultWindowSize)
	}
	if config.HeartbeatIntervalSeconds != 30 || config.HeartbeatMissThreshold != 3 {
		t.Fatalf("unexpected heartbeat defaults: interval=%d threshold=%d", config.HeartbeatIntervalSeconds, config.HeartbeatMissThreshold)
	}
}

func TestManagerSeparatesConnectionStateAndSupplierEventAPIs(t *testing.T) {
	manager := &Manager{APIBaseURL: "http://main-api/api", EventAPIBaseURL: "http://callback-api/api"}
	pool := manager.newConnectionPool("channel-1", "channel-1:primary", queue.UpstreamConfig{DesiredConnections: 1})
	if pool.apiBaseURL != manager.EventAPIBaseURL {
		t.Fatalf("supplier events must use callback API, got %q", pool.apiBaseURL)
	}
	if manager.APIBaseURL == pool.apiBaseURL {
		t.Fatal("connection-state control API must remain separate from supplier events")
	}
}

func queueUpstreamConfigForTest() queue.UpstreamConfig {
	return queue.UpstreamConfig{
		GatewayHost:    "127.0.0.1",
		GatewayPort:    17890,
		Account:        "account",
		PasswordCipher: "secret",
		CMPPVersion:    "3.0",
	}
}