feat(bigtable): introduce prime() method and load counting based on unary,streaming and ability to remove connection

bigtable/bigtable.go

@@ -183,9 +183,23 @@ func NewClientWithConfig(ctx context.Context, project, instance string, config C
 	var connPoolErr error
 	enableBigtableConnPool := btopt.EnableBigtableConnectionPool()
 	if enableBigtableConnPool {
-		connPool, connPoolErr = btransport.NewBigtableChannelPool(defaultBigtableConnPoolSize, btopt.BigtableLoadBalancingStrategy(), func() (*grpc.ClientConn, error) {
-			return gtransport.Dial(ctx, o...)
-		})
+		fullInstanceName := fmt.Sprintf("projects/%s/instances/%s", project, instance)
+		connPool, connPoolErr = btransport.NewBigtableChannelPool(ctx,
+			defaultBigtableConnPoolSize,
+			btopt.LeastInFlight,
+			func() (*btransport.BigtableConn, error) {
+				grpcConn, err := gtransport.Dial(ctx, o...)
+				if err != nil {
+					return nil, err
+				}
+				return btransport.NewBigtableConn(grpcConn), nil
+			},
+			// options
+			btransport.WithInstanceName(fullInstanceName),
+			btransport.WithAppProfile(config.AppProfile),
+			btransport.WithFeatureFlagsMetadata(ffMD),
+		)
+
 	} else {
 		// use to regular ConnPool
 		connPool, connPoolErr = gtransport.DialPool(ctx, o...)

bigtable/internal/option/option.go

@@ -20,6 +20,7 @@ package option
 import (
 	"context"
 	"fmt"
+	"log"
 	"os"
 	"strconv"
 	"strings"
@@ -201,3 +202,27 @@ func EnableBigtableConnectionPool() bool {
 	}
 	return enableBigtableConnPool
 }
+
+// Logf logs the given message to the given logger, or the standard logger if
+// the given logger is nil.
+func logf(logger *log.Logger, format string, v ...interface{}) {
+	if logger == nil {
+		log.Printf(format, v...)
+	} else {
+		logger.Printf(format, v...)
+	}
+}
+
+var debug = os.Getenv("CBT_ENABLE_DEBUG") == "true"
+
+// Debugf logs the given message *only if* the global Debug flag is true.
+// It reuses Logf to handle the nil logger logic and prepends "DEBUG: "
+// to the message.
+func Debugf(logger *log.Logger, format string, v ...interface{}) {
+	// Only log if the Debug flag is set
+	if debug {
+		// Prepend "DEBUG: " to the format string
+		debugFormat := "DEBUG: " + format
+		logf(logger, debugFormat, v...)
+	}
+}