Added MaxActiveConns (#2646)

* Added the ability to set a connection growth limit when there are not enough connections in the pool using MaxActiveConns

* fix comment

* fix

* fix

---------

Co-authored-by: ofekshenawa <104765379+ofekshenawa@users.noreply.github.com>

internal/pool/pool.go

@@ -15,6 +15,10 @@ var (
 	// ErrClosed performs any operation on the closed client will return this error.
 	ErrClosed = errors.New("redis: client is closed")
 
+	// ErrPoolExhausted is returned from a pool connection method
+	// when the maximum number of database connections in the pool has been reached.
+	ErrPoolExhausted = errors.New("redis: connection pool exhausted")
+
 	// ErrPoolTimeout timed out waiting to get a connection from the connection pool.
 	ErrPoolTimeout = errors.New("redis: connection pool timeout")
 )
@@ -61,6 +65,7 @@ type Options struct {
 	PoolTimeout     time.Duration
 	MinIdleConns    int
 	MaxIdleConns    int
+	MaxActiveConns  int
 	ConnMaxIdleTime time.Duration
 	ConnMaxLifetime time.Duration
 }
@@ -159,6 +164,14 @@ func (p *ConnPool) NewConn(ctx context.Context) (*Conn, error) {
 }
 
 func (p *ConnPool) newConn(ctx context.Context, pooled bool) (*Conn, error) {
+	if p.closed() {
+		return nil, ErrClosed
+	}
+
+	if p.cfg.MaxActiveConns > 0 && p.poolSize >= p.cfg.MaxActiveConns {
+		return nil, ErrPoolExhausted
+	}
+
 	cn, err := p.dialConn(ctx, pooled)
 	if err != nil {
 		return nil, err
@@ -167,12 +180,6 @@ func (p *ConnPool) newConn(ctx context.Context, pooled bool) (*Conn, error) {
 	p.connsMu.Lock()
 	defer p.connsMu.Unlock()
 
-	// It is not allowed to add new connections to the closed connection pool.
-	if p.closed() {
-		_ = cn.Close()
-		return nil, ErrClosed
-	}
-
 	p.conns = append(p.conns, cn)
 	if pooled {
 		// If pool is full remove the cn on next Put.