Renamed dms to dmsg as per Brandon's request.

cmd/manager-node/commands/gen-config.go

@@ -4,11 +4,10 @@ import (
 	"fmt"
 	"path/filepath"
 
-	"github.com/skycoin/skywire/pkg/util/pathutil"
-
 	"github.com/spf13/cobra"
 
 	"github.com/skycoin/skywire/pkg/manager"
+	"github.com/skycoin/skywire/pkg/util/pathutil"
 )
 
 var (

cmd/messaging-server/commands/root.go

@@ -15,7 +15,7 @@ import (
 	"github.com/spf13/cobra"
 
 	"github.com/skycoin/skywire/pkg/cipher"
-	"github.com/skycoin/skywire/pkg/dms"
+	"github.com/skycoin/skywire/pkg/dmsg"
 	"github.com/skycoin/skywire/pkg/messaging-discovery/client"
 )
 
@@ -72,7 +72,7 @@ var rootCmd = &cobra.Command{
 		}()
 
 		// Start
-		srv := dms.NewServer(conf.PubKey, conf.SecKey, conf.PublicAddress, client.NewHTTP(conf.Discovery))
+		srv := dmsg.NewServer(conf.PubKey, conf.SecKey, conf.PublicAddress, client.NewHTTP(conf.Discovery))
 		log.Fatal(srv.ListenAndServe(conf.LocalAddress))
 	},
 }

cmd/skywire-cli/commands/node/transports.go

@@ -7,6 +7,8 @@ import (
 	"text/tabwriter"
 	"time"
 
+	"github.com/skycoin/skywire/pkg/dmsg"
+
 	"github.com/spf13/cobra"
 
 	"github.com/skycoin/skywire/cmd/skywire-cli/internal"
@@ -77,7 +79,7 @@ var (
 )
 
 func init() {
-	addTpCmd.Flags().StringVar(&transportType, "type", "dms", "type of transport to add")
+	addTpCmd.Flags().StringVar(&transportType, "type", dmsg.Type, "type of transport to add")
 	addTpCmd.Flags().BoolVar(&public, "public", true, "whether to make the transport public")
 	addTpCmd.Flags().DurationVarP(&timeout, "timeout", "t", 0, "if specified, sets an operation timeout")
 }

go.mod

@@ -19,7 +19,6 @@ require (
 	github.com/mgutz/ansi v0.0.0-20170206155736-9520e82c474b // indirect
 	github.com/mitchellh/go-homedir v1.1.0
 	github.com/prometheus/client_golang v0.9.2
-	github.com/prometheus/common v0.0.0-20181126121408-4724e9255275
 	github.com/sirupsen/logrus v1.4.1
 	github.com/skycoin/skycoin v0.25.1
 	github.com/spf13/cobra v0.0.3

pkg/dms/client.go → pkg/dmsg/client.go

@@ -1,4 +1,4 @@
-package dms
+package dmsg
 
 import (
 	"context"
@@ -9,7 +9,6 @@ import (
 	"sync"
 	"time"
 
-	"github.com/prometheus/common/log"
 	"github.com/skycoin/skycoin/src/util/logging"
 
 	"github.com/skycoin/skywire/internal/noise"
@@ -25,13 +24,13 @@ var (
 	ErrClientClosed = errors.New("client closed")
 )
 
-// ClientConn represents a connection between a dms.Client and dms.Server from a client's perspective.
+// ClientConn represents a connection between a dmsg.Client and dmsg.Server from a client's perspective.
 type ClientConn struct {
 	log *logging.Logger
 
-	net.Conn                // conn to dms server
+	net.Conn                // conn to dmsg server
 	local     cipher.PubKey // local client's pk
-	remoteSrv cipher.PubKey // dms server's public key
+	remoteSrv cipher.PubKey // dmsg server's public key
 
 	// nextInitID keeps track of unused tp_ids to assign a future locally-initiated tp.
 	// locally-initiated tps use an even tp_id between local and intermediary dms_server.
@@ -83,7 +82,7 @@ func (c *ClientConn) addTp(ctx context.Context, clientPK cipher.PubKey) (*Transp
 
 	id := c.nextInitID
 	c.nextInitID = id + 2
-	ch := NewTransport(c.Conn, c.local, clientPK, id)
+	ch := NewTransport(c.Conn, c.log, c.local, clientPK, id)
 	c.tps[id] = ch
 	return ch, nil
 }
@@ -110,7 +109,7 @@ func (c *ClientConn) handleRequestFrame(ctx context.Context, accept chan<- *Tran
 		return initPK, ErrRequestCheckFailed
 	}
 
-	tp := NewTransport(c.Conn, c.local, initPK, id)
+	tp := NewTransport(c.Conn, c.log, c.local, initPK, id)
 	if err := tp.Handshake(ctx); err != nil {
 		// return err here as response handshake is send via ClientConn and that shouldn't fail.
 		c.Close()
@@ -274,16 +273,16 @@ func (c *Client) connCount() int {
 }
 
 // InitiateServerConnections initiates connections with dms_servers.
-func (c *Client) InitiateServerConnections(ctx context.Context, n int) error {
-	if n == 0 {
+func (c *Client) InitiateServerConnections(ctx context.Context, min int) error {
+	if min == 0 {
 		return nil
 	}
 	entries, err := c.findServerEntries(ctx)
 	if err != nil {
 		return err
 	}
 	c.log.Info("found dms_server entries:", entries)
-	if err := c.connectToServers(ctx, entries, n); err != nil {
+	if err := c.findOrConnectToServers(ctx, entries, min); err != nil {
 		return err
 	}
 	if err := c.updateDiscEntry(ctx); err != nil {
@@ -300,6 +299,7 @@ func (c *Client) findServerEntries(ctx context.Context) ([]*client.Entry, error)
 			case <-ctx.Done():
 				return nil, fmt.Errorf("dms_servers are not available: %s", err)
 			default:
+				c.log.WithError(err).Warnf("no dms_servers found: trying again is 1 second...")
 				time.Sleep(time.Second)
 				continue
 			}
@@ -308,25 +308,19 @@ func (c *Client) findServerEntries(ctx context.Context) ([]*client.Entry, error)
 	}
 }
 
-func (c *Client) connectToServers(ctx context.Context, entries []*client.Entry, min int) error {
-	//c.mx.Lock()
-	//defer c.mx.Unlock()
-
+func (c *Client) findOrConnectToServers(ctx context.Context, entries []*client.Entry, min int) error {
 	for _, entry := range entries {
-		if _, ok := c.getConn(entry.Static); ok {
-			continue
-		}
 		_, err := c.findOrConnectToServer(ctx, entry.Static)
 		if err != nil {
-			c.log.Warnf("Failed to connect to server %s: %s", entry.Static, err)
+			c.log.Warnf("findOrConnectToServers: failed to find/connect to server %s: %s", entry.Static, err)
 			continue
 		}
-		c.log.Infof("Connected to server %s", entry.Static)
+		c.log.Infof("findOrConnectToServers: found/connected to server %s", entry.Static)
 		if c.connCount() >= min {
 			return nil
 		}
 	}
-	return fmt.Errorf("servers are not available: all servers failed")
+	return fmt.Errorf("findOrConnectToServers: all servers failed")
 }
 
 func (c *Client) findOrConnectToServer(ctx context.Context, srvPK cipher.PubKey) (*ClientConn, error) {
@@ -383,7 +377,7 @@ func (c *Client) findOrConnectToServer(ctx context.Context, srvPK cipher.PubKey)
 }
 
 func (c *Client) updateDiscEntry(ctx context.Context) error {
-	log.Info("updatingEntry")
+	c.log.Info("updatingEntry")
 	var srvPKs []cipher.PubKey
 	c.mx.RLock()
 	for pk := range c.conns {

pkg/dms/frame.go → pkg/dmsg/frame.go

@@ -1,4 +1,4 @@
-package dms
+package dmsg
 
 import (
 	"encoding/binary"
@@ -11,7 +11,7 @@ import (
 
 const (
 	// Type returns the transport type string.
-	Type = "dms"
+	Type = "dmsg"
 
 	hsTimeout    = time.Second * 10
 	readTimeout  = time.Second * 10
@@ -58,7 +58,7 @@ const (
 	AckType     = FrameType(11)
 )
 
-// Frame is the dms data unit.
+// Frame is the dmsg data unit.
 type Frame []byte
 
 // MakeFrame creates a new Frame.

pkg/dmsg/frame_test.go

@@ -0,0 +1 @@
+package dmsg

pkg/dms/server.go → pkg/dmsg/server.go

@@ -1,4 +1,4 @@
-package dms
+package dmsg
 
 import (
 	"context"
@@ -30,15 +30,15 @@ func (r *NextConn) writeFrame(ft FrameType, p []byte) error {
 	return nil
 }
 
-// ServerConn is a connection between a dms.Server and a dms.Client from a server's perspective.
+// ServerConn is a connection between a dmsg.Server and a dmsg.Client from a server's perspective.
 type ServerConn struct {
 	log *logging.Logger
 
 	net.Conn
 	remoteClient cipher.PubKey
 
 	nextRespID uint16
-	nextLinks  [math.MaxUint16]*NextConn
+	nextConns  [math.MaxUint16]*NextConn
 	mx         sync.RWMutex
 }
 
@@ -49,19 +49,19 @@ func NewServerConn(log *logging.Logger, conn net.Conn, remoteClient cipher.PubKe
 
 func (c *ServerConn) delNext(id uint16) {
 	c.mx.Lock()
-	c.nextLinks[id] = nil
+	c.nextConns[id] = nil
 	c.mx.Unlock()
 }
 
 func (c *ServerConn) setNext(id uint16, r *NextConn) {
 	c.mx.Lock()
-	c.nextLinks[id] = r
+	c.nextConns[id] = r
 	c.mx.Unlock()
 }
 
 func (c *ServerConn) getNext(id uint16) (*NextConn, bool) {
 	c.mx.RLock()
-	r := c.nextLinks[id]
+	r := c.nextConns[id]
 	c.mx.RUnlock()
 	return r, r != nil
 }
@@ -71,7 +71,7 @@ func (c *ServerConn) addNext(ctx context.Context, r *NextConn) (uint16, error) {
 	defer c.mx.Unlock()
 
 	for {
-		if r := c.nextLinks[c.nextRespID]; r == nil {
+		if r := c.nextConns[c.nextRespID]; r == nil {
 			break
 		}
 		c.nextRespID += 2
@@ -85,7 +85,7 @@ func (c *ServerConn) addNext(ctx context.Context, r *NextConn) (uint16, error) {
 
 	id := c.nextRespID
 	c.nextRespID = id + 2
-	c.nextLinks[id] = r
+	c.nextConns[id] = r
 	return id, nil
 }
 

pkg/dms/server_test.go → pkg/dmsg/server_test.go

@@ -1,8 +1,9 @@
-package dms
+package dmsg
 
 import (
 	"context"
 	"fmt"
+	"golang.org/x/net/nettest"
 	"sync"
 	"testing"
 	"time"
@@ -51,6 +52,7 @@ func TestNewClient(t *testing.T) {
 	a := NewClient(aPK, aSK, dc)
 	a.SetLogger(logging.MustGetLogger("A"))
 	require.NoError(t, a.InitiateServerConnections(context.Background(), 1))
+	nettest.NewLocalListener()
 
 	b := NewClient(bPK, bSK, dc)
 	b.SetLogger(logging.MustGetLogger("B"))

pkg/dms/transport.go → pkg/dmsg/transport.go

@@ -1,4 +1,4 @@
-package dms
+package dmsg
 
 import (
 	"bytes"
@@ -10,6 +10,8 @@ import (
 	"net"
 	"sync"
 
+	"github.com/skycoin/skycoin/src/util/logging"
+
 	"github.com/skycoin/skywire/internal/ioutil"
 	"github.com/skycoin/skywire/pkg/cipher"
 	"github.com/skycoin/skywire/pkg/transport"
@@ -21,10 +23,11 @@ var (
 	ErrRequestCheckFailed = errors.New("request check failed")
 )
 
-// Transport represents a connection from dms.Client to remote dms.Client (via dms.Server intermediary).
+// Transport represents a connection from dmsg.Client to remote dmsg.Client (via dmsg.Server intermediary).
 // It implements transport.Transport
 type Transport struct {
 	net.Conn // link with server.
+	log      *logging.Logger
 
 	id     uint16
 	local  cipher.PubKey
@@ -39,9 +42,10 @@ type Transport struct {
 }
 
 // NewTransport creates a new dms_tp.
-func NewTransport(conn net.Conn, local, remote cipher.PubKey, id uint16) *Transport {
+func NewTransport(conn net.Conn, log *logging.Logger, local, remote cipher.PubKey, id uint16) *Transport {
 	return &Transport{
 		Conn:   conn,
+		log:    log,
 		id:     id,
 		local:  local,
 		remote: remote,
@@ -79,7 +83,6 @@ func (c *Transport) awaitResponse(ctx context.Context) error {
 func (c *Transport) Handshake(ctx context.Context) error {
 	// if channel ID is even, client is initiator.
 	if isInitiatorID(c.id) {
-
 		pks := combinePKs(c.local, c.remote)
 		f := MakeFrame(RequestType, c.id, pks)
 		if err := writeFrame(c.Conn, f); err != nil {
@@ -91,11 +94,14 @@ func (c *Transport) Handshake(ctx context.Context) error {
 			return err
 		}
 	} else {
+		c.log.Infof("tp_hs responding...")
 		f := MakeFrame(AcceptType, c.id, combinePKs(c.remote, c.local))
 		if err := writeFrame(c.Conn, f); err != nil {
+			c.log.WithError(err).Error("tp_hs responded with error.")
 			c.close()
 			return err
 		}
+		c.log.Infoln("tp_hs responded:", f)
 	}
 	return nil
 }

pkg/dmsg/transport_test.go

@@ -0,0 +1 @@
+package dmsg

pkg/node/node.go

@@ -15,7 +15,7 @@ import (
 	"syscall"
 	"time"
 
-	"github.com/skycoin/skywire/pkg/dms"
+	"github.com/skycoin/skywire/pkg/dmsg"
 
 	"github.com/skycoin/skycoin/src/util/logging"
 
@@ -80,7 +80,7 @@ type PacketRouter interface {
 type Node struct {
 	config    *Config
 	router    PacketRouter
-	messenger *dms.Client
+	messenger *dmsg.Client
 	tm        *transport.Manager
 	rt        routing.Table
 	executer  appExecuter
@@ -117,8 +117,8 @@ func NewNode(config *Config) (*Node, error) {
 		return nil, fmt.Errorf("invalid Messaging config: %s", err)
 	}
 
-	node.messenger = dms.NewClient(mConfig.PubKey, mConfig.SecKey, mConfig.Discovery)
-	node.messenger.SetLogger(node.Logger.PackageLogger("dms"))
+	node.messenger = dmsg.NewClient(mConfig.PubKey, mConfig.SecKey, mConfig.Discovery)
+	node.messenger.SetLogger(node.Logger.PackageLogger(dmsg.Type))
 
 	trDiscovery, err := config.TransportDiscovery()
 	if err != nil {
@@ -200,7 +200,7 @@ func (node *Node) Start() error {
 	ctx := context.Background()
 	err := node.messenger.InitiateServerConnections(ctx, node.config.Messaging.ServerCount)
 	if err != nil {
-		return fmt.Errorf("dms: %s", err)
+		return fmt.Errorf("%s: %s", dmsg.Type, err)
 	}
 	node.logger.Info("Connected to messaging servers")