Remove unused maxIdle field from connection pools

Co-authored-by: asim <17530+asim@users.noreply.github.com>

broker/nats/nats.go

@@ -6,6 +6,7 @@ import (
 	"errors"
 	"strings"
 	"sync"
+	"time"
 
 	natsp "github.com/nats-io/nats.go"
 	"go-micro.dev/v5/broker"
@@ -22,10 +23,15 @@ type natsBroker struct {
 	connected bool
 
 	addrs []string
-	conn  *natsp.Conn
+	conn  *natsp.Conn      // single connection (used when pool is disabled)
+	pool  *connectionPool  // connection pool (used when pooling is enabled)
 	opts  broker.Options
 	nopts natsp.Options
 
+	// pool configuration
+	poolSize        int
+	poolIdleTimeout time.Duration
+
 	// should we drain the connection
 	drain   bool
 	closeCh chan (error)
@@ -109,6 +115,39 @@ func (n *natsBroker) Connect() error {
 		return nil
 	}
 
+	// Check if we should use connection pooling
+	if n.poolSize > 1 {
+		// Initialize connection pool
+		factory := func() (*natsp.Conn, error) {
+			opts := n.nopts
+			opts.Servers = n.addrs
+			opts.Secure = n.opts.Secure
+			opts.TLSConfig = n.opts.TLSConfig
+
+			// secure might not be set
+			if n.opts.TLSConfig != nil {
+				opts.Secure = true
+			}
+
+			return opts.Connect()
+		}
+
+		pool, err := newConnectionPool(n.poolSize, factory)
+		if err != nil {
+			return err
+		}
+
+		// Set idle timeout if configured
+		if n.poolIdleTimeout > 0 {
+			pool.idleTimeout = n.poolIdleTimeout
+		}
+
+		n.pool = pool
+		n.connected = true
+		return nil
+	}
+
+	// Single connection mode (original behavior)
 	status := natsp.CLOSED
 	if n.conn != nil {
 		status = n.conn.Status()
@@ -143,14 +182,26 @@ func (n *natsBroker) Disconnect() error {
 	n.Lock()
 	defer n.Unlock()
 
-	// drain the connection if specified
-	if n.drain {
-		n.conn.Drain()
-		n.closeCh <- nil
+	// Close connection pool if it exists
+	if n.pool != nil {
+		if err := n.pool.Close(); err != nil {
+			n.opts.Logger.Log(logger.ErrorLevel, "error closing connection pool:", err)
+		}
+		n.pool = nil
 	}
 
-	// close the client connection
-	n.conn.Close()
+	// Close single connection if it exists
+	if n.conn != nil {
+		// drain the connection if specified
+		if n.drain {
+			n.conn.Drain()
+			n.closeCh <- nil
+		}
+
+		// close the client connection
+		n.conn.Close()
+		n.conn = nil
+	}
 
 	// set not connected
 	n.connected = false
@@ -171,24 +222,42 @@ func (n *natsBroker) Publish(topic string, msg *broker.Message, opts ...broker.P
 	n.RLock()
 	defer n.RUnlock()
 
-	if n.conn == nil {
-		return errors.New("not connected")
-	}
-
 	b, err := n.opts.Codec.Marshal(msg)
 	if err != nil {
 		return err
 	}
+
+	// Use connection pool if enabled
+	if n.pool != nil {
+		poolConn, err := n.pool.Get()
+		if err != nil {
+			return err
+		}
+		defer n.pool.Put(poolConn)
+
+		conn := poolConn.Conn()
+		if conn == nil {
+			return errors.New("invalid connection from pool")
+		}
+		return conn.Publish(topic, b)
+	}
+
+	// Use single connection (original behavior)
+	if n.conn == nil {
+		return errors.New("not connected")
+	}
+
 	return n.conn.Publish(topic, b)
 }
 
 func (n *natsBroker) Subscribe(topic string, handler broker.Handler, opts ...broker.SubscribeOption) (broker.Subscriber, error) {
 	n.RLock()
-	if n.conn == nil {
-		n.RUnlock()
+	hasConnection := n.conn != nil || n.pool != nil
+	n.RUnlock()
+
+	if !hasConnection {
 		return nil, errors.New("not connected")
 	}
-	n.RUnlock()
 
 	opt := broker.SubscribeOptions{
 		AutoAck: true,
@@ -226,6 +295,38 @@ func (n *natsBroker) Subscribe(topic string, handler broker.Handler, opts ...bro
 	var sub *natsp.Subscription
 	var err error
 
+	// Use connection pool if enabled
+	if n.pool != nil {
+		poolConn, err := n.pool.Get()
+		if err != nil {
+			return nil, err
+		}
+
+		conn := poolConn.Conn()
+		if conn == nil {
+			n.pool.Put(poolConn)
+			return nil, errors.New("invalid connection from pool")
+		}
+
+		if len(opt.Queue) > 0 {
+			sub, err = conn.QueueSubscribe(topic, opt.Queue, fn)
+		} else {
+			sub, err = conn.Subscribe(topic, fn)
+		}
+
+		if err != nil {
+			n.pool.Put(poolConn)
+			return nil, err
+		}
+
+		// Return connection to pool after subscription is created
+		// The subscription keeps the connection alive
+		n.pool.Put(poolConn)
+
+		return &subscriber{s: sub, opts: opt}, nil
+	}
+
+	// Use single connection (original behavior)
 	n.RLock()
 	if len(opt.Queue) > 0 {
 		sub, err = n.conn.QueueSubscribe(topic, opt.Queue, fn)
@@ -250,12 +351,24 @@ func (n *natsBroker) setOption(opts ...broker.Option) {
 
 	n.Once.Do(func() {
 		n.nopts = natsp.GetDefaultOptions()
+		n.poolSize = 1 // Default to single connection (no pooling)
+		n.poolIdleTimeout = 5 * time.Minute
 	})
 
 	if nopts, ok := n.opts.Context.Value(optionsKey{}).(natsp.Options); ok {
 		n.nopts = nopts
 	}
 
+	// Set pool size if configured
+	if poolSize, ok := n.opts.Context.Value(poolSizeKey{}).(int); ok && poolSize > 0 {
+		n.poolSize = poolSize
+	}
+
+	// Set pool idle timeout if configured
+	if idleTimeout, ok := n.opts.Context.Value(poolIdleTimeoutKey{}).(time.Duration); ok {
+		n.poolIdleTimeout = idleTimeout
+	}
+
 	// broker.Options have higher priority than nats.Options
 	// only if Addrs, Secure or TLSConfig were not set through a broker.Option
 	// we read them from nats.Option

broker/nats/options.go

@@ -1,12 +1,16 @@
 package nats
 
 import (
+	"time"
+
 	natsp "github.com/nats-io/nats.go"
 	"go-micro.dev/v5/broker"
 )
 
 type optionsKey struct{}
 type drainConnectionKey struct{}
+type poolSizeKey struct{}
+type poolIdleTimeoutKey struct{}
 
 // Options accepts nats.Options.
 func Options(opts natsp.Options) broker.Option {
@@ -17,3 +21,17 @@ func Options(opts natsp.Options) broker.Option {
 func DrainConnection() broker.Option {
 	return setBrokerOption(drainConnectionKey{}, struct{}{})
 }
+
+// PoolSize sets the size of the connection pool.
+// If set to a value > 1, the broker will use a connection pool.
+// Default is 1 (no pooling).
+func PoolSize(size int) broker.Option {
+	return setBrokerOption(poolSizeKey{}, size)
+}
+
+// PoolIdleTimeout sets the timeout for idle connections in the pool.
+// Connections idle for longer than this duration will be closed.
+// Default is 5 minutes. Set to 0 to disable idle timeout.
+func PoolIdleTimeout(timeout time.Duration) broker.Option {
+	return setBrokerOption(poolIdleTimeoutKey{}, timeout)
+}

broker/nats/pool.go

@@ -0,0 +1,188 @@
+package nats
+
+import (
+	"errors"
+	"sync"
+	"time"
+
+	natsp "github.com/nats-io/nats.go"
+)
+
+var (
+	// ErrPoolExhausted is returned when no connections are available in the pool
+	ErrPoolExhausted = errors.New("connection pool exhausted")
+	// ErrPoolClosed is returned when trying to use a closed pool
+	ErrPoolClosed = errors.New("connection pool is closed")
+)
+
+// connectionPool manages a pool of NATS connections
+type connectionPool struct {
+	mu          sync.RWMutex
+	connections chan *pooledConnection
+	factory     func() (*natsp.Conn, error)
+	size        int
+	idleTimeout time.Duration
+	closed      bool
+}
+
+// pooledConnection wraps a NATS connection with metadata
+type pooledConnection struct {
+	conn      *natsp.Conn
+	createdAt time.Time
+	lastUsed  time.Time
+	mu        sync.Mutex
+}
+
+// newConnectionPool creates a new connection pool
+func newConnectionPool(size int, factory func() (*natsp.Conn, error)) (*connectionPool, error) {
+	if size <= 0 {
+		size = 1
+	}
+
+	pool := &connectionPool{
+		connections: make(chan *pooledConnection, size),
+		factory:     factory,
+		size:        size,
+		idleTimeout: 5 * time.Minute,
+		closed:      false,
+	}
+
+	return pool, nil
+}
+
+// Get retrieves a connection from the pool or creates a new one
+func (p *connectionPool) Get() (*pooledConnection, error) {
+	p.mu.RLock()
+	if p.closed {
+		p.mu.RUnlock()
+		return nil, ErrPoolClosed
+	}
+	p.mu.RUnlock()
+
+	// Try to get an existing connection from the pool
+	select {
+	case conn := <-p.connections:
+		// Check if connection is still valid and not idle for too long
+		if conn.isValid() && !conn.isExpired(p.idleTimeout) {
+			conn.updateLastUsed()
+			return conn, nil
+		}
+		// Connection is invalid or expired, close it and create a new one
+		conn.close()
+		return p.createConnection()
+	default:
+		// No connection available, create a new one
+		return p.createConnection()
+	}
+}
+
+// Put returns a connection to the pool
+func (p *connectionPool) Put(conn *pooledConnection) error {
+	p.mu.RLock()
+	defer p.mu.RUnlock()
+
+	if p.closed {
+		return conn.close()
+	}
+
+	// Check if connection is still valid
+	if !conn.isValid() {
+		return conn.close()
+	}
+
+	conn.updateLastUsed()
+
+	// Try to return connection to pool
+	select {
+	case p.connections <- conn:
+		return nil
+	default:
+		// Pool is full, close the connection
+		return conn.close()
+	}
+}
+
+// Close closes all connections in the pool
+func (p *connectionPool) Close() error {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+
+	if p.closed {
+		return nil
+	}
+
+	p.closed = true
+	close(p.connections)
+
+	// Close all connections in the pool
+	for conn := range p.connections {
+		conn.close()
+	}
+
+	return nil
+}
+
+// createConnection creates a new pooled connection
+func (p *connectionPool) createConnection() (*pooledConnection, error) {
+	conn, err := p.factory()
+	if err != nil {
+		return nil, err
+	}
+
+	return &pooledConnection{
+		conn:      conn,
+		createdAt: time.Now(),
+		lastUsed:  time.Now(),
+	}, nil
+}
+
+// isValid checks if the underlying NATS connection is valid
+func (pc *pooledConnection) isValid() bool {
+	pc.mu.Lock()
+	defer pc.mu.Unlock()
+
+	if pc.conn == nil {
+		return false
+	}
+
+	status := pc.conn.Status()
+	return status == natsp.CONNECTED || status == natsp.RECONNECTING
+}
+
+// isExpired checks if the connection has been idle for too long
+func (pc *pooledConnection) isExpired(timeout time.Duration) bool {
+	pc.mu.Lock()
+	defer pc.mu.Unlock()
+
+	if timeout <= 0 {
+		return false
+	}
+
+	return time.Since(pc.lastUsed) > timeout
+}
+
+// close closes the underlying NATS connection
+func (pc *pooledConnection) close() error {
+	pc.mu.Lock()
+	defer pc.mu.Unlock()
+
+	if pc.conn != nil {
+		pc.conn.Close()
+		pc.conn = nil
+	}
+	return nil
+}
+
+// Conn returns the underlying NATS connection
+func (pc *pooledConnection) Conn() *natsp.Conn {
+	pc.mu.Lock()
+	defer pc.mu.Unlock()
+	return pc.conn
+}
+
+// updateLastUsed updates the last used timestamp in a thread-safe manner
+func (pc *pooledConnection) updateLastUsed() {
+	pc.mu.Lock()
+	defer pc.mu.Unlock()
+	pc.lastUsed = time.Now()
+}

broker/nats/pool_test.go

@@ -0,0 +1,204 @@
+package nats
+
+import (
+	"sync"
+	"testing"
+	"time"
+
+	natsp "github.com/nats-io/nats.go"
+)
+
+func TestConnectionPool_GetPut(t *testing.T) {
+	// Mock factory that creates connections
+	connCount := 0
+	factory := func() (*natsp.Conn, error) {
+		connCount++
+		// Return a mock connection (we can't create real NATS connections in tests without a server)
+		// This test is more about the pool logic
+		return nil, nil
+	}
+
+	pool, err := newConnectionPool(3, factory)
+	if err != nil {
+		t.Fatalf("Failed to create pool: %v", err)
+	}
+	defer pool.Close()
+
+	// Get a connection (should create one)
+	conn1, err := pool.Get()
+	if err != nil {
+		t.Fatalf("Failed to get connection: %v", err)
+	}
+	if conn1 == nil {
+		t.Fatal("Expected connection, got nil")
+	}
+
+	// Put it back
+	if err := pool.Put(conn1); err != nil {
+		t.Fatalf("Failed to put connection: %v", err)
+	}
+
+	// Get it again (should reuse the same one)
+	conn2, err := pool.Get()
+	if err != nil {
+		t.Fatalf("Failed to get connection: %v", err)
+	}
+
+	// Since we can't compare actual connections easily, just verify we got one
+	if conn2 == nil {
+		t.Fatal("Expected connection, got nil")
+	}
+}
+
+func TestConnectionPool_Concurrent(t *testing.T) {
+	connCount := 0
+	mu := sync.Mutex{}
+	factory := func() (*natsp.Conn, error) {
+		mu.Lock()
+		connCount++
+		mu.Unlock()
+		return nil, nil
+	}
+
+	pool, err := newConnectionPool(5, factory)
+	if err != nil {
+		t.Fatalf("Failed to create pool: %v", err)
+	}
+	defer pool.Close()
+
+	// Simulate concurrent access
+	var wg sync.WaitGroup
+	for i := 0; i < 10; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			conn, err := pool.Get()
+			if err != nil {
+				t.Errorf("Failed to get connection: %v", err)
+				return
+			}
+			// Simulate some work
+			time.Sleep(10 * time.Millisecond)
+			if err := pool.Put(conn); err != nil {
+				t.Errorf("Failed to put connection: %v", err)
+			}
+		}()
+	}
+
+	wg.Wait()
+
+	// We should have created some connections
+	mu.Lock()
+	if connCount == 0 {
+		t.Error("Expected at least one connection to be created")
+	}
+	mu.Unlock()
+}
+
+func TestConnectionPool_Close(t *testing.T) {
+	factory := func() (*natsp.Conn, error) {
+		return nil, nil
+	}
+
+	pool, err := newConnectionPool(3, factory)
+	if err != nil {
+		t.Fatalf("Failed to create pool: %v", err)
+	}
+
+	// Get a connection
+	conn, err := pool.Get()
+	if err != nil {
+		t.Fatalf("Failed to get connection: %v", err)
+	}
+
+	// Close the pool
+	if err := pool.Close(); err != nil {
+		t.Fatalf("Failed to close pool: %v", err)
+	}
+
+	// Put connection back to closed pool should not panic
+	// The connection will be closed instead of returned to pool
+	_ = pool.Put(conn)
+
+	// Try to get from closed pool
+	_, err = pool.Get()
+	if err != ErrPoolClosed {
+		t.Errorf("Expected ErrPoolClosed, got: %v", err)
+	}
+}
+
+func TestPooledConnection_IsValid(t *testing.T) {
+	pc := &pooledConnection{
+		conn:      nil, // nil connection should be invalid
+		createdAt: time.Now(),
+		lastUsed:  time.Now(),
+	}
+
+	if pc.isValid() {
+		t.Error("Expected nil connection to be invalid")
+	}
+}
+
+func TestPooledConnection_IsExpired(t *testing.T) {
+	pc := &pooledConnection{
+		conn:      nil,
+		createdAt: time.Now(),
+		lastUsed:  time.Now().Add(-10 * time.Minute), // 10 minutes ago
+	}
+
+	// With 5 minute timeout, should be expired
+	if !pc.isExpired(5 * time.Minute) {
+		t.Error("Expected connection to be expired")
+	}
+
+	// With 0 timeout, should never expire
+	if pc.isExpired(0) {
+		t.Error("Expected connection not to expire with 0 timeout")
+	}
+
+	// With 20 minute timeout, should not be expired
+	if pc.isExpired(20 * time.Minute) {
+		t.Error("Expected connection not to be expired")
+	}
+}
+
+func TestNatsBroker_PoolConfiguration(t *testing.T) {
+	// Test that pool size is set correctly
+	br := NewNatsBroker(PoolSize(5))
+	nb, ok := br.(*natsBroker)
+	if !ok {
+		t.Fatal("Expected broker to be of type *natsBroker")
+	}
+
+	if nb.poolSize != 5 {
+		t.Errorf("Expected pool size 5, got %d", nb.poolSize)
+	}
+
+	// Test with custom idle timeout
+	br2 := NewNatsBroker(PoolSize(3), PoolIdleTimeout(10*time.Minute))
+	nb2, ok := br2.(*natsBroker)
+	if !ok {
+		t.Fatal("Expected broker to be of type *natsBroker")
+	}
+
+	if nb2.poolSize != 3 {
+		t.Errorf("Expected pool size 3, got %d", nb2.poolSize)
+	}
+
+	if nb2.poolIdleTimeout != 10*time.Minute {
+		t.Errorf("Expected idle timeout 10m, got %v", nb2.poolIdleTimeout)
+	}
+}
+
+func TestNatsBroker_DefaultSingleConnection(t *testing.T) {
+	// Test that default behavior is single connection (pool size 1)
+	br := NewNatsBroker()
+	nb, ok := br.(*natsBroker)
+	if !ok {
+		t.Fatal("Expected broker to be of type *natsBroker")
+	}
+
+	if nb.poolSize != 1 {
+		t.Errorf("Expected default pool size 1, got %d", nb.poolSize)
+	}
+}

config/source/nats/nats.go

@@ -15,6 +15,7 @@ type nats struct {
 	url    string
 	bucket string
 	key    string
+	conn   *natsgo.Conn     // store connection for lifecycle management
 	kv     natsgo.KeyValue
 	opts   source.Options
 }
@@ -128,7 +129,18 @@ func NewSource(opts ...source.Option) source.Source {
 		url:    config.Url,
 		bucket: bucket,
 		key:    key,
+		conn:   nc,  // store connection reference
 		kv:     kv,
 		opts:   options,
 	}
 }
+
+// Close implements io.Closer and closes the underlying NATS connection.
+// This method is optional but recommended to prevent connection leaks.
+func (n *nats) Close() error {
+	if n.conn != nil {
+		n.conn.Close()
+		n.conn = nil
+	}
+	return nil
+}

events/natsjs/nats.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"io"
 	"strings"
 	"time"
 
@@ -35,11 +36,12 @@ func NewStream(opts ...Option) (events.Stream, error) {
 
 	s := &stream{opts: options}
 
-	natsJetStreamCtx, err := connectToNatsJetStream(options)
+	conn, natsJetStreamCtx, err := connectToNatsJetStream(options)
 	if err != nil {
 		return nil, fmt.Errorf("error connecting to nats cluster %v: %w", options.ClusterID, err)
 	}
 
+	s.conn = conn
 	s.natsJetStreamCtx = natsJetStreamCtx
 
 	return s, nil
@@ -47,10 +49,11 @@ func NewStream(opts ...Option) (events.Stream, error) {
 
 type stream struct {
 	opts             Options
+	conn             *nats.Conn           // store connection for lifecycle management
 	natsJetStreamCtx nats.JetStreamContext
 }
 
-func connectToNatsJetStream(options Options) (nats.JetStreamContext, error) {
+func connectToNatsJetStream(options Options) (*nats.Conn, nats.JetStreamContext, error) {
 	nopts := nats.GetDefaultOptions()
 	if options.TLSConfig != nil {
 		nopts.Secure = true
@@ -77,15 +80,16 @@ func connectToNatsJetStream(options Options) (nats.JetStreamContext, error) {
 	conn, err := nopts.Connect()
 	if err != nil {
 		tls := nopts.TLSConfig != nil
-		return nil, fmt.Errorf("error connecting to nats at %v with tls enabled (%v): %w", options.Address, tls, err)
+		return nil, nil, fmt.Errorf("error connecting to nats at %v with tls enabled (%v): %w", options.Address, tls, err)
 	}
 
 	js, err := conn.JetStream()
 	if err != nil {
-		return nil, fmt.Errorf("error while obtaining JetStream context: %w", err)
+		conn.Close() // Close connection if JetStream context fails
+		return nil, nil, fmt.Errorf("error while obtaining JetStream context: %w", err)
 	}
 
-	return js, nil
+	return conn, js, nil
 }
 
 // Publish a message to a topic.
@@ -268,3 +272,16 @@ func (s *stream) Consume(topic string, opts ...events.ConsumeOption) (<-chan eve
 
 	return channel, nil
 }
+
+// Close implements io.Closer and closes the underlying NATS connection.
+// This method is optional but recommended to prevent connection leaks.
+func (s *stream) Close() error {
+	if s.conn != nil {
+		s.conn.Close()
+		s.conn = nil
+	}
+	return nil
+}
+
+// Ensure stream implements io.Closer
+var _ io.Closer = (*stream)(nil)

transport/nats/nats.go

@@ -16,19 +16,25 @@ import (
 )
 
 type ntport struct {
-	addrs []string
-	opts  transport.Options
-	nopts nats.Options
+	addrs           []string
+	opts            transport.Options
+	nopts           nats.Options
+	pool            *connectionPool  // connection pool for clients
+	poolSize        int
+	poolIdleTimeout time.Duration
+	mu              sync.RWMutex
 }
 
 type ntportClient struct {
-	conn   *nats.Conn
-	addr   string
-	id     string
-	local  string
-	remote string
-	sub    *nats.Subscription
-	opts   transport.Options
+	conn       *nats.Conn
+	pooledConn *pooledConnection  // reference to pooled connection if using pool
+	pool       *connectionPool    // reference to pool to return connection on close
+	addr       string
+	id         string
+	local      string
+	remote     string
+	sub        *nats.Subscription
+	opts       transport.Options
 }
 
 type ntportSocket struct {
@@ -67,8 +73,20 @@ func configure(n *ntport, opts ...transport.Option) {
 	}
 
 	natsOptions := nats.GetDefaultOptions()
-	if n, ok := n.opts.Context.Value(optionsKey{}).(nats.Options); ok {
-		natsOptions = n
+	if no, ok := n.opts.Context.Value(optionsKey{}).(nats.Options); ok {
+		natsOptions = no
+	}
+
+	// Set pool size (default is 1 - no pooling)
+	n.poolSize = 1
+	if poolSize, ok := n.opts.Context.Value(poolSizeKey{}).(int); ok && poolSize > 0 {
+		n.poolSize = poolSize
+	}
+
+	// Set pool idle timeout (default is 5 minutes)
+	n.poolIdleTimeout = 5 * time.Minute
+	if idleTimeout, ok := n.opts.Context.Value(poolIdleTimeoutKey{}).(time.Duration); ok {
+		n.poolIdleTimeout = idleTimeout
 	}
 
 	// transport.Options have higher priority than nats.Options
@@ -91,6 +109,31 @@ func configure(n *ntport, opts ...transport.Option) {
 	n.opts.Addrs = setAddrs(n.opts.Addrs)
 	n.nopts = natsOptions
 	n.addrs = n.opts.Addrs
+
+	// Initialize connection pool if size > 1
+	if n.poolSize > 1 && n.pool == nil {
+		factory := func() (*nats.Conn, error) {
+			opts := n.nopts
+			opts.Servers = n.addrs
+			opts.Secure = n.opts.Secure
+			opts.TLSConfig = n.opts.TLSConfig
+
+			// secure might not be set
+			if n.opts.TLSConfig != nil {
+				opts.Secure = true
+			}
+
+			return opts.Connect()
+		}
+
+		pool, err := newConnectionPool(n.poolSize, factory)
+		if err == nil {
+			if n.poolIdleTimeout > 0 {
+				pool.idleTimeout = n.poolIdleTimeout
+			}
+			n.pool = pool
+		}
+	}
 }
 
 func setAddrs(addrs []string) []string {
@@ -166,6 +209,13 @@ func (n *ntportClient) Recv(m *transport.Message) error {
 
 func (n *ntportClient) Close() error {
 	n.sub.Unsubscribe()
+
+	// If using a pooled connection, return it to the pool
+	if n.pool != nil && n.pooledConn != nil {
+		return n.pool.Put(n.pooledConn)
+	}
+
+	// Otherwise, close the connection directly
 	n.conn.Close()
 	return nil
 }
@@ -344,37 +394,68 @@ func (n *ntport) Dial(addr string, dialOpts ...transport.DialOption) (transport.
 		o(&dopts)
 	}
 
-	opts := n.nopts
-	opts.Servers = n.addrs
-	opts.Secure = n.opts.Secure
-	opts.TLSConfig = n.opts.TLSConfig
-	opts.Timeout = dopts.Timeout
+	var c *nats.Conn
+	var pooledConn *pooledConnection
+	var err error
 
-	// secure might not be set
-	if n.opts.TLSConfig != nil {
-		opts.Secure = true
-	}
+	// Use connection pool if available
+	n.mu.RLock()
+	hasPool := n.pool != nil
+	n.mu.RUnlock()
 
-	c, err := opts.Connect()
-	if err != nil {
-		return nil, err
+	if hasPool {
+		pooledConn, err = n.pool.Get()
+		if err != nil {
+			return nil, err
+		}
+		c = pooledConn.Conn()
+		if c == nil {
+			n.pool.Put(pooledConn)
+			return nil, errors.New("invalid connection from pool")
+		}
+	} else {
+		// Create a new connection (original behavior)
+		opts := n.nopts
+		opts.Servers = n.addrs
+		opts.Secure = n.opts.Secure
+		opts.TLSConfig = n.opts.TLSConfig
+		opts.Timeout = dopts.Timeout
+
+		// secure might not be set
+		if n.opts.TLSConfig != nil {
+			opts.Secure = true
+		}
+
+		c, err = opts.Connect()
+		if err != nil {
+			return nil, err
+		}
 	}
 
 	id := nats.NewInbox()
 	sub, err := c.SubscribeSync(id)
 	if err != nil {
+		if pooledConn != nil {
+			n.pool.Put(pooledConn)
+		} else {
+			c.Close()
+		}
 		return nil, err
 	}
 
-	return &ntportClient{
-		conn:   c,
-		addr:   addr,
-		id:     id,
-		sub:    sub,
-		opts:   n.opts,
-		local:  id,
-		remote: addr,
-	}, nil
+	client := &ntportClient{
+		conn:       c,
+		pooledConn: pooledConn,
+		pool:       n.pool,
+		addr:       addr,
+		id:         id,
+		sub:        sub,
+		opts:       n.opts,
+		local:      id,
+		remote:     addr,
+	}
+
+	return client, nil
 }
 
 func (n *ntport) Listen(addr string, listenOpts ...transport.ListenOption) (transport.Listener, error) {

transport/nats/options.go

@@ -2,12 +2,15 @@ package nats
 
 import (
 	"context"
+	"time"
 
 	"github.com/nats-io/nats.go"
 	"go-micro.dev/v5/transport"
 )
 
 type optionsKey struct{}
+type poolSizeKey struct{}
+type poolIdleTimeoutKey struct{}
 
 // Options allow to inject a nats.Options struct for configuring
 // the nats connection.
@@ -19,3 +22,27 @@ func Options(nopts nats.Options) transport.Option {
 		o.Context = context.WithValue(o.Context, optionsKey{}, nopts)
 	}
 }
+
+// PoolSize sets the size of the connection pool.
+// If set to a value > 1, the transport will use a connection pool.
+// Default is 1 (no pooling).
+func PoolSize(size int) transport.Option {
+	return func(o *transport.Options) {
+		if o.Context == nil {
+			o.Context = context.Background()
+		}
+		o.Context = context.WithValue(o.Context, poolSizeKey{}, size)
+	}
+}
+
+// PoolIdleTimeout sets the timeout for idle connections in the pool.
+// Connections idle for longer than this duration will be closed.
+// Default is 5 minutes. Set to 0 to disable idle timeout.
+func PoolIdleTimeout(timeout time.Duration) transport.Option {
+	return func(o *transport.Options) {
+		if o.Context == nil {
+			o.Context = context.Background()
+		}
+		o.Context = context.WithValue(o.Context, poolIdleTimeoutKey{}, timeout)
+	}
+}

transport/nats/pool.go

@@ -0,0 +1,188 @@
+package nats
+
+import (
+	"errors"
+	"sync"
+	"time"
+
+	natsp "github.com/nats-io/nats.go"
+)
+
+var (
+	// ErrPoolExhausted is returned when no connections are available in the pool
+	ErrPoolExhausted = errors.New("connection pool exhausted")
+	// ErrPoolClosed is returned when trying to use a closed pool
+	ErrPoolClosed = errors.New("connection pool is closed")
+)
+
+// connectionPool manages a pool of NATS connections
+type connectionPool struct {
+	mu          sync.RWMutex
+	connections chan *pooledConnection
+	factory     func() (*natsp.Conn, error)
+	size        int
+	idleTimeout time.Duration
+	closed      bool
+}
+
+// pooledConnection wraps a NATS connection with metadata
+type pooledConnection struct {
+	conn      *natsp.Conn
+	createdAt time.Time
+	lastUsed  time.Time
+	mu        sync.Mutex
+}
+
+// newConnectionPool creates a new connection pool
+func newConnectionPool(size int, factory func() (*natsp.Conn, error)) (*connectionPool, error) {
+	if size <= 0 {
+		size = 1
+	}
+
+	pool := &connectionPool{
+		connections: make(chan *pooledConnection, size),
+		factory:     factory,
+		size:        size,
+		idleTimeout: 5 * time.Minute,
+		closed:      false,
+	}
+
+	return pool, nil
+}
+
+// Get retrieves a connection from the pool or creates a new one
+func (p *connectionPool) Get() (*pooledConnection, error) {
+	p.mu.RLock()
+	if p.closed {
+		p.mu.RUnlock()
+		return nil, ErrPoolClosed
+	}
+	p.mu.RUnlock()
+
+	// Try to get an existing connection from the pool
+	select {
+	case conn := <-p.connections:
+		// Check if connection is still valid and not idle for too long
+		if conn.isValid() && !conn.isExpired(p.idleTimeout) {
+			conn.updateLastUsed()
+			return conn, nil
+		}
+		// Connection is invalid or expired, close it and create a new one
+		conn.close()
+		return p.createConnection()
+	default:
+		// No connection available, create a new one
+		return p.createConnection()
+	}
+}
+
+// Put returns a connection to the pool
+func (p *connectionPool) Put(conn *pooledConnection) error {
+	p.mu.RLock()
+	defer p.mu.RUnlock()
+
+	if p.closed {
+		return conn.close()
+	}
+
+	// Check if connection is still valid
+	if !conn.isValid() {
+		return conn.close()
+	}
+
+	conn.updateLastUsed()
+
+	// Try to return connection to pool
+	select {
+	case p.connections <- conn:
+		return nil
+	default:
+		// Pool is full, close the connection
+		return conn.close()
+	}
+}
+
+// Close closes all connections in the pool
+func (p *connectionPool) Close() error {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+
+	if p.closed {
+		return nil
+	}
+
+	p.closed = true
+	close(p.connections)
+
+	// Close all connections in the pool
+	for conn := range p.connections {
+		conn.close()
+	}
+
+	return nil
+}
+
+// createConnection creates a new pooled connection
+func (p *connectionPool) createConnection() (*pooledConnection, error) {
+	conn, err := p.factory()
+	if err != nil {
+		return nil, err
+	}
+
+	return &pooledConnection{
+		conn:      conn,
+		createdAt: time.Now(),
+		lastUsed:  time.Now(),
+	}, nil
+}
+
+// isValid checks if the underlying NATS connection is valid
+func (pc *pooledConnection) isValid() bool {
+	pc.mu.Lock()
+	defer pc.mu.Unlock()
+
+	if pc.conn == nil {
+		return false
+	}
+
+	status := pc.conn.Status()
+	return status == natsp.CONNECTED || status == natsp.RECONNECTING
+}
+
+// isExpired checks if the connection has been idle for too long
+func (pc *pooledConnection) isExpired(timeout time.Duration) bool {
+	pc.mu.Lock()
+	defer pc.mu.Unlock()
+
+	if timeout <= 0 {
+		return false
+	}
+
+	return time.Since(pc.lastUsed) > timeout
+}
+
+// close closes the underlying NATS connection
+func (pc *pooledConnection) close() error {
+	pc.mu.Lock()
+	defer pc.mu.Unlock()
+
+	if pc.conn != nil {
+		pc.conn.Close()
+		pc.conn = nil
+	}
+	return nil
+}
+
+// Conn returns the underlying NATS connection
+func (pc *pooledConnection) Conn() *natsp.Conn {
+	pc.mu.Lock()
+	defer pc.mu.Unlock()
+	return pc.conn
+}
+
+// updateLastUsed updates the last used timestamp in a thread-safe manner
+func (pc *pooledConnection) updateLastUsed() {
+	pc.mu.Lock()
+	defer pc.mu.Unlock()
+	pc.lastUsed = time.Now()
+}

transport/nats/pool_test.go

@@ -0,0 +1,170 @@
+package nats
+
+import (
+	"sync"
+	"testing"
+	"time"
+
+	natsp "github.com/nats-io/nats.go"
+)
+
+func TestTransportConnectionPool_GetPut(t *testing.T) {
+	// Mock factory that creates connections
+	connCount := 0
+	factory := func() (*natsp.Conn, error) {
+		connCount++
+		return nil, nil
+	}
+
+	pool, err := newConnectionPool(3, factory)
+	if err != nil {
+		t.Fatalf("Failed to create pool: %v", err)
+	}
+	defer pool.Close()
+
+	// Get a connection (should create one)
+	conn1, err := pool.Get()
+	if err != nil {
+		t.Fatalf("Failed to get connection: %v", err)
+	}
+	if conn1 == nil {
+		t.Fatal("Expected connection, got nil")
+	}
+
+	// Put it back
+	if err := pool.Put(conn1); err != nil {
+		t.Fatalf("Failed to put connection: %v", err)
+	}
+
+	// Get it again (should reuse the same one)
+	conn2, err := pool.Get()
+	if err != nil {
+		t.Fatalf("Failed to get connection: %v", err)
+	}
+
+	if conn2 == nil {
+		t.Fatal("Expected connection, got nil")
+	}
+}
+
+func TestTransportConnectionPool_Concurrent(t *testing.T) {
+	connCount := 0
+	mu := sync.Mutex{}
+	factory := func() (*natsp.Conn, error) {
+		mu.Lock()
+		connCount++
+		mu.Unlock()
+		return nil, nil
+	}
+
+	pool, err := newConnectionPool(5, factory)
+	if err != nil {
+		t.Fatalf("Failed to create pool: %v", err)
+	}
+	defer pool.Close()
+
+	// Simulate concurrent access
+	var wg sync.WaitGroup
+	for i := 0; i < 10; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			conn, err := pool.Get()
+			if err != nil {
+				t.Errorf("Failed to get connection: %v", err)
+				return
+			}
+			// Simulate some work
+			time.Sleep(10 * time.Millisecond)
+			if err := pool.Put(conn); err != nil {
+				t.Errorf("Failed to put connection: %v", err)
+			}
+		}()
+	}
+
+	wg.Wait()
+
+	// We should have created some connections
+	mu.Lock()
+	if connCount == 0 {
+		t.Error("Expected at least one connection to be created")
+	}
+	mu.Unlock()
+}
+
+func TestTransportConnectionPool_Close(t *testing.T) {
+	factory := func() (*natsp.Conn, error) {
+		return nil, nil
+	}
+
+	pool, err := newConnectionPool(3, factory)
+	if err != nil {
+		t.Fatalf("Failed to create pool: %v", err)
+	}
+
+	// Get a connection
+	conn, err := pool.Get()
+	if err != nil {
+		t.Fatalf("Failed to get connection: %v", err)
+	}
+
+	// Close the pool
+	if err := pool.Close(); err != nil {
+		t.Fatalf("Failed to close pool: %v", err)
+	}
+
+	// Put connection back to closed pool should not panic
+	_ = pool.Put(conn)
+
+	// Try to get from closed pool
+	_, err = pool.Get()
+	if err != ErrPoolClosed {
+		t.Errorf("Expected ErrPoolClosed, got: %v", err)
+	}
+}
+
+func TestTransportPoolConfiguration(t *testing.T) {
+	// Test with pool size 5
+	tr := NewTransport(PoolSize(5))
+	nt, ok := tr.(*ntport)
+	if !ok {
+		t.Fatal("Expected transport to be of type *ntport")
+	}
+
+	if nt.poolSize != 5 {
+		t.Errorf("Expected pool size 5, got %d", nt.poolSize)
+	}
+
+	// Test with custom idle timeout
+	tr2 := NewTransport(PoolSize(3), PoolIdleTimeout(10*time.Minute))
+	nt2, ok := tr2.(*ntport)
+	if !ok {
+		t.Fatal("Expected transport to be of type *ntport")
+	}
+
+	if nt2.poolSize != 3 {
+		t.Errorf("Expected pool size 3, got %d", nt2.poolSize)
+	}
+
+	if nt2.poolIdleTimeout != 10*time.Minute {
+		t.Errorf("Expected idle timeout 10m, got %v", nt2.poolIdleTimeout)
+	}
+}
+
+func TestTransportDefaultSingleConnection(t *testing.T) {
+	// Test that default behavior is single connection (pool size 1)
+	tr := NewTransport()
+	nt, ok := tr.(*ntport)
+	if !ok {
+		t.Fatal("Expected transport to be of type *ntport")
+	}
+
+	if nt.poolSize != 1 {
+		t.Errorf("Expected default pool size 1, got %d", nt.poolSize)
+	}
+
+	// With size 1, pool should not be created
+	if nt.pool != nil {
+		t.Error("Expected no pool with size 1")
+	}
+}