GO: Lazy connect implementation

Signed-off-by: affonsov <67347924+affonsov@users.noreply.github.com>

Author: affonsov

go/config/config.go

@@ -101,6 +101,7 @@ type baseClientConfiguration struct {
 	clientName        string
 	clientAZ          string
 	reconnectStrategy *BackoffStrategy
+	lazyConnect       bool
 }
 
 func (config *baseClientConfiguration) toProtobuf() (*protobuf.ConnectionRequest, error) {
@@ -147,6 +148,10 @@ func (config *baseClientConfiguration) toProtobuf() (*protobuf.ConnectionRequest
 		request.ConnectionRetryStrategy = config.reconnectStrategy.toProtobuf()
 	}
 
+	if config.lazyConnect {
+		request.LazyConnect = config.lazyConnect
+	}
+
 	return &request, nil
 }
 
@@ -269,6 +274,14 @@ func (config *ClientConfiguration) WithUseTLS(useTLS bool) *ClientConfiguration
 	return config
 }
 
+// WithLazyConnect configures whether the client should establish connections lazily. When set to true,
+// the client will only establish connections when needed for the first operation, rather than
+// immediately upon client creation.
+func (config *ClientConfiguration) WithLazyConnect(lazyConnect bool) *ClientConfiguration {
+	config.lazyConnect = lazyConnect
+	return config
+}
+
 // WithCredentials sets the credentials for the authentication process. If none are set, the client will not authenticate
 // itself with the server.
 func (config *ClientConfiguration) WithCredentials(credentials *ServerCredentials) *ClientConfiguration {
@@ -410,6 +423,14 @@ func (config *ClusterClientConfiguration) WithUseTLS(useTLS bool) *ClusterClient
 	return config
 }
 
+// WithLazyConnect configures whether the client should establish connections lazily. When set to true,
+// the client will only establish connections when needed for the first operation, rather than
+// immediately upon client creation.
+func (config *ClusterClientConfiguration) WithLazyConnect(lazyConnect bool) *ClusterClientConfiguration {
+	config.lazyConnect = lazyConnect
+	return config
+}
+
 // WithCredentials sets the credentials for the authentication process. If none are set, the client will not authenticate
 // itself with the server.
 func (config *ClusterClientConfiguration) WithCredentials(

go/config/config_test.go

@@ -296,3 +296,47 @@ func TestConfig_InvalidRequestAndConnectionTimeouts(t *testing.T) {
 	_, err8 := config8.ToProtobuf()
 	assert.EqualError(t, err8, "setting connection timeout returned an error: invalid duration was specified")
 }
+
+func TestConfig_LazyConnect(t *testing.T) {
+	// Test for ClientConfiguration
+	clientConfig := NewClientConfiguration().
+		WithLazyConnect(true)
+
+	clientResult, err := clientConfig.ToProtobuf()
+	if err != nil {
+		t.Fatalf("Failed to convert client config to protobuf: %v", err)
+	}
+
+	assert.True(t, clientResult.LazyConnect)
+
+	// Test for ClusterClientConfiguration
+	clusterConfig := NewClusterClientConfiguration().
+		WithLazyConnect(true)
+
+	clusterResult, err := clusterConfig.ToProtobuf()
+	if err != nil {
+		t.Fatalf("Failed to convert cluster config to protobuf: %v", err)
+	}
+
+	assert.True(t, clusterResult.LazyConnect)
+
+	// Test default value (false) for ClientConfiguration
+	defaultClientConfig := NewClientConfiguration()
+
+	defaultClientResult, err := defaultClientConfig.ToProtobuf()
+	if err != nil {
+		t.Fatalf("Failed to convert default client config to protobuf: %v", err)
+	}
+
+	assert.False(t, defaultClientResult.LazyConnect)
+
+	// Test default value (false) for ClusterClientConfiguration
+	defaultClusterConfig := NewClusterClientConfiguration()
+
+	defaultClusterResult, err := defaultClusterConfig.ToProtobuf()
+	if err != nil {
+		t.Fatalf("Failed to convert default cluster config to protobuf: %v", err)
+	}
+
+	assert.False(t, defaultClusterResult.LazyConnect)
+}

go/integTest/connection_test.go

@@ -4,6 +4,7 @@ package integTest
 
 import (
 	"context"
+	"fmt"
 	"strings"
 	"sync"
 	"time"
@@ -14,6 +15,147 @@ import (
 	"github.com/valkey-io/valkey-glide/go/v2/internal/interfaces"
 )
 
+func newDedicatedValkey(suite *GlideTestSuite, clusterMode bool) (string, error) {
+	// Build command arguments
+	args := []string{}
+	args = append(args, "start")
+	if clusterMode {
+		args = append(args, "--cluster-mode")
+	}
+
+	// shardCount := 1
+	// if clusterMode {
+	// 	shardCount = 3
+	// }
+	// args = append(args, fmt.Sprintf("-n %d", shardCount))
+	args = append(args, fmt.Sprintf("-r %d", 1))
+
+	// Execute cluster manager script
+	output := runClusterManager(suite, args, false)
+
+	return output, nil
+}
+
+func stopDedicatedValkey(suite *GlideTestSuite, clusterFolder string) {
+	args := []string{}
+	args = append(args, "stop", "--cluster-folder", clusterFolder)
+
+	runClusterManager(suite, args, false)
+}
+
+func createDedicatedClient(
+	addresses []config.NodeAddress,
+	clusterMode bool,
+	lazyConnect bool,
+) (interfaces.BaseClientCommands, error) {
+	if clusterMode {
+		cfg := config.NewClusterClientConfiguration()
+		for _, addr := range addresses {
+			cfg.WithAddress(&addr)
+		}
+
+		cfg.WithRequestTimeout(3 * time.Second)
+		advCfg := config.NewAdvancedClusterClientConfiguration()
+		advCfg.WithConnectionTimeout(3 * time.Second)
+		cfg.WithAdvancedConfiguration(advCfg)
+		cfg.WithLazyConnect(lazyConnect)
+
+		return glide.NewClusterClient(cfg)
+	}
+
+	cfg := config.NewClientConfiguration()
+	for _, addr := range addresses {
+		cfg.WithAddress(&addr)
+	}
+
+	cfg.WithRequestTimeout(3 * time.Second)
+	advCfg := config.NewAdvancedClientConfiguration()
+	advCfg.WithConnectionTimeout(3 * time.Second)
+	cfg.WithAdvancedConfiguration(advCfg)
+	cfg.WithLazyConnect(lazyConnect)
+
+	return glide.NewClient(cfg)
+}
+
+// getClientListOutputCount parses CLIENT LIST output and returns the number of clients
+func getClientListOutputCount(output interface{}) int {
+	if output == nil {
+		return 0
+	}
+
+	var text string
+	switch v := output.(type) {
+	case []byte:
+		text = string(v)
+	case string:
+		text = v
+	default:
+		return 0
+	}
+
+	if text = strings.TrimSpace(text); text == "" {
+		return 0
+	}
+
+	return len(strings.Split(text, "\n"))
+}
+
+// getClientCount returns the number of connected clients
+func getClientCount(ctx context.Context, client interfaces.BaseClientCommands) (int, error) {
+	if clusterClient, ok := client.(interfaces.GlideClusterClientCommands); ok {
+		// For cluster client, execute CLIENT LIST on all nodes
+		result, err := clusterClient.CustomCommandWithRoute(ctx, []string{"CLIENT", "LIST"}, config.AllNodes)
+		if err != nil {
+			return 0, err
+		}
+
+		// Result will be a map with node addresses as keys and CLIENT LIST output as values
+		totalCount := 0
+		for _, nodeOutput := range result.MultiValue() {
+			totalCount += getClientListOutputCount(nodeOutput)
+		}
+		return totalCount, nil
+	}
+
+	// For standalone client, execute CLIENT LIST directly
+	glideClient := client.(interfaces.GlideClientCommands)
+	result, err := glideClient.CustomCommand(ctx, []string{"CLIENT", "LIST"})
+	if err != nil {
+		return 0, err
+	}
+	return getClientListOutputCount(result), nil
+}
+
+// getExpectedNewConnections returns the expected number of new connections when a lazy client is initialized
+func getExpectedNewConnections(ctx context.Context, client interfaces.BaseClientCommands) (int, error) {
+	if clusterClient, ok := client.(interfaces.GlideClusterClientCommands); ok {
+		// For cluster, get node count and multiply by 2 (2 connections per node)
+		result, err := clusterClient.CustomCommand(ctx, []string{"CLUSTER", "NODES"})
+		if err != nil {
+			return 0, err
+		}
+
+		var nodesInfo string
+		switch v := result.SingleValue().(type) {
+		case []byte:
+			nodesInfo = string(v)
+		case string:
+			nodesInfo = v
+		default:
+			nodesInfo = ""
+		}
+
+		if nodesInfo = strings.TrimSpace(nodesInfo); nodesInfo == "" {
+			return 0, nil
+		}
+
+		return len(strings.Split(nodesInfo, "\n")) * 2, nil
+	}
+
+	// For standalone, always expect 1 new connection
+	return 1, nil
+}
+
 func (suite *GlideTestSuite) TestStandaloneConnect() {
 	config := config.NewClientConfiguration().
 		WithAddress(&suite.standaloneHosts[0])
@@ -156,3 +298,60 @@ func (suite *GlideTestSuite) TestConnectionTimeout() {
 		}
 	})
 }
+
+func (suite *GlideTestSuite) TestLazyConnectionEstablishesOnFirstCommand() {
+	// Run test for both standalone and cluster modes
+	suite.runWithTimeoutClients(func(client interfaces.BaseClientCommands) {
+		ctx := context.Background()
+		_, isCluster := client.(interfaces.GlideClusterClientCommands)
+
+		// Create a monitoring client (eagerly connected)
+		output, err := newDedicatedValkey(suite, isCluster)
+		suite.NoError(err)
+		clusterFolder := extractClusterFolder(suite, output)
+		addresses := extractAddresses(suite, output)
+		defer stopDedicatedValkey(suite, clusterFolder)
+		monitoringClient, err := createDedicatedClient(addresses, isCluster, false)
+		suite.NoError(err)
+		defer monitoringClient.Close()
+
+		// Get initial client count
+		clientsBeforeLazyInit, err := getClientCount(ctx, monitoringClient)
+		suite.NoError(err)
+
+		// Create the "lazy" client
+		lazyClient, err := createDedicatedClient(addresses, isCluster, true)
+		suite.NoError(err)
+		defer lazyClient.Close()
+
+		// Check count (should not change)
+		clientsAfterLazyInit, err := getClientCount(ctx, monitoringClient)
+		suite.NoError(err)
+		suite.Equal(clientsBeforeLazyInit, clientsAfterLazyInit,
+			"Lazy client should not connect before the first command")
+
+		// Send the first command using the lazy client
+		var result interface{}
+		if isCluster {
+			clusterClient := lazyClient.(interfaces.GlideClusterClientCommands)
+			result, err = clusterClient.Ping(ctx)
+		} else {
+			glideClient := lazyClient.(interfaces.GlideClientCommands)
+			result, err = glideClient.Ping(ctx)
+		}
+		suite.NoError(err)
+
+		// Assert PING success for both modes
+		suite.Equal("PONG", result)
+
+		// Check client count after the first command
+		clientsAfterFirstCommand, err := getClientCount(ctx, monitoringClient)
+		suite.NoError(err)
+
+		expectedNewConnections, err := getExpectedNewConnections(ctx, monitoringClient)
+		suite.NoError(err)
+
+		suite.Equal(clientsBeforeLazyInit+expectedNewConnections, clientsAfterFirstCommand,
+			"Lazy client should establish expected number of new connections after the first command")
+	})
+}

go/integTest/glide_test_suite_test.go

@@ -146,6 +146,28 @@ func parseHosts(suite *GlideTestSuite, addresses string) []config.NodeAddress {
 	return result
 }
 
+func extractClusterFolder(suite *GlideTestSuite, output string) string {
+	lines := strings.Split(output, "\n")
+	foundFolder := false
+	clusterFolder := ""
+
+	for _, line := range lines {
+		if strings.Contains(line, "CLUSTER_FOLDER=") {
+			parts := strings.SplitN(line, "CLUSTER_FOLDER=", 2)
+			if len(parts) != 2 {
+				suite.T().Fatalf("invalid CLUSTER_FOLDER line format: %s", line)
+			}
+			clusterFolder = strings.TrimSpace(parts[1])
+			foundFolder = true
+		}
+	}
+
+	if !foundFolder {
+		suite.T().Fatalf("missing required output fields")
+	}
+	return clusterFolder
+}
+
 func extractAddresses(suite *GlideTestSuite, output string) []config.NodeAddress {
 	for _, line := range strings.Split(output, "\n") {
 		if !strings.HasPrefix(line, "CLUSTER_NODES=") {