pool: add external load balancing methods support

I added the ability to use custom balancing methods when connecting with a pool.

Author: Maksim Konovalov

CHANGELOG.md

@@ -14,6 +14,7 @@ Versioning](http://semver.org/spec/v2.0.0.html) except to the first release.
   also added logs for error case of `ConnectionPool.tryConnect()` calls in 
   `ConnectionPool.controller()` and `ConnectionPool.reconnect()`
 - Methods that are implemented but not included in the pooler interface (#395).
+- Add support for external load balancing methods when connecting via Pool (#400).
 
 ### Changed
 

pool/balancing.go

@@ -0,0 +1,14 @@
+package pool
+
+import "github.com/tarantool/go-tarantool/v2"
+
+type BalancingMethod = func(int) BalancingPool
+
+type BalancingPool interface {
+	IsEmpty() bool
+	GetConnection(string) *tarantool.Connection
+	DeleteConnection(string) *tarantool.Connection
+	AddConnection(id string, conn *tarantool.Connection)
+	GetNextConnection() *tarantool.Connection
+	GetConnections() map[string]*tarantool.Connection
+}

pool/connection_pool.go

@@ -81,6 +81,8 @@ type Opts struct {
 	CheckTimeout time.Duration
 	// ConnectionHandler provides an ability to handle connection updates.
 	ConnectionHandler ConnectionHandler
+	// BalancingMethod is how connections for the request will be selected
+	BalancingMethod BalancingMethod
 }
 
 /*
@@ -110,9 +112,9 @@ type ConnectionPool struct {
 
 	state            state
 	done             chan struct{}
-	roPool           *roundRobinStrategy
-	rwPool           *roundRobinStrategy
-	anyPool          *roundRobinStrategy
+	roPool           BalancingPool
+	rwPool           BalancingPool
+	anyPool          BalancingPool
 	poolsMutex       sync.RWMutex
 	watcherContainer watcherContainer
 }
@@ -153,6 +155,10 @@ func newEndpoint(name string, dialer tarantool.Dialer, opts tarantool.Opts) *end
 // opts. Instances must have unique names.
 func ConnectWithOpts(ctx context.Context, instances []Instance,
 	opts Opts) (*ConnectionPool, error) {
+	if opts.BalancingMethod == nil {
+		opts.BalancingMethod = NewRoundRobinStrategy
+	}
+
 	unique := make(map[string]bool)
 	for _, instance := range instances {
 		if _, ok := unique[instance.Name]; ok {
@@ -166,9 +172,9 @@ func ConnectWithOpts(ctx context.Context, instances []Instance,
 	}
 
 	size := len(instances)
-	rwPool := newRoundRobinStrategy(size)
-	roPool := newRoundRobinStrategy(size)
-	anyPool := newRoundRobinStrategy(size)
+	rwPool := opts.BalancingMethod(size)
+	roPool := opts.BalancingMethod(size)
+	anyPool := opts.BalancingMethod(size)
 
 	connPool := &ConnectionPool{
 		ends:    make(map[string]*endpoint),

pool/round_robin.go

@@ -7,24 +7,26 @@ import (
 	"github.com/tarantool/go-tarantool/v2"
 )
 
-type roundRobinStrategy struct {
+var _ BalancingPool = (*RoundRobinStrategy)(nil)
+
+type RoundRobinStrategy struct {
 	conns     []*tarantool.Connection
 	indexById map[string]uint
 	mutex     sync.RWMutex
 	size      uint64
 	current   uint64
 }
 
-func newRoundRobinStrategy(size int) *roundRobinStrategy {
-	return &roundRobinStrategy{
+func NewRoundRobinStrategy(size int) BalancingPool {
+	return &RoundRobinStrategy{
 		conns:     make([]*tarantool.Connection, 0, size),
 		indexById: make(map[string]uint, size),
 		size:      0,
 		current:   0,
 	}
 }
 
-func (r *roundRobinStrategy) GetConnection(id string) *tarantool.Connection {
+func (r *RoundRobinStrategy) GetConnection(id string) *tarantool.Connection {
 	r.mutex.RLock()
 	defer r.mutex.RUnlock()
 
@@ -36,7 +38,7 @@ func (r *roundRobinStrategy) GetConnection(id string) *tarantool.Connection {
 	return r.conns[index]
 }
 
-func (r *roundRobinStrategy) DeleteConnection(id string) *tarantool.Connection {
+func (r *RoundRobinStrategy) DeleteConnection(id string) *tarantool.Connection {
 	r.mutex.Lock()
 	defer r.mutex.Unlock()
 
@@ -64,14 +66,14 @@ func (r *roundRobinStrategy) DeleteConnection(id string) *tarantool.Connection {
 	return conn
 }
 
-func (r *roundRobinStrategy) IsEmpty() bool {
+func (r *RoundRobinStrategy) IsEmpty() bool {
 	r.mutex.RLock()
 	defer r.mutex.RUnlock()
 
 	return r.size == 0
 }
 
-func (r *roundRobinStrategy) GetNextConnection() *tarantool.Connection {
+func (r *RoundRobinStrategy) GetNextConnection() *tarantool.Connection {
 	r.mutex.RLock()
 	defer r.mutex.RUnlock()
 
@@ -81,7 +83,7 @@ func (r *roundRobinStrategy) GetNextConnection() *tarantool.Connection {
 	return r.conns[r.nextIndex()]
 }
 
-func (r *roundRobinStrategy) GetConnections() map[string]*tarantool.Connection {
+func (r *RoundRobinStrategy) GetConnections() map[string]*tarantool.Connection {
 	r.mutex.RLock()
 	defer r.mutex.RUnlock()
 
@@ -93,7 +95,7 @@ func (r *roundRobinStrategy) GetConnections() map[string]*tarantool.Connection {
 	return conns
 }
 
-func (r *roundRobinStrategy) AddConnection(id string, conn *tarantool.Connection) {
+func (r *RoundRobinStrategy) AddConnection(id string, conn *tarantool.Connection) {
 	r.mutex.Lock()
 	defer r.mutex.Unlock()
 
@@ -106,7 +108,7 @@ func (r *roundRobinStrategy) AddConnection(id string, conn *tarantool.Connection
 	}
 }
 
-func (r *roundRobinStrategy) nextIndex() uint64 {
+func (r *RoundRobinStrategy) nextIndex() uint64 {
 	next := atomic.AddUint64(&r.current, 1)
 	return (next - 1) % r.size
 }

pool/round_robin_test.go

@@ -12,7 +12,7 @@ const (
 )
 
 func TestRoundRobinAddDelete(t *testing.T) {
-	rr := newRoundRobinStrategy(10)
+	rr := NewRoundRobinStrategy(10)
 
 	addrs := []string{validAddr1, validAddr2}
 	conns := []*tarantool.Connection{&tarantool.Connection{}, &tarantool.Connection{}}
@@ -32,7 +32,7 @@ func TestRoundRobinAddDelete(t *testing.T) {
 }
 
 func TestRoundRobinAddDuplicateDelete(t *testing.T) {
-	rr := newRoundRobinStrategy(10)
+	rr := NewRoundRobinStrategy(10)
 
 	conn1 := &tarantool.Connection{}
 	conn2 := &tarantool.Connection{}
@@ -52,7 +52,7 @@ func TestRoundRobinAddDuplicateDelete(t *testing.T) {
 }
 
 func TestRoundRobinGetNextConnection(t *testing.T) {
-	rr := newRoundRobinStrategy(10)
+	rr := NewRoundRobinStrategy(10)
 
 	addrs := []string{validAddr1, validAddr2}
 	conns := []*tarantool.Connection{&tarantool.Connection{}, &tarantool.Connection{}}
@@ -70,7 +70,7 @@ func TestRoundRobinGetNextConnection(t *testing.T) {
 }
 
 func TestRoundRobinStrategy_GetConnections(t *testing.T) {
-	rr := newRoundRobinStrategy(10)
+	rr := NewRoundRobinStrategy(10)
 
 	addrs := []string{validAddr1, validAddr2}
 	conns := []*tarantool.Connection{&tarantool.Connection{}, &tarantool.Connection{}}