Merge pull request #423 from nkryuchkov/feature/dmsg-reconnect-test

DMSG reconnection tests

go.sum

@@ -118,8 +118,7 @@ golang.org/x/sys v0.0.0-20190222072716-a9d3bda3a223/go.mod h1:STP8DvDyc/dI5b8T5h
 golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190422165155-953cdadca894 h1:Cz4ceDQGXuKRnVBDTS23GTn/pU5OE2C0WrNTOYK1Uuc=
 golang.org/x/sys v0.0.0-20190422165155-953cdadca894/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.0.0-20190624142023-c5567b49c5d0 h1:HyfiK1WMnHj5FXFXatD+Qs1A/xC2Run6RzeW1SyHxpc=
-golang.org/x/sys v0.0.0-20190624142023-c5567b49c5d0/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=

pkg/dmsg/client.go

@@ -19,6 +19,10 @@ import (
 	"github.com/skycoin/skywire/pkg/transport"
 )
 
+const (
+	clientReconnectInterval = 3 * time.Second
+)
+
 var (
 	// ErrNoSrv indicate that remote client does not have DelegatedServers in entry.
 	ErrNoSrv = errors.New("remote has no DelegatedServers")
@@ -468,7 +472,7 @@ func (c *Client) findOrConnectToServer(ctx context.Context, srvPK cipher.PubKey)
 		select {
 		case <-c.done:
 		case <-ctx.Done():
-		case <-time.After(time.Second * 3):
+		case <-time.After(clientReconnectInterval):
 			conn.log.WithField("remoteServer", srvPK).Warn("Reconnecting")
 			if _, err := c.findOrConnectToServer(ctx, srvPK); err != nil {
 				conn.log.WithError(err).WithField("remoteServer", srvPK).Warn("ReconnectionFailed")

pkg/dmsg/server_test.go

@@ -889,6 +889,115 @@ func TestServer_Serve(t *testing.T) {
 		require.Equal(t, true, ok)
 		require.Equal(t, true, tp.IsClosed())*/
 	})
+
+	t.Run("Reconnect to server should succeed", func(t *testing.T) {
+		t.Run("Same address", func(t *testing.T) {
+			t.Parallel()
+			testReconnect(t, false)
+		})
+
+		t.Run("Random address", func(t *testing.T) {
+			t.Parallel()
+			testReconnect(t, true)
+		})
+	})
+}
+
+func testReconnect(t *testing.T, randomAddr bool) {
+	const smallDelay = 100 * time.Millisecond
+	ctx := context.TODO()
+
+	serverPK, serverSK := cipher.GenerateKeyPair()
+	dc := client.NewMock()
+
+	l, err := nettest.NewLocalListener("tcp")
+	require.NoError(t, err)
+
+	s, err := NewServer(serverPK, serverSK, "", l, dc)
+	require.NoError(t, err)
+
+	serverAddr := s.Addr()
+
+	go s.Serve() // nolint:errcheck
+
+	remotePK, remoteSK := cipher.GenerateKeyPair()
+	initiatorPK, initiatorSK := cipher.GenerateKeyPair()
+
+	assert.Equal(t, 0, s.connCount())
+
+	remote := NewClient(remotePK, remoteSK, dc, SetLogger(logging.MustGetLogger("remote")))
+	err = remote.InitiateServerConnections(ctx, 1)
+	require.NoError(t, err)
+
+	require.NoError(t, testWithTimeout(smallDelay, func() error {
+		if s.connCount() != 1 {
+			return errors.New("s.conns is not equal to 1")
+		}
+		return nil
+	}))
+
+	initiator := NewClient(initiatorPK, initiatorSK, dc, SetLogger(logging.MustGetLogger("initiator")))
+	err = initiator.InitiateServerConnections(ctx, 1)
+	require.NoError(t, err)
+
+	initiatorTransport, err := initiator.Dial(ctx, remotePK)
+	require.NoError(t, err)
+
+	remoteTransport, err := remote.Accept(context.Background())
+	require.NoError(t, err)
+
+	require.NoError(t, testWithTimeout(smallDelay, func() error {
+		if s.connCount() != 2 {
+			return errors.New("s.conns is not equal to 2")
+		}
+		return nil
+	}))
+
+	err = s.Close()
+	assert.NoError(t, err)
+
+	initTr := initiatorTransport.(*Transport)
+	assert.False(t, isDoneChannelOpen(initTr.done))
+	assert.False(t, isReadChannelOpen(initTr.inCh))
+
+	remoteTr := remoteTransport.(*Transport)
+	assert.False(t, isDoneChannelOpen(remoteTr.done))
+	assert.False(t, isReadChannelOpen(remoteTr.inCh))
+
+	assert.Equal(t, 0, s.connCount())
+
+	addr := ""
+	if !randomAddr {
+		addr = serverAddr
+	}
+
+	l, err = net.Listen("tcp", serverAddr)
+	require.NoError(t, err)
+
+	s, err = NewServer(serverPK, serverSK, addr, l, dc)
+	require.NoError(t, err)
+
+	go s.Serve() // nolint:errcheck
+
+	require.NoError(t, testWithTimeout(clientReconnectInterval+smallDelay, func() error {
+		if s.connCount() != 2 {
+			return errors.New("s.conns is not equal to 2")
+		}
+		return nil
+	}))
+
+	require.NoError(t, testWithTimeout(smallDelay, func() error {
+		_, err = initiator.Dial(ctx, remotePK)
+		if err != nil {
+			return err
+		}
+
+		_, err = remote.Accept(context.Background())
+		return err
+	}))
+
+	err = s.Close()
+	assert.NoError(t, err)
 }
 
 // Given two client instances (a & b) and a server instance (s),