Use the specified connection timeout when creating connections (apach…

…e#137)

* Use the specified connection timeout when creating connections

* Fixed compile and other errors

pulsar/client.go

@@ -63,6 +63,7 @@ type ClientOptions struct {
 	// This parameter is required
 	URL string
 
+	// Timeout for the establishment of a TCP connection (default: 30 seconds)
 	ConnectionTimeout time.Duration
 
 	// Set the operation timeout (default: 30 seconds)

pulsar/client_impl.go

@@ -21,6 +21,7 @@ import (
 	"errors"
 	"fmt"
 	"net/url"
+	"time"
 
 	"github.com/golang/protobuf/proto"
 
@@ -31,6 +32,10 @@ import (
 	"github.com/apache/pulsar-client-go/pulsar/internal/pb"
 )
 
+const (
+	defaultConnectionTimeout = 30*time.Second
+)
+
 type client struct {
 	options ClientOptions
 
@@ -81,8 +86,13 @@ func newClient(options ClientOptions) (Client, error) {
 		}
 	}
 
+	connectionTimeout := options.ConnectionTimeout
+	if connectionTimeout.Nanoseconds() == 0 {
+		connectionTimeout = defaultConnectionTimeout
+	}
+
 	c := &client{
-		cnxPool: internal.NewConnectionPool(tlsConfig, authProvider),
+		cnxPool: internal.NewConnectionPool(tlsConfig, authProvider, connectionTimeout),
 	}
 	c.rpcClient = internal.NewRPCClient(url, c.cnxPool)
 	c.lookupService = internal.NewLookupService(c.rpcClient, url)

pulsar/internal/connection.go

@@ -115,8 +115,9 @@ type incomingCmd struct {
 
 type connection struct {
 	sync.Mutex
-	cond  *sync.Cond
-	state connectionState
+	cond              *sync.Cond
+	state             connectionState
+	connectionTimeout time.Duration
 
 	logicalAddr  *url.URL
 	physicalAddr *url.URL
@@ -150,9 +151,11 @@ type connection struct {
 	auth       auth.Provider
 }
 
-func newConnection(logicalAddr *url.URL, physicalAddr *url.URL, tlsOptions *TLSOptions, auth auth.Provider) *connection {
+func newConnection(logicalAddr *url.URL, physicalAddr *url.URL, tlsOptions *TLSOptions,
+	connectionTimeout time.Duration, auth auth.Provider) *connection {
 	cnx := &connection{
 		state:                connectionInit,
+		connectionTimeout:    connectionTimeout,
 		logicalAddr:          logicalAddr,
 		physicalAddr:         physicalAddr,
 		writeBuffer:          NewBuffer(4096),
@@ -202,7 +205,7 @@ func (c *connection) connect() bool {
 
 	if c.tlsOptions == nil {
 		// Clear text connection
-		cnx, err = net.Dial("tcp", c.physicalAddr.Host)
+		cnx, err = net.DialTimeout("tcp", c.physicalAddr.Host, c.connectionTimeout)
 	} else {
 		// TLS connection
 		tlsConfig, err = c.getTLSConfig()
@@ -211,7 +214,8 @@ func (c *connection) connect() bool {
 			return false
 		}
 
-		cnx, err = tls.Dial("tcp", c.physicalAddr.Host, tlsConfig)
+		d := &net.Dialer{Timeout: c.connectionTimeout}
+		cnx, err = tls.DialWithDialer(d, "tcp", c.physicalAddr.Host, tlsConfig)
 	}
 
 	if err != nil {

pulsar/internal/connection_pool.go

@@ -20,6 +20,7 @@ package internal
 import (
 	"net/url"
 	"sync"
+	"time"
 
 	"github.com/apache/pulsar-client-go/pulsar/internal/auth"
 
@@ -36,16 +37,18 @@ type ConnectionPool interface {
 }
 
 type connectionPool struct {
-	pool       sync.Map
-	tlsOptions *TLSOptions
-	auth       auth.Provider
+	pool              sync.Map
+	connectionTimeout time.Duration
+	tlsOptions        *TLSOptions
+	auth              auth.Provider
 }
 
 // NewConnectionPool init connection pool.
-func NewConnectionPool(tlsOptions *TLSOptions, auth auth.Provider) ConnectionPool {
+func NewConnectionPool(tlsOptions *TLSOptions, auth auth.Provider, connectionTimeout time.Duration) ConnectionPool {
 	return &connectionPool{
-		tlsOptions: tlsOptions,
-		auth:       auth,
+		tlsOptions:        tlsOptions,
+		auth:              auth,
+		connectionTimeout: connectionTimeout,
 	}
 }
 
@@ -66,7 +69,7 @@ func (p *connectionPool) GetConnection(logicalAddr *url.URL, physicalAddr *url.U
 
 	// Try to create a new connection
 	newCnx, wasCached := p.pool.LoadOrStore(logicalAddr.Host,
-		newConnection(logicalAddr, physicalAddr, p.tlsOptions, p.auth))
+		newConnection(logicalAddr, physicalAddr, p.tlsOptions, p.connectionTimeout, p.auth))
 	cnx := newCnx.(*connection)
 	if !wasCached {
 		cnx.start()