move serve and connecter into transports package

daemon/transport/connecter/connect_local.go

@@ -0,0 +1,30 @@
+package connecter
+
+import (
+	"context"
+	"fmt"
+	"github.com/zrepl/zrepl/config"
+	"github.com/zrepl/zrepl/daemon/transport/serve"
+	"net"
+)
+
+type LocalConnecter struct {
+	listenerName string
+	clientIdentity string
+}
+
+func LocalConnecterFromConfig(in *config.LocalConnect) (*LocalConnecter, error) {
+	if in.ClientIdentity == "" {
+		return nil, fmt.Errorf("ClientIdentity must not be empty")
+	}
+	if in.ListenerName == "" {
+		return nil, fmt.Errorf("ListenerName must not be empty")
+	}
+	return &LocalConnecter{listenerName: in.ListenerName, clientIdentity: in.ClientIdentity}, nil
+}
+
+func (c *LocalConnecter) Connect(dialCtx context.Context) (conn net.Conn, err error) {
+	l := serve.GetLocalListener(c.listenerName)
+	return l.Connect(dialCtx, c.clientIdentity)
+}
+

daemon/transport/connecter/connect_ssh.go

@@ -0,0 +1,66 @@
+package connecter
+
+import (
+	"context"
+	"github.com/jinzhu/copier"
+	"github.com/pkg/errors"
+	"github.com/problame/go-netssh"
+	"github.com/problame/go-streamrpc"
+	"github.com/zrepl/zrepl/config"
+	"net"
+	"time"
+)
+
+type SSHStdinserverConnecter struct {
+	Host                 string
+	User                 string
+	Port                 uint16
+	IdentityFile         string
+	TransportOpenCommand []string
+	SSHCommand           string
+	Options              []string
+	dialTimeout          time.Duration
+}
+
+var _ streamrpc.Connecter = &SSHStdinserverConnecter{}
+
+func SSHStdinserverConnecterFromConfig(in *config.SSHStdinserverConnect) (c *SSHStdinserverConnecter, err error) {
+
+	c = &SSHStdinserverConnecter{
+		Host:         in.Host,
+		User:         in.User,
+		Port:         in.Port,
+		IdentityFile: in.IdentityFile,
+		SSHCommand:   in.SSHCommand,
+		Options:      in.Options,
+		dialTimeout:  in.DialTimeout,
+	}
+	return
+
+}
+
+type netsshConnToConn struct{ *netssh.SSHConn }
+
+var _ net.Conn = netsshConnToConn{}
+
+func (netsshConnToConn) SetDeadline(dl time.Time) error      { return nil }
+func (netsshConnToConn) SetReadDeadline(dl time.Time) error  { return nil }
+func (netsshConnToConn) SetWriteDeadline(dl time.Time) error { return nil }
+
+func (c *SSHStdinserverConnecter) Connect(dialCtx context.Context) (net.Conn, error) {
+
+	var endpoint netssh.Endpoint
+	if err := copier.Copy(&endpoint, c); err != nil {
+		return nil, errors.WithStack(err)
+	}
+	dialCtx, dialCancel := context.WithTimeout(dialCtx, c.dialTimeout) // context.TODO tied to error handling below
+	defer dialCancel()
+	nconn, err := netssh.Dial(dialCtx, endpoint)
+	if err != nil {
+		if err == context.DeadlineExceeded {
+			err = errors.Errorf("dial_timeout of %s exceeded", c.dialTimeout)
+		}
+		return nil, err
+	}
+	return netsshConnToConn{nconn}, nil
+}

daemon/transport/connecter/connect_tcp.go

@@ -0,0 +1,24 @@
+package connecter
+
+import (
+	"context"
+	"github.com/zrepl/zrepl/config"
+	"net"
+)
+
+type TCPConnecter struct {
+	Address string
+	dialer  net.Dialer
+}
+
+func TCPConnecterFromConfig(in *config.TCPConnect) (*TCPConnecter, error) {
+	dialer := net.Dialer{
+		Timeout: in.DialTimeout,
+	}
+
+	return &TCPConnecter{in.Address, dialer}, nil
+}
+
+func (c *TCPConnecter) Connect(dialCtx context.Context) (conn net.Conn, err error) {
+	return c.dialer.DialContext(dialCtx, "tcp", c.Address)
+}

daemon/transport/connecter/connect_tls.go

@@ -0,0 +1,43 @@
+package connecter
+
+import (
+	"context"
+	"crypto/tls"
+	"github.com/pkg/errors"
+	"github.com/zrepl/zrepl/config"
+	"github.com/zrepl/zrepl/tlsconf"
+	"net"
+)
+
+type TLSConnecter struct {
+	Address   string
+	dialer    net.Dialer
+	tlsConfig *tls.Config
+}
+
+func TLSConnecterFromConfig(in *config.TLSConnect) (*TLSConnecter, error) {
+	dialer := net.Dialer{
+		Timeout: in.DialTimeout,
+	}
+
+	ca, err := tlsconf.ParseCAFile(in.Ca)
+	if err != nil {
+		return nil, errors.Wrap(err, "cannot parse ca file")
+	}
+
+	cert, err := tls.LoadX509KeyPair(in.Cert, in.Key)
+	if err != nil {
+		return nil, errors.Wrap(err, "cannot parse cert/key pair")
+	}
+
+	tlsConfig, err := tlsconf.ClientAuthClient(in.ServerCN, ca, cert)
+	if err != nil {
+		return nil, errors.Wrap(err, "cannot build tls config")
+	}
+
+	return &TLSConnecter{in.Address, dialer, tlsConfig}, nil
+}
+
+func (c *TLSConnecter) Connect(dialCtx context.Context) (conn net.Conn, err error) {
+	return tls.DialWithDialer(&c.dialer, "tcp", c.Address, c.tlsConfig)
+}

daemon/transport/connecter/connecter.go

@@ -0,0 +1,54 @@
+package connecter
+
+import (
+	"fmt"
+	"github.com/problame/go-streamrpc"
+	"github.com/zrepl/zrepl/config"
+	"github.com/zrepl/zrepl/daemon/streamrpcconfig"
+)
+
+func FromConfig(g *config.Global, in config.ConnectEnum) (*ClientFactory, error) {
+	var (
+		connecter            streamrpc.Connecter
+		errConnecter, errRPC error
+		connConf             *streamrpc.ConnConfig
+	)
+	switch v := in.Ret.(type) {
+	case *config.SSHStdinserverConnect:
+		connecter, errConnecter = SSHStdinserverConnecterFromConfig(v)
+		connConf, errRPC = streamrpcconfig.FromDaemonConfig(g, v.RPC)
+	case *config.TCPConnect:
+		connecter, errConnecter = TCPConnecterFromConfig(v)
+		connConf, errRPC = streamrpcconfig.FromDaemonConfig(g, v.RPC)
+	case *config.TLSConnect:
+		connecter, errConnecter = TLSConnecterFromConfig(v)
+		connConf, errRPC = streamrpcconfig.FromDaemonConfig(g, v.RPC)
+	case *config.LocalConnect:
+		connecter, errConnecter = LocalConnecterFromConfig(v)
+		connConf, errRPC = streamrpcconfig.FromDaemonConfig(g, v.RPC)
+	default:
+		panic(fmt.Sprintf("implementation error: unknown connecter type %T", v))
+	}
+
+	if errConnecter != nil {
+		return nil, errConnecter
+	}
+	if errRPC != nil {
+		return nil, errRPC
+	}
+
+	config := streamrpc.ClientConfig{ConnConfig: connConf}
+	if err := config.Validate(); err != nil {
+		return nil, err
+	}
+	return &ClientFactory{connecter: connecter, config: &config}, nil
+}
+
+type ClientFactory struct {
+	connecter streamrpc.Connecter
+	config    *streamrpc.ClientConfig
+}
+
+func (f ClientFactory) NewClient() (*streamrpc.Client, error) {
+	return streamrpc.NewClient(f.connecter, f.config)
+}