Excluded Close and Park methods from conn.Conn interface

Author: asmyasnikov

CHANGELOG.md

@@ -1,6 +1,7 @@
 ## 3.13.1
 * Improved error messages
 * Defended `cluster.balancer` with `sync.RWMutex` on `cluster.Insert`, `cluster.Update`, `cluster.Remove` and `cluster.Get`
+* Excluded `Close` and `Park` methods from `conn.Conn` interface
 
 ## 3.13.0
 * Refactored `Connection` interface

internal/cluster/cluster.go

@@ -382,11 +382,6 @@ func (c *cluster) Remove(ctx context.Context, e endpoint.Endpoint, opts ...crudO
 	delete(c.index, e.Address())
 	delete(c.endpoints, e.NodeID())
 
-	if entry.Conn != nil {
-		// entry.Conn may be nil when connection is being tracked after unsuccessful dial().
-		_ = entry.Conn.Close(ctx)
-	}
-
 	return entry.Conn
 }
 

internal/cluster/cluster_test.go

@@ -46,10 +46,10 @@ func TestClusterFastRedial(t *testing.T) {
 		done := make(chan struct{})
 		go func() {
 			for i := 0; i < size*10; i++ {
-				c, err := c.Get(context.Background())
+				cc, err := c.Get(context.Background())
 				// enforce close bad connects to track them
-				if err == nil && c != nil && c.Endpoint().Address() == "bad:0" {
-					_ = c.Close(ctx)
+				if err == nil && c != nil && cc.Endpoint().Address() == "bad:0" {
+					_ = c.Remove(ctx, cc.Endpoint())
 				}
 			}
 			close(done)

internal/conn/conn.go

@@ -37,9 +37,6 @@ type Conn interface {
 	IsState(states ...State) bool
 	GetState() State
 	SetState(State) State
-
-	Close(ctx context.Context) error
-	Park(ctx context.Context) error
 }
 
 func (c *conn) Address() string {
@@ -56,7 +53,7 @@ type conn struct {
 	state    State
 	usages   int32
 	ttl      timeutil.Timer
-	onClose  []func(Conn)
+	onClose  []func(*conn)
 }
 
 func (c *conn) IsState(states ...State) bool {
@@ -426,21 +423,21 @@ func (c *conn) NewStream(
 
 type option func(c *conn)
 
-func withOnClose(onClose func(Conn)) option {
+func withOnClose(onClose func(*conn)) option {
 	return func(c *conn) {
 		if onClose != nil {
 			c.onClose = append(c.onClose, onClose)
 		}
 	}
 }
 
-func New(e endpoint.Endpoint, config Config, opts ...option) Conn {
+func newConn(e endpoint.Endpoint, config Config, opts ...option) *conn {
 	c := &conn{
 		state:    Created,
 		endpoint: e,
 		config:   config,
 		done:     make(chan struct{}),
-		onClose:  make([]func(Conn), 0),
+		onClose:  make([]func(*conn), 0),
 	}
 	for _, o := range opts {
 		o(c)
@@ -450,3 +447,7 @@ func New(e endpoint.Endpoint, config Config, opts ...option) Conn {
 	}
 	return c
 }
+
+func New(e endpoint.Endpoint, config Config, opts ...option) Conn {
+	return newConn(e, config, opts...)
+}

internal/conn/pool.go

@@ -46,7 +46,7 @@ type pool struct {
 	config Config
 	mtx    sync.RWMutex
 	opts   []grpc.DialOption
-	conns  map[string]Conn
+	conns  map[string]*conn
 	done   chan struct{}
 }
 
@@ -101,10 +101,10 @@ func (p *pool) GetConn(e endpoint.Endpoint) Conn {
 	if cc, ok := p.conns[e.Address()]; ok {
 		return cc
 	}
-	cc := New(
+	cc := newConn(
 		e,
 		p.config,
-		withOnClose(func(c Conn) {
+		withOnClose(func(c *conn) {
 			// conn.Conn.Close() must called on under locked p.mtx
 			delete(p.conns, c.Endpoint().Address())
 		}),
@@ -141,7 +141,7 @@ func NewPool(
 		usages: 1,
 		config: config,
 		opts:   config.GrpcDialOptions(),
-		conns:  make(map[string]Conn),
+		conns:  make(map[string]*conn),
 		done:   make(chan struct{}),
 	}
 	if ttl := config.ConnectionTTL(); ttl > 0 {