func (c *Connection) DoCommand(ctx context.Context, name string, timeout time.Duration,
	rpcFunc func(GenericClient) error) error {
	if timeout > 0 {
		var timeoutCancel context.CancelFunc
		ctx, timeoutCancel = context.WithTimeout(ctx, timeout)
		defer timeoutCancel()
	}
	for {
		if (c.firstConnectDelayDuration != 0 ||
			c.initialReconnectBackoffWindow != nil) && isWithFireNow(ctx) {
			c.connectDelayTimer.FireNow()
		}
		connErr := c.waitForConnection(ctx, false)
		if connErr != nil {
			return connErr
		}
		var rpcErr error
		throttleErr := backoff.RetryNotify(func() error {
			rawClient := func() GenericClient {
				c.mutex.Lock()
				defer c.mutex.Unlock()
				return c.client
			}()
			throttleErr := rpcFunc(rawClient)
			if throttleErr != nil && c.handler.ShouldRetry(name, throttleErr) {
				return throttleErr
			}
			rpcErr = throttleErr
			return nil
		}, c.doCommandBackoff(), c.handler.OnDoCommandError)
		if throttleErr != nil {
			return throttleErr
		}
		if !c.checkForRetry(rpcErr) {
			return rpcErr
		}
	}
}
DoCommand executes the specific rpc command wrapped in rpcFunc.
package rpc
import (
	"crypto/tls"
	"crypto/x509"
	"errors"
	"io"
	"net"
	"sync"
	"time"
	"github.com/keybase/backoff"
	"github.com/keybase/go-framed-msgpack-rpc/rpc/resinit"
	"golang.org/x/net/context"
)
type DisconnectStatus int
const (
	_ = iota
	UsingExistingConnection DisconnectStatus = iota
	StartingFirstConnection
	StartingNonFirstConnection
)
type Dialable interface {
	SetOpts(timeout time.Duration, keepAlive time.Duration)
	Dial(ctx context.Context, network string, addr string) (net.Conn, error)
}
type ConnectionTransport interface {
	Dial(ctx context.Context) (Transporter, error)
	IsConnected() bool
	Finalize()
	Close()
}
type connTransport struct {
	uri                 *FMPURI
	instrumenterStorage NetworkInstrumenterStorage
	l                   LogFactory
	wef                 WrapErrorFunc
	maxFrameLength      int32
	dialable            Dialable
	conn                net.Conn
	transport           Transporter
	stagedTransport     Transporter
}
var _ ConnectionTransport = (*connTransport)(nil)
func NewConnectionTransport(uri *FMPURI, l LogFactory, instrumenterStorage NetworkInstrumenterStorage,
	wef WrapErrorFunc, maxFrameLength int32) ConnectionTransport {
	return &connTransport{
		uri:                 uri,
		l:                   l,
		instrumenterStorage: instrumenterStorage,
		wef:                 wef,
		maxFrameLength:      maxFrameLength,
	}
}
func NewConnectionTransportWithDialable(uri *FMPURI, l LogFactory, instrumenterStorage NetworkInstrumenterStorage,
	wef WrapErrorFunc, maxFrameLength int32, dialable Dialable) ConnectionTransport {
	return &connTransport{
		uri:                 uri,
		l:                   l,
		instrumenterStorage: instrumenterStorage,
		wef:                 wef,
		maxFrameLength:      maxFrameLength,
		dialable:            dialable,
	}
}
func (t *connTransport) Dial(ctx context.Context) (Transporter, error) {
	var err error
	if t.conn != nil {
		t.conn.Close()
	}
	if t.dialable != nil {
		t.conn, err = t.dialable.Dial(ctx, "tcp", t.uri.HostPort)
	} else {
		t.conn, err = t.uri.Dial()
	}
	if err != nil {
		resinit.ResInitIfDNSError(err)
		return nil, err
	}
	if err = DisableSigPipe(t.conn); err != nil {
		return nil, err
	}
	if t.stagedTransport != nil {
		t.stagedTransport.Close()
	}
	t.stagedTransport = NewTransport(t.conn, t.l, t.instrumenterStorage, t.wef, t.maxFrameLength)
	return t.stagedTransport, nil
}
func (t *connTransport) IsConnected() bool {
	return t.transport != nil && t.transport.IsConnected()
}
func (t *connTransport) Finalize() {
	if t.transport != nil {
		t.transport.Close()
	}
	t.transport = t.stagedTransport
	t.stagedTransport = nil
}
func (t *connTransport) Close() {
	t.conn.Close()
	if t.transport != nil {
		t.transport.Close()
	}
	t.transport = nil
	if t.stagedTransport != nil {
		t.stagedTransport.Close()
	}
	t.stagedTransport = nil
}
type ConnectionHandler interface {
	OnConnect(context.Context, *Connection, GenericClient, *Server) error
	OnConnectError(err error, reconnectThrottleDuration time.Duration)
	OnDoCommandError(err error, nextTime time.Duration)
	OnDisconnected(ctx context.Context, status DisconnectStatus)
	ShouldRetry(name string, err error) bool
	ShouldRetryOnConnect(err error) bool
	HandlerName() string
}
type ConnectionTransportTLS struct {
	rootCerts      []byte
	srvRemote      Remote
	tlsConfig      *tls.Config
	maxFrameLength int32
	dialable       Dialable
	mutex               sync.Mutex
	transport           Transporter
	stagedTransport     Transporter
	conn                net.Conn
	dialerTimeout       time.Duration
	handshakeTimeout    time.Duration
	instrumenterStorage NetworkInstrumenterStorage
	logFactory          LogFactory
	wef                 WrapErrorFunc
	log                 ConnectionLog
}
var _ ConnectionTransport = (*ConnectionTransportTLS)(nil)
const keepAlive = 10 * time.Second
func (ct *ConnectionTransportTLS) Dial(ctx context.Context) (
	Transporter, error) {
	addr := ct.srvRemote.GetAddress()
	config := ct.tlsConfig
	host, _, err := net.SplitHostPort(addr)
	if err != nil {
		return nil, err
	}
	if config == nil && ct.rootCerts != nil {
		certs := x509.NewCertPool()
		if !certs.AppendCertsFromPEM(ct.rootCerts) {
			return nil, errors.New("Unable to load root certificates")
		}
		config = &tls.Config{
			RootCAs:    certs,
			ServerName: host,
		}
	}
	if config == nil {
		config = &tls.Config{ServerName: host}
	}
	ct.log.Debug("%s %s",
		LogField{Key: ConnectionLogMsgKey, Value: "Dialing"},
		LogField{Key: "remote-addr", Value: addr})
	var baseConn net.Conn
	if ct.dialable != nil {
		ct.dialable.SetOpts(ct.dialerTimeout, keepAlive)
		baseConn, err = ct.dialable.Dial(ctx, "tcp", addr)
	} else {
		dialer := net.Dialer{
			Timeout:   ct.dialerTimeout,
			KeepAlive: keepAlive,
		}
		baseConn, err = dialer.DialContext(ctx, "tcp", addr)
	}
	if err != nil {
		resinit.ResInitIfDNSError(err)
		return nil, err
	}
	ct.log.Debug("baseConn: %s; Calling %s",
		LogField{Key: "local-addr", Value: baseConn.LocalAddr()},
		LogField{Key: ConnectionLogMsgKey, Value: "Handshake"})
	conn := tls.Client(baseConn, config)
	errCh := make(chan error, 1)
	go func() {
		errCh <- conn.Handshake()
	}()
	handshakeTimeout := ct.handshakeTimeout
	if handshakeTimeout == 0 {
		handshakeTimeout = time.Minute
	}
	select {
	case err := <-errCh:
		if err != nil {
			return nil, err
		}
	case <-time.After(handshakeTimeout):
		return nil, errors.New("handshake timeout")
	}
	ct.log.Debug("%s", LogField{Key: ConnectionLogMsgKey, Value: "Handshaken"})
	err = DisableSigPipe(baseConn)
	if err != nil {
		return nil, err
	}
	ct.mutex.Lock()
	defer ct.mutex.Unlock()
	if ct.conn != nil {
		ct.conn.Close()
	}
	transport := NewTransport(conn, ct.logFactory, ct.instrumenterStorage, ct.wef, ct.maxFrameLength)
	ct.conn = conn
	if ct.stagedTransport != nil {
		ct.stagedTransport.Close()
	}
	ct.stagedTransport = transport
	return transport, nil
}
func (ct *ConnectionTransportTLS) IsConnected() bool {
	ct.mutex.Lock()
	defer ct.mutex.Unlock()
	return ct.transport != nil && ct.transport.IsConnected()
}
func (ct *ConnectionTransportTLS) Finalize() {
	ct.mutex.Lock()
	defer ct.mutex.Unlock()
	if ct.transport != nil {
		ct.transport.Close()
	}
	ct.transport = ct.stagedTransport
	ct.stagedTransport = nil
	ct.srvRemote.Reset()
}
func (ct *ConnectionTransportTLS) Close() {
	ct.mutex.Lock()
	defer ct.mutex.Unlock()
	if ct.conn != nil {
		ct.conn.Close()
	}
	if ct.transport != nil {
		ct.transport.Close()
	}
	if ct.stagedTransport != nil {
		ct.stagedTransport.Close()
	}
}
type LogTagsFromContext func(ctx context.Context) (map[interface{}]string, bool)
type Connection struct {
	handler          ConnectionHandler
	transport        ConnectionTransport
	errorUnwrapper   ErrorUnwrapper
	reconnectBackoff func() backoff.BackOff
	doCommandBackoff func() backoff.BackOff
	wef              WrapErrorFunc
	tagsFunc         LogTagsFromContext
	log              ConnectionLog
	protocols        []Protocol
	mutex             sync.Mutex
	client            GenericClient
	server            *Server
	reconnectChan     chan struct{}
	reconnectErrPtr   *error              
	cancelFunc        context.CancelFunc  
	reconnectedBefore bool
	firstConnectDelayDuration     time.Duration
	initialReconnectBackoffWindow func() time.Duration
	connectDelayTimer             CancellableTimer
	reconnectCompleteForTest chan struct{}
}
type ConnectionOpts struct {
	TagsFunc         LogTagsFromContext
	Protocols        []Protocol
	DontConnectNow   bool
	WrapErrorFunc    WrapErrorFunc
	ReconnectBackoff func() backoff.BackOff
	CommandBackoff   func() backoff.BackOff
	FirstConnectDelayDuration time.Duration
	InitialReconnectBackoffWindow func() time.Duration
	ForceInitialBackoff bool
	DialerTimeout time.Duration
	HandshakeTimeout time.Duration
}
func NewTLSConnectionWithConnectionLogFactory(
	srvRemote Remote,
	rootCerts []byte,
	errorUnwrapper ErrorUnwrapper,
	handler ConnectionHandler,
	logFactory LogFactory,
	instrumenterStorage NetworkInstrumenterStorage,
	connectionLogFactory ConnectionLogFactory,
	maxFrameLength int32,
	opts ConnectionOpts,
) *Connection {
	transport := &ConnectionTransportTLS{
		rootCerts:           rootCerts,
		srvRemote:           srvRemote,
		maxFrameLength:      maxFrameLength,
		instrumenterStorage: instrumenterStorage,
		logFactory:          logFactory,
		wef:                 opts.WrapErrorFunc,
		dialerTimeout:       opts.DialerTimeout,
		handshakeTimeout:    opts.HandshakeTimeout,
		log:                 connectionLogFactory.Make("conn_tspt"),
	}
	connLog := connectionLogFactory.Make("conn")
	return newConnectionWithTransportAndProtocolsWithLog(
		handler, transport, errorUnwrapper, connLog, opts)
}
func NewTLSConnection(
	srvRemote Remote,
	rootCerts []byte,
	errorUnwrapper ErrorUnwrapper,
	handler ConnectionHandler,
	logFactory LogFactory,
	instrumenterStorage NetworkInstrumenterStorage,
	logOutput LogOutputWithDepthAdder,
	maxFrameLength int32,
	opts ConnectionOpts,
) *Connection {
	transport := &ConnectionTransportTLS{
		rootCerts:           rootCerts,
		srvRemote:           srvRemote,
		maxFrameLength:      maxFrameLength,
		logFactory:          logFactory,
		instrumenterStorage: instrumenterStorage,
		wef:                 opts.WrapErrorFunc,
		dialerTimeout:       opts.DialerTimeout,
		handshakeTimeout:    opts.HandshakeTimeout,
		log:                 newConnectionLogUnstructured(logOutput, "CONNTSPT"),
	}
	return newConnectionWithTransportAndProtocols(handler, transport, errorUnwrapper, logOutput, opts)
}
func NewTLSConnectionWithTLSConfig(
	srvRemote Remote,
	tlsConfig *tls.Config,
	errorUnwrapper ErrorUnwrapper,
	handler ConnectionHandler,
	logFactory LogFactory,
	instrumenterStorage NetworkInstrumenterStorage,
	logOutput LogOutputWithDepthAdder,
	maxFrameLength int32,
	opts ConnectionOpts,
) *Connection {
	transport := &ConnectionTransportTLS{
		srvRemote:           srvRemote,
		tlsConfig:           copyTLSConfig(tlsConfig),
		maxFrameLength:      maxFrameLength,
		logFactory:          logFactory,
		instrumenterStorage: instrumenterStorage,
		wef:                 opts.WrapErrorFunc,
		dialerTimeout:       opts.DialerTimeout,
		handshakeTimeout:    opts.HandshakeTimeout,
		log:                 newConnectionLogUnstructured(logOutput, "CONNTSPT"),
	}
	return newConnectionWithTransportAndProtocols(handler, transport, errorUnwrapper, logOutput, opts)
}
func NewTLSConnectionWithDialable(
	srvRemote Remote,
	rootCerts []byte,
	errorUnwrapper ErrorUnwrapper,
	handler ConnectionHandler,
	logFactory LogFactory,
	instrumenterStorage NetworkInstrumenterStorage,
	logOutput LogOutputWithDepthAdder,
	maxFrameLength int32,
	opts ConnectionOpts,
	dialable Dialable,
) *Connection {
	transport := &ConnectionTransportTLS{
		rootCerts:           rootCerts,
		srvRemote:           srvRemote,
		maxFrameLength:      maxFrameLength,
		logFactory:          logFactory,
		instrumenterStorage: instrumenterStorage,
		wef:                 opts.WrapErrorFunc,
		dialerTimeout:       opts.DialerTimeout,
		handshakeTimeout:    opts.HandshakeTimeout,
		log:                 newConnectionLogUnstructured(logOutput, "CONNTSPT"),
		dialable:            dialable,
	}
	return newConnectionWithTransportAndProtocols(handler, transport, errorUnwrapper, logOutput, opts)
}
func NewConnectionWithTransport(
	handler ConnectionHandler,
	transport ConnectionTransport,
	errorUnwrapper ErrorUnwrapper,
	logOutput LogOutputWithDepthAdder,
	opts ConnectionOpts,
) *Connection {
	return newConnectionWithTransportAndProtocols(handler, transport, errorUnwrapper, logOutput, opts)
}
func newConnectionWithTransportAndProtocolsWithLog(handler ConnectionHandler,
	transport ConnectionTransport, errorUnwrapper ErrorUnwrapper,
	log ConnectionLog, opts ConnectionOpts) *Connection {
	defaultBackoff := func() backoff.BackOff {
		b := backoff.NewExponentialBackOff()
		b.MaxElapsedTime = 0
		b.MaxInterval = 10 * time.Second
		return b
	}
	reconnectBackoff := opts.ReconnectBackoff
	if reconnectBackoff == nil {
		reconnectBackoff = defaultBackoff
	}
	commandBackoff := opts.CommandBackoff
	if commandBackoff == nil {
		commandBackoff = defaultBackoff
	}
	connection := &Connection{
		handler:                       handler,
		transport:                     transport,
		errorUnwrapper:                errorUnwrapper,
		reconnectBackoff:              reconnectBackoff,
		doCommandBackoff:              commandBackoff,
		firstConnectDelayDuration:     opts.FirstConnectDelayDuration,
		initialReconnectBackoffWindow: opts.InitialReconnectBackoffWindow,
		wef:                           opts.WrapErrorFunc,
		tagsFunc:                      opts.TagsFunc,
		log:                           log,
		protocols:                     opts.Protocols,
		reconnectedBefore:             opts.ForceInitialBackoff,
	}
	if !opts.DontConnectNow {
		connection.getReconnectChan()
	}
	return connection
}
func newConnectionWithTransportAndProtocols(handler ConnectionHandler,
	transport ConnectionTransport, errorUnwrapper ErrorUnwrapper,
	log LogOutputWithDepthAdder, opts ConnectionOpts) *Connection {
	return newConnectionWithTransportAndProtocolsWithLog(
		handler, transport, errorUnwrapper,
		newConnectionLogUnstructured(log, "CONN "+handler.HandlerName()), opts)
}
func (c *Connection) setReconnectCompleteForTest(ch chan struct{}) {
	c.mutex.Lock()
	defer c.mutex.Unlock()
	c.reconnectCompleteForTest = ch
}
func (c *Connection) connect(ctx context.Context) error {
	c.log.Debug("Connection: %s",
		LogField{Key: ConnectionLogMsgKey, Value: "dialing transport"})
	transport, err := c.transport.Dial(ctx)
	if err != nil {
		c.log.Warning("Connection: error %s: %v",
			LogField{Key: ConnectionLogMsgKey, Value: "dialing transport"},
			LogField{Key: "error", Value: err})
		return err
	}
	client := NewClient(transport, c.errorUnwrapper, c.tagsFunc)
	server := NewServer(transport, c.wef)
	for _, p := range c.protocols {
		if err := server.Register(p); err != nil {
			return err
		}
	}
	c.log.Debug("Connection: %s", LogField{Key: ConnectionLogMsgKey, Value: "calling OnConnect"})
	err = c.handler.OnConnect(ctx, c, client, server)
	if err != nil {
		c.log.Warning("Connection: error calling %s handler: %v",
			LogField{Key: ConnectionLogMsgKey, Value: "OnConnect"},
			LogField{Key: "error", Value: err})
		return err
	}
	c.mutex.Lock()
	defer c.mutex.Unlock()
	c.client = client
	c.server = server
	c.transport.Finalize()
	c.log.Debug("Connection: %s", LogField{Key: ConnectionLogMsgKey, Value: "connected"})
	return nil
}