Merge pull request #508 from evanlinjin/feature/decouple-managedTrans…

…port

[WIP] Decouple managedTransport and improve transport handling logic.

pkg/route-finder/client/mock.go

@@ -34,15 +34,15 @@ func (r *mockClient) PairedRoutes(src, dst cipher.PubKey, minHops, maxHops uint1
 				&routing.Hop{
 					From:      src,
 					To:        dst,
-					Transport: transport.MakeTransportID(src, dst, "", true),
+					Transport: transport.MakeTransportID(src, dst, ""),
 				},
 			},
 		}, []routing.Route{
 			{
 				&routing.Hop{
 					From:      src,
 					To:        dst,
-					Transport: transport.MakeTransportID(src, dst, "", true),
+					Transport: transport.MakeTransportID(src, dst, ""),
 				},
 			},
 		}, nil

pkg/router/app_manager.go

@@ -1,8 +1,10 @@
 package router
 
 import (
+	"context"
 	"encoding/json"
 	"errors"
+	"time"
 
 	"github.com/skycoin/skycoin/src/util/logging"
 
@@ -13,9 +15,9 @@ import (
 const supportedProtocolVersion = "0.0.1"
 
 type appCallbacks struct {
-	CreateLoop func(conn *app.Protocol, raddr routing.Addr) (laddr routing.Addr, err error)
-	CloseLoop  func(conn *app.Protocol, loop routing.Loop) error
-	Forward    func(conn *app.Protocol, packet *app.Packet) error
+	CreateLoop func(ctx context.Context, conn *app.Protocol, raddr routing.Addr) (laddr routing.Addr, err error)
+	CloseLoop  func(ctx context.Context, conn *app.Protocol, loop routing.Loop) error
+	Forward    func(ctx context.Context, conn *app.Protocol, packet *app.Packet) error
 }
 
 type appManager struct {
@@ -29,15 +31,19 @@ type appManager struct {
 func (am *appManager) Serve() error {
 	return am.proto.Serve(func(frame app.Frame, payload []byte) (res interface{}, err error) {
 		am.Logger.Infof("Got new App request with type %s: %s", frame, string(payload))
+
+		ctx, cancel := context.WithTimeout(context.Background(), time.Second*30)
+		defer cancel()
+
 		switch frame {
 		case app.FrameInit:
 			err = am.initApp(payload)
 		case app.FrameCreateLoop:
-			res, err = am.setupLoop(payload)
+			res, err = am.setupLoop(ctx, payload)
 		case app.FrameClose:
-			err = am.handleCloseLoop(payload)
+			err = am.handleCloseLoop(ctx, payload)
 		case app.FrameSend:
-			err = am.forwardAppPacket(payload)
+			err = am.forwardAppPacket(ctx, payload)
 		default:
 			err = errors.New("unexpected frame")
 		}
@@ -72,29 +78,26 @@ func (am *appManager) initApp(payload []byte) error {
 	return nil
 }
 
-func (am *appManager) setupLoop(payload []byte) (routing.Addr, error) {
+func (am *appManager) setupLoop(ctx context.Context, payload []byte) (routing.Addr, error) {
 	var raddr routing.Addr
 	if err := json.Unmarshal(payload, &raddr); err != nil {
 		return routing.Addr{}, err
 	}
-
-	return am.callbacks.CreateLoop(am.proto, raddr)
+	return am.callbacks.CreateLoop(ctx, am.proto, raddr)
 }
 
-func (am *appManager) handleCloseLoop(payload []byte) error {
+func (am *appManager) handleCloseLoop(ctx context.Context, payload []byte) error {
 	var loop routing.Loop
 	if err := json.Unmarshal(payload, &loop); err != nil {
 		return err
 	}
-
-	return am.callbacks.CloseLoop(am.proto, loop)
+	return am.callbacks.CloseLoop(ctx, am.proto, loop)
 }
 
-func (am *appManager) forwardAppPacket(payload []byte) error {
+func (am *appManager) forwardAppPacket(ctx context.Context, payload []byte) error {
 	packet := &app.Packet{}
 	if err := json.Unmarshal(payload, packet); err != nil {
 		return err
 	}
-
-	return am.callbacks.Forward(am.proto, packet)
+	return am.callbacks.Forward(ctx, am.proto, packet)
 }

pkg/router/app_manager_test.go

@@ -1,6 +1,7 @@
 package router
 
 import (
+	"context"
 	"net"
 	"testing"
 
@@ -66,7 +67,7 @@ func TestAppManagerSetupLoop(t *testing.T) {
 		app.NewProtocol(out),
 		&app.Config{AppName: "foo", AppVersion: "0.0.1"},
 		&appCallbacks{
-			CreateLoop: func(conn *app.Protocol, raddr routing.Addr) (laddr routing.Addr, err error) {
+			CreateLoop: func(ctx context.Context, conn *app.Protocol, raddr routing.Addr) (laddr routing.Addr, err error) {
 				return raddr, nil
 			},
 		},
@@ -102,7 +103,7 @@ func TestAppManagerCloseLoop(t *testing.T) {
 		app.NewProtocol(out),
 		&app.Config{AppName: "foo", AppVersion: "0.0.1"},
 		&appCallbacks{
-			CloseLoop: func(conn *app.Protocol, loop routing.Loop) error {
+			CloseLoop: func(ctx context.Context, conn *app.Protocol, loop routing.Loop) error {
 				inLoop = loop
 				return nil
 			},
@@ -139,7 +140,7 @@ func TestAppManagerForward(t *testing.T) {
 		app.NewProtocol(out),
 		&app.Config{AppName: "foo", AppVersion: "0.0.1"},
 		&appCallbacks{
-			Forward: func(conn *app.Protocol, packet *app.Packet) error {
+			Forward: func(ctx context.Context, conn *app.Protocol, packet *app.Packet) error {
 				inPacket = packet
 				return nil
 			},

pkg/router/route_manager.go

@@ -35,6 +35,12 @@ func (rm *routeManager) GetRule(routeID routing.RouteID) (routing.Rule, error) {
 		return nil, errors.New("unknown RouteID")
 	}
 
+	// TODO(evanlinjin): This is a workaround for ensuring the read-in rule is of the correct size.
+	// Sometimes it is not, causing a segfault later down the line.
+	if len(rule) < routing.RuleHeaderSize {
+		return nil, errors.New("corrupted rule")
+	}
+
 	if rule.Expiry().Before(time.Now()) {
 		return nil, errors.New("expired routing rule")
 	}

pkg/router/route_manager_test.go

@@ -1,6 +1,7 @@
 package router
 
 import (
+	"context"
 	"net"
 	"testing"
 	"time"
@@ -81,7 +82,7 @@ func TestRouteManagerAddRemoveRule(t *testing.T) {
 	proto := setup.NewSetupProtocol(in)
 
 	rule := routing.ForwardRule(time.Now(), 3, uuid.New())
-	id, err := setup.AddRule(proto, rule)
+	id, err := setup.AddRule(context.TODO(), proto, rule)
 	require.NoError(t, err)
 	assert.Equal(t, routing.RouteID(1), id)
 
@@ -111,7 +112,7 @@ func TestRouteManagerDeleteRules(t *testing.T) {
 	require.NoError(t, err)
 	assert.Equal(t, 1, rt.Count())
 
-	require.NoError(t, setup.DeleteRule(proto, id))
+	require.NoError(t, setup.DeleteRule(context.TODO(), proto, id))
 	assert.Equal(t, 0, rt.Count())
 
 	require.NoError(t, in.Close())
@@ -157,7 +158,7 @@ func TestRouteManagerConfirmLoop(t *testing.T) {
 		},
 		RouteID: 1,
 	}
-	err := setup.ConfirmLoop(proto, ld)
+	err := setup.ConfirmLoop(context.TODO(), proto, ld)
 	require.NoError(t, err)
 	assert.Equal(t, rule, inRule)
 	assert.Equal(t, routing.Port(2), inLoop.Local.Port)
@@ -207,7 +208,7 @@ func TestRouteManagerLoopClosed(t *testing.T) {
 		},
 		RouteID: 1,
 	}
-	require.NoError(t, setup.LoopClosed(proto, ld))
+	require.NoError(t, setup.LoopClosed(context.TODO(), proto, ld))
 	assert.Equal(t, routing.Port(2), inLoop.Local.Port)
 	assert.Equal(t, routing.Port(3), inLoop.Remote.Port)
 	assert.Equal(t, pk, inLoop.Remote.PubKey)