[WIP] Feature: TCPFactory/TCPTransport instead of dmsg

cmd/skywire-cli/commands/node/gen-config.go

@@ -100,10 +100,12 @@ func defaultConfig() *visor.Config {
 	conf.Routing.RouteFinder = "https://routefinder.skywire.skycoin.net/"
 
 	const defaultSetupNodePK = "0324579f003e6b4048bae2def4365e634d8e0e3054a20fc7af49daf2a179658557"
+
 	sPK := cipher.PubKey{}
 	if err := sPK.UnmarshalText([]byte(defaultSetupNodePK)); err != nil {
 		log.WithError(err).Warnf("Failed to unmarshal default setup node public key %s", defaultSetupNodePK)
 	}
+
 	conf.Routing.SetupNodes = []cipher.PubKey{sPK}
 	conf.Routing.Table.Type = "boltdb"
 	conf.Routing.Table.Location = "./skywire/routing.db"
@@ -120,5 +122,8 @@ func defaultConfig() *visor.Config {
 
 	conf.Interfaces.RPCAddress = "localhost:3435"
 
+	conf.TransportType = "dmsg"
+	conf.PubKeysFile = "./local/pubkeys"
+
 	return conf
 }

go.mod

@@ -20,8 +20,10 @@ require (
 	github.com/spf13/cobra v0.0.5
 	github.com/stretchr/testify v1.3.0
 	go.etcd.io/bbolt v1.3.3
-	golang.org/x/crypto v0.0.0-20190621222207-cc06ce4a13d4
-	golang.org/x/net v0.0.0-20190620200207-3b0461eec859
+	golang.org/x/crypto v0.0.0-20190701094942-4def268fd1a4
+	golang.org/x/net v0.0.0-20190724013045-ca1201d0de80
+	golang.org/x/sys v0.0.0-20190712062909-fae7ac547cb7 // indirect
+	golang.org/x/tools v0.0.0-20190725161231-2e34cfcb95cb // indirect
 )
 
 // Uncomment for tests with alternate branches of 'dmsg'

go.sum

@@ -119,11 +119,15 @@ golang.org/x/crypto v0.0.0-20181203042331-505ab145d0a9/go.mod h1:6SG95UA2DQfeDnf
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20190621222207-cc06ce4a13d4 h1:ydJNl0ENAG67pFbB+9tfhiL2pYqLhfoaZFw/cjLhY4A=
 golang.org/x/crypto v0.0.0-20190621222207-cc06ce4a13d4/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
+golang.org/x/crypto v0.0.0-20190701094942-4def268fd1a4 h1:HuIa8hRrWRSrqYzx1qI49NNxhdi2PrY7gxVSq1JjLDc=
+golang.org/x/crypto v0.0.0-20190701094942-4def268fd1a4/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/net v0.0.0-20181114220301-adae6a3d119a/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20190620200207-3b0461eec859 h1:R/3boaszxrf1GEUWTVDzSKVwLmSJpwZ1yqXm8j0v2QI=
 golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
+golang.org/x/net v0.0.0-20190724013045-ca1201d0de80 h1:Ao/3l156eZf2AW5wK8a7/smtodRU+gha3+BeqJ69lRk=
+golang.org/x/net v0.0.0-20190724013045-ca1201d0de80/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
@@ -136,12 +140,16 @@ golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20190422165155-953cdadca894/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190626221950-04f50cda93cb h1:fgwFCsaw9buMuxNd6+DQfAuSFqbNiQZpcgJQAgJsK6k=
 golang.org/x/sys v0.0.0-20190626221950-04f50cda93cb/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20190712062909-fae7ac547cb7 h1:LepdCS8Gf/MVejFIt8lsiexZATdoGVyp5bcyS+rYoUI=
+golang.org/x/sys v0.0.0-20190712062909-fae7ac547cb7/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/text v0.3.0 h1:g61tztE5qeGQ89tm6NTjjM9VPIm088od1l6aSorWRWg=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.2 h1:tW2bmiBqwgJj/UpqtC8EpXEZVYOwU0yG4iWbprSVAcs=
 golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20190627182818-9947fec5c3ab/go.mod h1:/rFqwRUd4F7ZHNgwSSTFct+R/Kf4OFW1sUzUTQQTgfc=
+golang.org/x/tools v0.0.0-20190725161231-2e34cfcb95cb h1:Zi4or4sGkVpI7V5TX4JDMHJAthfKDKg8WotOBKXqmTs=
+golang.org/x/tools v0.0.0-20190725161231-2e34cfcb95cb/go.mod h1:jcCCGcm9btYwXyDqrUWc6MKQKKGJCWEQ3AfLSRIbEuI=
 gopkg.in/alecthomas/kingpin.v2 v2.2.6 h1:jMFz6MfLP0/4fUyZle81rXUoxOBFi19VUFKVDOQfozc=
 gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=

pkg/setup/config.go

@@ -16,5 +16,9 @@ type Config struct {
 
 	TransportDiscovery string `json:"transport_discovery"`
 
+	TransportType    string `json:"transport_type"`
+	PubKeysFile      string `json:"pubkeys_file"`
+	TCPTransportAddr string `json:"tcptransport_addr"`
+
 	LogLevel string `json:"log_level"`
 }

pkg/setup/node.go

@@ -30,7 +30,7 @@ type Node struct {
 	Logger *logging.Logger
 
 	tm        *transport.Manager
-	messenger *dmsg.Client
+	messenger transport.Factory
 
 	srvCount int
 	metrics  metrics.Recorder
@@ -45,7 +45,21 @@ func NewNode(conf *Config, metrics metrics.Recorder) (*Node, error) {
 	if lvl, err := logging.LevelFromString(conf.LogLevel); err == nil {
 		logger.SetLevel(lvl)
 	}
-	messenger := dmsg.NewClient(pk, sk, disc.NewHTTP(conf.Messaging.Discovery), dmsg.SetLogger(logger.PackageLogger(dmsg.Type)))
+
+	var (
+		factory transport.Factory
+		err     error
+	)
+	switch conf.TransportType {
+	case "dmsg":
+		factory = dmsg.NewClient(pk, sk, disc.NewHTTP(conf.Messaging.Discovery),
+			dmsg.SetLogger(logger.PackageLogger(dmsg.Type)))
+	case "tcp-transport":
+		factory, err = transport.NewTCPFactory(pk, conf.PubKeysFile, conf.TCPTransportAddr)
+		if err != nil {
+			return nil, err
+		}
+	}
 
 	trDiscovery, err := trClient.NewHTTP(conf.TransportDiscovery, pk, sk)
 	if err != nil {
@@ -59,7 +73,7 @@ func NewNode(conf *Config, metrics metrics.Recorder) (*Node, error) {
 		LogStore:        transport.InMemoryTransportLogStore(),
 	}
 
-	tm, err := transport.NewManager(tmConf, messenger)
+	tm, err := transport.NewManager(tmConf, factory)
 	if err != nil {
 		log.Fatal("Failed to setup Transport Manager: ", err)
 	}
@@ -69,18 +83,22 @@ func NewNode(conf *Config, metrics metrics.Recorder) (*Node, error) {
 		Logger:    logger.PackageLogger("routesetup"),
 		metrics:   metrics,
 		tm:        tm,
-		messenger: messenger,
+		messenger: factory,
 		srvCount:  conf.Messaging.ServerCount,
 	}, nil
 }
 
 // Serve starts transport listening loop.
 func (sn *Node) Serve(ctx context.Context) error {
-	if sn.srvCount > 0 {
-		if err := sn.messenger.InitiateServerConnections(ctx, sn.srvCount); err != nil {
-			return fmt.Errorf("messaging: %s", err)
+
+	switch factory := sn.messenger.(type) {
+	case *dmsg.Client:
+		if sn.srvCount > 0 {
+			if err := factory.InitiateServerConnections(ctx, sn.srvCount); err != nil {
+				return fmt.Errorf("messaging: %s", err)
+			}
+			sn.Logger.Info("Connected to messaging servers")
 		}
-		sn.Logger.Info("Connected to messaging servers")
 	}
 
 	go func() {

pkg/transport/tcp_transport.go

@@ -4,6 +4,7 @@ import (
 	"bufio"
 	"context"
 	"errors"
+	"fmt"
 	"net"
 	"os"
 	"path/filepath"
@@ -18,35 +19,51 @@ var ErrUnknownRemote = errors.New("unknown remote")
 
 // TCPFactory implements Factory over TCP connection.
 type TCPFactory struct {
-	l   *net.TCPListener
-	lpk cipher.PubKey
-	pkt PubKeyTable
+	Pk      cipher.PubKey
+	PkTable PubKeyTable
+	Lsr     *net.TCPListener
 }
 
-// NewTCPFactory constructs a new TCP Factory.
-func NewTCPFactory(lpk cipher.PubKey, pkt PubKeyTable, l *net.TCPListener) Factory {
-	return &TCPFactory{l, lpk, pkt}
+// NewTCPFactory constructs a new TCP Factory
+func NewTCPFactory(pk cipher.PubKey, pubkeysFile string, tcpAddr string) (Factory, error) {
+
+	pkTbl, err := FilePubKeyTable(pubkeysFile)
+	if err != nil {
+		return nil, fmt.Errorf("error %v reading %v", err, pubkeysFile)
+	}
+
+	addr, err := net.ResolveTCPAddr("tcp", tcpAddr)
+	if err != nil {
+		return nil, fmt.Errorf("error %v resolving %v", err, tcpAddr)
+	}
+
+	tcpListener, err := net.ListenTCP("tcp", addr)
+	if err != nil {
+		return nil, fmt.Errorf("error %v listening %v", err, tcpAddr)
+	}
+
+	return &TCPFactory{pk, pkTbl, tcpListener}, nil
 }
 
 // Accept accepts a remotely-initiated Transport.
 func (f *TCPFactory) Accept(ctx context.Context) (Transport, error) {
-	conn, err := f.l.AcceptTCP()
+	conn, err := f.Lsr.AcceptTCP()
 	if err != nil {
 		return nil, err
 	}
 
 	raddr := conn.RemoteAddr().(*net.TCPAddr)
-	rpk := f.pkt.RemotePK(raddr.IP)
+	rpk := f.PkTable.RemotePK(raddr.IP)
 	if rpk.Null() {
 		return nil, ErrUnknownRemote
 	}
 
-	return &TCPTransport{conn, [2]cipher.PubKey{f.lpk, rpk}}, nil
+	return &TCPTransport{conn, [2]cipher.PubKey{f.Pk, rpk}}, nil
 }
 
 // Dial initiates a Transport with a remote node.
 func (f *TCPFactory) Dial(ctx context.Context, remote cipher.PubKey) (Transport, error) {
-	raddr := f.pkt.RemoteAddr(remote)
+	raddr := f.PkTable.RemoteAddr(remote)
 	if raddr == nil {
 		return nil, ErrUnknownRemote
 	}
@@ -56,20 +73,20 @@ func (f *TCPFactory) Dial(ctx context.Context, remote cipher.PubKey) (Transport,
 		return nil, err
 	}
 
-	return &TCPTransport{conn, [2]cipher.PubKey{f.lpk, remote}}, nil
+	return &TCPTransport{conn, [2]cipher.PubKey{f.Pk, remote}}, nil
 }
 
 // Close implements io.Closer
 func (f *TCPFactory) Close() error {
 	if f == nil {
 		return nil
 	}
-	return f.l.Close()
+	return f.Lsr.Close()
 }
 
 // Local returns the local public key.
 func (f *TCPFactory) Local() cipher.PubKey {
-	return f.lpk
+	return f.Pk
 }
 
 // Type returns the Transport type.

pkg/transport/tcp_transport_test.go

@@ -32,7 +32,7 @@ func TestTCPFactory(t *testing.T) {
 	pkt1 := transport.InMemoryPubKeyTable(map[cipher.PubKey]*net.TCPAddr{pk2: addr2})
 	pkt2 := transport.InMemoryPubKeyTable(map[cipher.PubKey]*net.TCPAddr{pk1: addr1})
 
-	f1 := transport.NewTCPFactory(pk1, pkt1, l1)
+	f1 := &transport.TCPFactory{pk1, pkt1, l1}
 	errCh := make(chan error)
 	go func() {
 		tr, err := f1.Accept(context.TODO())
@@ -49,7 +49,7 @@ func TestTCPFactory(t *testing.T) {
 		errCh <- nil
 	}()
 
-	f2 := transport.NewTCPFactory(pk2, pkt2, l2)
+	f2 := &transport.TCPFactory{pk2, pkt2, l2}
 	assert.Equal(t, "tcp", f2.Type())
 	assert.Equal(t, pk2, f2.Local())
 

pkg/visor/config.go

@@ -57,10 +57,13 @@ type Config struct {
 	AppsPath  string `json:"apps_path"`
 	LocalPath string `json:"local_path"`
 
-	LogLevel        string   `json:"log_level"`
-	ShutdownTimeout Duration `json:"shutdown_timeout"` // time value, examples: 10s, 1m, etc
+	TransportType    string `json:"transport_type"`
+	PubKeysFile      string `json:"pubkeys_file"`
+	TCPTransportAddr string `json:"tcptransport_addr"`
 
-	Interfaces InterfaceConfig `json:"interfaces"`
+	ShutdownTimeout Duration        `json:"shutdown_timeout"` // time value, examples: 10s, 1m, etc
+	Interfaces      InterfaceConfig `json:"interfaces"`
+	LogLevel        string          `json:"log_level"`
 }
 
 // MessagingConfig returns config for dmsg client.

pkg/visor/visor.go

@@ -86,7 +86,7 @@ type PacketRouter interface {
 type Node struct {
 	config    *Config
 	router    PacketRouter
-	messenger *dmsg.Client
+	messenger transport.Factory
 	tm        *transport.Manager
 	rt        routing.Table
 	executer  appExecuter
@@ -122,16 +122,27 @@ func NewNode(config *Config, masterLogger *logging.MasterLogger) (*Node, error)
 	sk := config.Node.StaticSecKey
 	mConfig, err := config.MessagingConfig()
 	if err != nil {
-		return nil, fmt.Errorf("invalid Messaging config: %s", err)
+		return nil, fmt.Errorf("invalid messaging config: %s", err)
 	}
 
-	node.messenger = dmsg.NewClient(mConfig.PubKey, mConfig.SecKey, mConfig.Discovery, dmsg.SetLogger(node.Logger.PackageLogger(dmsg.Type)))
+	switch config.TransportType {
+	case "dmsg":
+		node.messenger = dmsg.NewClient(mConfig.PubKey, mConfig.SecKey,
+			mConfig.Discovery, dmsg.SetLogger(node.Logger.PackageLogger(dmsg.Type)))
+	case "tcp-transport":
+		var err error
+		node.messenger, err = transport.NewTCPFactory(config.Node.StaticPubKey, config.PubKeysFile, config.TCPTransportAddr)
+		if err != nil {
+			return nil, err
+		}
+	}
 
 	trDiscovery, err := config.TransportDiscovery()
 	if err != nil {
 		return nil, fmt.Errorf("invalid MessagingConfig: %s", err)
 	}
 	logStore, err := config.TransportLogStore()
+
 	if err != nil {
 		return nil, fmt.Errorf("invalid TransportLogStore: %s", err)
 	}
@@ -205,11 +216,17 @@ func NewNode(config *Config, masterLogger *logging.MasterLogger) (*Node, error)
 // Start spawns auto-started Apps, starts router and RPC interfaces .
 func (node *Node) Start() error {
 	ctx := context.Background()
-	err := node.messenger.InitiateServerConnections(ctx, node.config.Messaging.ServerCount)
-	if err != nil {
-		return fmt.Errorf("%s: %s", dmsg.Type, err)
+
+	switch factory := node.messenger.(type) {
+	case *dmsg.Client:
+		err := factory.InitiateServerConnections(ctx, node.config.Messaging.ServerCount)
+		if err != nil {
+			return fmt.Errorf("%s: %s", dmsg.Type, err)
+		}
+		node.logger.Info("Connected to messaging servers")
+	case *transport.TCPFactory:
+		node.logger.Info("TCPFactory: ignoring configured messaging servers")
 	}
-	node.logger.Info("Connected to messaging servers")
 
 	pathutil.EnsureDir(node.dir())
 	node.closePreviousApps()