Merge branch 'feature/dmsg' of https://github.com/evanlinjin/skywire …

…into feature/dmsg-tests

go.mod

@@ -3,33 +3,23 @@ module github.com/skycoin/skywire
 go 1.12
 
 require (
-	github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc // indirect
-	github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf // indirect
 	github.com/armon/go-socks5 v0.0.0-20160902184237-e75332964ef5
 	github.com/flynn/noise v0.0.0-20180327030543-2492fe189ae6
 	github.com/go-chi/chi v4.0.2+incompatible
 	github.com/google/uuid v1.1.1
 	github.com/gorilla/handlers v1.4.0
 	github.com/gorilla/securecookie v1.1.1
 	github.com/hashicorp/yamux v0.0.0-20181012175058-2f1d1f20f75d
-	github.com/inconshreveable/mousetrap v1.0.0 // indirect
-	github.com/kr/pretty v0.1.0 // indirect
-	github.com/kr/pty v1.1.4
-	github.com/mattn/go-colorable v0.1.1 // indirect
+	github.com/kr/pty v1.1.5
+	github.com/mattn/go-colorable v0.1.2 // indirect
 	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/sirupsen/logrus v1.4.1
-	github.com/skycoin/skycoin v0.25.1
-	github.com/spf13/cobra v0.0.3
-	github.com/spf13/pflag v1.0.3 // indirect
+	github.com/prometheus/client_golang v1.0.0
+	github.com/sirupsen/logrus v1.4.2
+	github.com/skycoin/skycoin v0.26.0
+	github.com/spf13/cobra v0.0.5
 	github.com/stretchr/testify v1.3.0
-	go.etcd.io/bbolt v1.3.2
+	go.etcd.io/bbolt v1.3.3
 	golang.org/x/crypto v0.0.0-20190618222545-ea8f1a30c443
 	golang.org/x/net v0.0.0-20190619014844-b5b0513f8c1b
-	golang.org/x/sys v0.0.0-20190618155005-516e3c20635f // indirect
-	golang.org/x/text v0.3.2 // indirect
-	golang.org/x/tools v0.0.0-20190618233249-04b924abaa25 // indirect
-	gopkg.in/alecthomas/kingpin.v2 v2.2.6 // indirect
-	gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 // indirect
 )

pkg/dmsg/client.go

@@ -200,7 +200,7 @@ func (c *ClientConn) Serve(ctx context.Context, accept chan<- *Transport) (err e
 		// delete tp on any failure.
 
 		if tp, ok := c.getTp(id); ok {
-			if err := tp.Inject(f); err != nil {
+			if err := tp.HandleFrame(f); err != nil {
 				log.WithError(err).Warnf("Rejected [%s]: Transport closed.", ft)
 			}
 			continue
@@ -272,6 +272,20 @@ func (c *ClientConn) Close() error {
 	return nil
 }
 
+// ClientOption represents an optional argument for Client.
+type ClientOption func(c *Client) error
+
+// SetLogger sets the internal logger for Client.
+func SetLogger(log *logging.Logger) ClientOption {
+	return func(c *Client) error {
+		if log == nil {
+			return errors.New("nil logger set")
+		}
+		c.log = log
+		return nil
+	}
+}
+
 // Client implements transport.Factory
 type Client struct {
 	log *logging.Logger
@@ -289,8 +303,8 @@ type Client struct {
 }
 
 // NewClient creates a new Client.
-func NewClient(pk cipher.PubKey, sk cipher.SecKey, dc client.APIClient) *Client {
-	return &Client{
+func NewClient(pk cipher.PubKey, sk cipher.SecKey, dc client.APIClient, opts ...ClientOption) *Client {
+	c := &Client{
 		log:    logging.MustGetLogger("dmsg_client"),
 		pk:     pk,
 		sk:     sk,
@@ -299,11 +313,12 @@ func NewClient(pk cipher.PubKey, sk cipher.SecKey, dc client.APIClient) *Client
 		accept: make(chan *Transport, AcceptBufferSize),
 		done:   make(chan struct{}),
 	}
-}
-
-// SetLogger sets the dms_client's logger.
-func (c *Client) SetLogger(log *logging.Logger) {
-	c.log = log
+	for _, opt := range opts {
+		if err := opt(c); err != nil {
+			panic(err)
+		}
+	}
+	return c
 }
 
 func (c *Client) updateDiscEntry(ctx context.Context) error {

pkg/dmsg/server_test.go

@@ -189,13 +189,11 @@ func TestServer_Serve(t *testing.T) {
 		aPK, aSK := cipher.GenerateKeyPair()
 		bPK, bSK := cipher.GenerateKeyPair()
 
-		a := NewClient(aPK, aSK, dc)
-		a.SetLogger(logging.MustGetLogger("A"))
+		a := NewClient(aPK, aSK, dc, SetLogger(logging.MustGetLogger("A")))
 		err := a.InitiateServerConnections(context.Background(), 1)
 		require.NoError(t, err)
 
-		b := NewClient(bPK, bSK, dc)
-		b.SetLogger(logging.MustGetLogger("B"))
+		b := NewClient(bPK, bSK, dc, SetLogger(logging.MustGetLogger("B")))
 		err = b.InitiateServerConnections(context.Background(), 1)
 		require.NoError(t, err)
 
@@ -315,8 +313,7 @@ func TestServer_Serve(t *testing.T) {
 		for i := 0; i < initiatorsCount; i++ {
 			pk, sk := cipher.GenerateKeyPair()
 
-			c := NewClient(pk, sk, dc)
-			c.SetLogger(logging.MustGetLogger(fmt.Sprintf("Initiator %d", i)))
+			c := NewClient(pk, sk, dc, SetLogger(logging.MustGetLogger(fmt.Sprintf("initiator_%d", i))))
 			err := c.InitiateServerConnections(context.Background(), 1)
 			require.NoError(t, err)
 
@@ -327,8 +324,7 @@ func TestServer_Serve(t *testing.T) {
 		for i := 0; i < remotesCount; i++ {
 			pk, sk := cipher.GenerateKeyPair()
 
-			c := NewClient(pk, sk, dc)
-			c.SetLogger(logging.MustGetLogger(fmt.Sprintf("Remote %d", i)))
+			c := NewClient(pk, sk, dc, SetLogger(logging.MustGetLogger(fmt.Sprintf("remote_%d", i))))
 			if _, ok := usedRemotes[i]; ok {
 				err := c.InitiateServerConnections(context.Background(), 1)
 				require.NoError(t, err)
@@ -550,14 +546,12 @@ func TestServer_Serve(t *testing.T) {
 		bPK, bSK := cipher.GenerateKeyPair()
 
 		// create remote
-		a := NewClient(aPK, aSK, dc)
-		a.SetLogger(logging.MustGetLogger("A"))
+		a := NewClient(aPK, aSK, dc, SetLogger(logging.MustGetLogger("A")))
 		err = a.InitiateServerConnections(context.Background(), 1)
 		require.NoError(t, err)
 
 		// create initiator
-		b := NewClient(bPK, bSK, dc)
-		b.SetLogger(logging.MustGetLogger("B"))
+		b := NewClient(bPK, bSK, dc, SetLogger(logging.MustGetLogger("B")))
 		err = b.InitiateServerConnections(context.Background(), 1)
 		require.NoError(t, err)
 
@@ -648,14 +642,12 @@ func TestServer_Serve(t *testing.T) {
 		bPK, bSK := cipher.GenerateKeyPair()
 
 		// create remote
-		a := NewClient(aPK, aSK, dc)
-		a.SetLogger(logging.MustGetLogger("A"))
+		a := NewClient(aPK, aSK, dc, SetLogger(logging.MustGetLogger("A")))
 		err = a.InitiateServerConnections(context.Background(), 1)
 		require.NoError(t, err)
 
 		// create initiator
-		b := NewClient(bPK, bSK, dc)
-		b.SetLogger(logging.MustGetLogger("B"))
+		b := NewClient(bPK, bSK, dc, SetLogger(logging.MustGetLogger("B")))
 		err = b.InitiateServerConnections(context.Background(), 1)
 		require.NoError(t, err)
 
@@ -804,12 +796,10 @@ func TestNewClient(t *testing.T) {
 
 	go s.Serve() //nolint:errcheck
 
-	a := NewClient(aPK, aSK, dc)
-	a.SetLogger(logging.MustGetLogger("A"))
+	a := NewClient(aPK, aSK, dc, SetLogger(logging.MustGetLogger("A")))
 	require.NoError(t, a.InitiateServerConnections(context.Background(), 1))
 
-	b := NewClient(bPK, bSK, dc)
-	b.SetLogger(logging.MustGetLogger("B"))
+	b := NewClient(bPK, bSK, dc, SetLogger(logging.MustGetLogger("B")))
 	require.NoError(t, b.InitiateServerConnections(context.Background(), 1))
 
 	wg := new(sync.WaitGroup)

pkg/dmsg/transport.go

@@ -128,9 +128,8 @@ func (tp *Transport) Type() string {
 	return Type
 }
 
-// Inject injects a frame from 'ClientConn' to transport.
-// Frame is then handled by 'tp.Serve'.
-func (tp *Transport) Inject(f Frame) error {
+// HandleFrame allows 'tp.Serve' to handle the frame (typically from 'ClientConn').
+func (tp *Transport) HandleFrame(f Frame) error {
 	if tp.IsClosed() {
 		return io.ErrClosedPipe
 	}

pkg/node/node.go

@@ -124,8 +124,7 @@ func NewNode(config *Config) (*Node, error) {
 		return nil, fmt.Errorf("invalid Messaging config: %s", err)
 	}
 
-	node.messenger = dmsg.NewClient(mConfig.PubKey, mConfig.SecKey, mConfig.Discovery)
-	node.messenger.SetLogger(node.Logger.PackageLogger(dmsg.Type))
+	node.messenger = dmsg.NewClient(mConfig.PubKey, mConfig.SecKey, mConfig.Discovery, dmsg.SetLogger(node.Logger.PackageLogger(dmsg.Type)))
 
 	trDiscovery, err := config.TransportDiscovery()
 	if err != nil {

pkg/setup/node.go

@@ -45,8 +45,7 @@ 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, mClient.NewHTTP(conf.Messaging.Discovery))
-	messenger.SetLogger(logger.PackageLogger(dmsg.Type))
+	messenger := dmsg.NewClient(pk, sk, mClient.NewHTTP(conf.Messaging.Discovery), dmsg.SetLogger(logger.PackageLogger(dmsg.Type)))
 
 	trDiscovery, err := trClient.NewHTTP(conf.TransportDiscovery, pk, sk)
 	if err != nil {