Support for factory method for outbound connections (#759)

New optional factory method on ChannelOptions to allow passing in custom dialer
for outbound connections. This could be used for things like TLS handshake
considering tchannel already has support for custom TLS listener.

channel.go

@@ -113,6 +113,10 @@ type ChannelOptions struct {
 	// Handler is an alternate handler for all inbound requests, overriding the
 	// default handler that delegates to a subchannel.
 	Handler Handler
+
+	// Dialer is optional factory method which can be used for overriding
+	// outbound connections for things like TLS handshake
+	Dialer func(ctx context.Context, network, hostPort string) (net.Conn, error)
 }
 
 // ChannelState is the state of a channel.
@@ -158,6 +162,7 @@ type Channel struct {
 	internalHandlers    *handlerMap
 	handler             Handler
 	onPeerStatusChanged func(*Peer)
+	dialer              func(ctx context.Context, hostPort string) (net.Conn, error)
 	closed              chan struct{}
 
 	// mutable contains all the members of Channel which are mutable.
@@ -244,6 +249,14 @@ func NewChannel(serviceName string, opts *ChannelOptions) (*Channel, error) {
 		return nil, err
 	}
 
+	// Default to dialContext if dialer is not passed in as an option
+	dialCtx := dialContext
+	if opts.Dialer != nil {
+		dialCtx = func (ctx context.Context, hostPort string) (net.Conn, error) {
+			return opts.Dialer(ctx, "tcp", hostPort)
+		}
+	}
+
 	ch := &Channel{
 		channelConnectionCommon: channelConnectionCommon{
 			log:           logger,
@@ -259,6 +272,7 @@ func NewChannel(serviceName string, opts *ChannelOptions) (*Channel, error) {
 		relayHost:         opts.RelayHost,
 		relayMaxTimeout:   validateRelayMaxTimeout(opts.RelayMaxTimeout, logger),
 		relayTimerVerify:  opts.RelayTimerVerification,
+		dialer:            dialCtx,
 		closed:            make(chan struct{}),
 	}
 	ch.peers = newRootPeerList(ch, opts.OnPeerStatusChanged).newChild()
@@ -563,7 +577,7 @@ func (ch *Channel) Connect(ctx context.Context, hostPort string) (*Connection, e
 	}
 
 	timeout := getTimeout(ctx)
-	tcpConn, err := dialContext(ctx, hostPort)
+	tcpConn, err := ch.dialer(ctx, hostPort)
 	if err != nil {
 		if ne, ok := err.(net.Error); ok && ne.Timeout() {
 			ch.log.WithFields(

connection_test.go

@@ -1322,3 +1322,27 @@ func TestInvalidTransportHeaders(t *testing.T) {
 		})
 	}
 }
+
+func TestCustomDialer(t *testing.T) {
+	sopts := testutils.NewOpts()
+	testutils.WithTestServer(t, sopts, func(t testing.TB, ts *testutils.TestServer) {
+		server := ts.Server()
+		testutils.RegisterEcho(server, nil)
+		customDialerCalledCount := 0
+
+		copts := testutils.NewOpts().SetDialer(func(ctx context.Context, network, hostPort string) (net.Conn, error) {
+			customDialerCalledCount++
+			d := net.Dialer{}
+			return d.DialContext(ctx, network, hostPort)
+		})
+
+		// Induce the creation of a connection from client to server.
+		client := ts.NewClient(copts)
+		testutils.AssertEcho(t, client, ts.HostPort(), ts.ServiceName())
+		assert.Equal(t, 1, customDialerCalledCount, "custom dialer used for establishing connection")
+
+		// Re-use
+		testutils.AssertEcho(t, client, ts.HostPort(), ts.ServiceName())
+		assert.Equal(t, 1, customDialerCalledCount, "custom dialer used for establishing connection")
+	})
+}

testutils/channel_opts.go

@@ -22,13 +22,15 @@ package testutils
 
 import (
 	"flag"
+	"net"
 	"testing"
 	"time"
 
 	"github.com/uber/tchannel-go"
 	"github.com/uber/tchannel-go/tos"
 
 	"go.uber.org/atomic"
+	"golang.org/x/net/context"
 )
 
 var connectionLog = flag.Bool("connectionLog", false, "Enables connection logging in tests")
@@ -233,6 +235,12 @@ func (o *ChannelOpts) SetIdleCheckInterval(d time.Duration) *ChannelOpts {
 	return o
 }
 
+// SetDialer sets the dialer used for outbound connections
+func (o *ChannelOpts) SetDialer(f func(context.Context, string, string) (net.Conn, error)) *ChannelOpts {
+	o.ChannelOptions.Dialer = f
+	return o
+}
+
 func defaultString(v string, defaultValue string) string {
 	if v == "" {
 		return defaultValue