feat(core): add network.Driver interface + an Enqueuer mock for tests

.gometalinter.json

@@ -12,8 +12,7 @@
     ".*\\.gen\\.go",
     "should have comment or be unexported",
     "error return value not checked \\(defer",
-    "jsonPrint is unused",
-    "jsonPrintIndent is unused"
+    "_test.go:"
   ],
   "EnableGC": true,
   "Enable": [

core/api/p2p/event.go

@@ -1,6 +1,7 @@
 package p2p
 
 import (
+	"encoding/json"
 	"strings"
 	"time"
 )
@@ -44,3 +45,9 @@ func (e Event) Copy() *Event {
 func (e Event) Author() string {
 	return strings.Split(e.ID, ":")[0]
 }
+
+func (e Event) ToJSON() string {
+	// FIXME: use jsonpb
+	out, _ := json.Marshal(e)
+	return string(out)
+}

core/network/driver.go

@@ -0,0 +1,11 @@
+package network
+
+import (
+	"context"
+
+	"github.com/berty/berty/core/api/p2p"
+)
+
+type Driver interface {
+	SendEvent(ctx context.Context, event *p2p.Event) error
+}

core/network/drivermock/euqueuer.go

@@ -0,0 +1,29 @@
+package drivermock
+
+import (
+	"context"
+
+	"github.com/berty/berty/core/api/p2p"
+	"github.com/berty/berty/core/network"
+)
+
+type Enqueuer struct {
+	network.Driver
+
+	queue chan *p2p.Event
+}
+
+func NewEnqueuer() *Enqueuer {
+	return &Enqueuer{
+		queue: make(chan *p2p.Event, 100),
+	}
+}
+
+func (e *Enqueuer) Queue() chan *p2p.Event {
+	return e.queue
+}
+
+func (e *Enqueuer) SendEvent(_ context.Context, event *p2p.Event) error {
+	e.queue <- event
+	return nil
+}

core/node/network.go

@@ -0,0 +1,9 @@
+package node
+
+import "github.com/berty/berty/core/network"
+
+func WithNetworkDriver(driver network.Driver) NewNodeOption {
+	return func(n *Node) {
+		n.networkDriver = driver
+	}
+}

core/node/node.go

@@ -1,6 +1,7 @@
 package node
 
 import (
+	"context"
 	"fmt"
 	"sync"
 
@@ -11,6 +12,7 @@ import (
 
 	"github.com/berty/berty/core/api/p2p"
 	"github.com/berty/berty/core/entity"
+	"github.com/berty/berty/core/network"
 )
 
 // Node is the top-level object of a Berty peer
@@ -21,6 +23,7 @@ type Node struct {
 	config         *entity.Config
 	initDevice     *entity.Device
 	handleMutex    sync.Mutex
+	networkDriver  network.Driver
 }
 
 // New initializes a new Node object
@@ -59,7 +62,15 @@ func New(opts ...NewNodeOption) (*Node, error) {
 
 // Start is the node's mainloop
 func (n *Node) Start() error {
-	select {}
+	ctx := context.Background()
+	for {
+		select {
+		case event := <-n.outgoingEvents:
+			if err := n.networkDriver.SendEvent(ctx, event); err != nil {
+				zap.L().Warn("failed to send outgoing event", zap.Error(err), zap.String("event", event.ToJSON()))
+			}
+		}
+	}
 }
 
 // Close closes object initialized by Node itself
@@ -71,7 +82,7 @@ func (n *Node) Close() error {
 
 // Validate returns an error if object is invalid
 func (n *Node) Validate() error {
-	if n == nil || n.sql == nil || n.initDevice == nil {
+	if n == nil || n.sql == nil || n.initDevice == nil || n.networkDriver == nil {
 		return errors.New("missing required fields to create a new Node")
 	}
 	return nil
@@ -93,5 +104,4 @@ func (n *Node) UserID() string {
 	return n.config.Myself.ID
 }
 
-func (n *Node) OutgoingEventsChan() chan *p2p.Event { return n.outgoingEvents }
-func (n *Node) ClientEventsChan() chan *p2p.Event   { return n.clientEvents }
+func (n *Node) ClientEventsChan() chan *p2p.Event { return n.clientEvents }

core/test/app_mock.go

@@ -5,40 +5,45 @@ import (
 	"fmt"
 	"io/ioutil"
 	"net"
+	"strings"
 
 	"github.com/jinzhu/gorm"
 	"github.com/pkg/errors"
+	"go.uber.org/zap"
 	"google.golang.org/grpc"
 
 	"github.com/berty/berty/core/api/p2p"
 	"github.com/berty/berty/core/client"
 	"github.com/berty/berty/core/entity"
+	"github.com/berty/berty/core/network"
 	"github.com/berty/berty/core/network/netutil"
 	"github.com/berty/berty/core/node"
 	"github.com/berty/berty/core/sql"
 	"github.com/berty/berty/core/sql/sqlcipher"
 )
 
 type AppMock struct {
-	dbPath     string
-	listener   net.Listener
-	db         *gorm.DB
-	node       *node.Node
-	clientConn *grpc.ClientConn
-	client     *client.Client
-	ctx        context.Context
-	device     *entity.Device
+	dbPath        string
+	listener      net.Listener
+	db            *gorm.DB
+	node          *node.Node
+	clientConn    *grpc.ClientConn
+	client        *client.Client
+	ctx           context.Context
+	device        *entity.Device
+	networkDriver network.Driver
 }
 
-func NewAppMock(device *entity.Device) (*AppMock, error) {
+func NewAppMock(device *entity.Device, networkDriver network.Driver) (*AppMock, error) {
 	tmpFile, err := ioutil.TempFile("", "sqlite")
 	if err != nil {
 		return nil, err
 	}
 
 	a := AppMock{
-		dbPath: tmpFile.Name(),
-		device: device,
+		dbPath:        tmpFile.Name(),
+		device:        device,
+		networkDriver: networkDriver,
 	}
 
 	if err := a.Open(); err != nil {
@@ -76,12 +81,25 @@ func (a *AppMock) Open() error {
 		node.WithP2PGrpcServer(gs),
 		node.WithNodeGrpcServer(gs),
 		node.WithDevice(a.device),
+		node.WithNetworkDriver(a.networkDriver),
 	); err != nil {
 		return err
 	}
 
 	go func() {
-		_ = gs.Serve(a.listener)
+		if err := gs.Serve(a.listener); err != nil {
+			// app.Close() generates this error
+			if strings.Contains(err.Error(), "use of closed network connection") {
+				return
+			}
+			zap.L().Error("grpc server error", zap.Error(err))
+		}
+	}()
+
+	go func() {
+		if err := a.node.Start(); err != nil {
+			zap.L().Error("node routine error", zap.Error(err))
+		}
 	}()
 
 	a.clientConn, err = grpc.Dial(fmt.Sprintf(":%d", port), grpc.WithInsecure())

core/test/e2e_test.go

@@ -6,10 +6,13 @@ import (
 	"testing"
 
 	. "github.com/smartystreets/goconvey/convey"
+	"go.uber.org/zap"
+	"go.uber.org/zap/zapcore"
 
 	"github.com/berty/berty/core/api/node"
 	"github.com/berty/berty/core/api/p2p"
 	"github.com/berty/berty/core/entity"
+	"github.com/berty/berty/core/network/drivermock"
 )
 
 func Test(t *testing.T) {
@@ -29,15 +32,29 @@ func Test(t *testing.T) {
 			eve.Close()
 		}
 	}()
-	Convey("End-to-end test", t, func() {
-		Convey("Initialize nodes", func() {
-			alice, err = NewAppMock(&entity.Device{Name: "Alice's iPhone"})
+
+	// initialize zap
+	config := zap.NewDevelopmentConfig()
+	config.Level.SetLevel(zap.InfoLevel)
+	config.DisableStacktrace = true
+	config.EncoderConfig.EncodeLevel = zapcore.CapitalColorLevelEncoder
+	logger, err := config.Build()
+	if err != nil {
+		panic(err)
+	}
+	zap.ReplaceGlobals(logger)
+
+	// let's test
+
+	Convey("End-to-end test (manual)", t, FailureHalts, func() {
+		Convey("Initialize nodes", FailureHalts, func() {
+			alice, err = NewAppMock(&entity.Device{Name: "Alice's iPhone"}, drivermock.NewEnqueuer())
 			So(err, ShouldBeNil)
 
-			bob, err = NewAppMock(&entity.Device{Name: "iPhone de Bob"})
+			bob, err = NewAppMock(&entity.Device{Name: "iPhone de Bob"}, drivermock.NewEnqueuer())
 			So(err, ShouldBeNil)
 
-			eve, err = NewAppMock(&entity.Device{Name: "Eve"})
+			eve, err = NewAppMock(&entity.Device{Name: "Eve"}, drivermock.NewEnqueuer())
 			So(err, ShouldBeNil)
 		})
 
@@ -155,7 +172,7 @@ func Test(t *testing.T) {
 				So(len(contacts[1].Devices), ShouldEqual, 0)
 			})
 			Convey("Alice sends a ContactRequest event to Bob", FailureHalts, func() {
-				event := <-alice.node.OutgoingEventsChan()
+				event := <-alice.networkDriver.(*drivermock.Enqueuer).Queue()
 				So(event.Author(), ShouldEqual, alice.node.UserID())
 				So(event.SenderID, ShouldEqual, alice.node.UserID())
 				So(event.Direction, ShouldEqual, p2p.Event_Outgoing)
@@ -196,7 +213,7 @@ func Test(t *testing.T) {
 				So(nodeChansLens(alice, bob, eve), ShouldResemble, []int{0, 0, 1, 0, 0, 0})
 			})
 			Convey("Bob replies an Ack event to Alice's ContactRequest", FailureHalts, func() {
-				event := <-bob.node.OutgoingEventsChan()
+				event := <-bob.networkDriver.(*drivermock.Enqueuer).Queue()
 				So(event.Author(), ShouldEqual, bob.node.UserID())
 				So(event.Kind, ShouldEqual, p2p.Kind_Ack)
 				So(event.SenderID, ShouldEqual, bob.node.UserID())
@@ -246,7 +263,7 @@ func Test(t *testing.T) {
 				So(nodeChansLens(alice, bob, eve), ShouldResemble, []int{0, 0, 2, 0, 0, 0})
 			})
 			Convey("Bob sends a ContactRequestAccepted event to Alice", FailureHalts, func() {
-				event := <-bob.node.OutgoingEventsChan()
+				event := <-bob.networkDriver.(*drivermock.Enqueuer).Queue()
 				So(event.Kind, ShouldEqual, p2p.Kind_ContactRequestAccepted)
 				So(event.SenderAPIVersion, ShouldEqual, p2p.Version)
 				So(event.SenderID, ShouldEqual, bob.node.UserID())
@@ -271,7 +288,7 @@ func Test(t *testing.T) {
 				So(nodeChansLens(alice, bob, eve), ShouldResemble, []int{2, 0, 1, 0, 0, 0})
 			})
 			Convey("Bob sends a ContactShareMe event to Alice", FailureHalts, func() {
-				event := <-bob.node.OutgoingEventsChan()
+				event := <-bob.networkDriver.(*drivermock.Enqueuer).Queue()
 				So(event.Kind, ShouldEqual, p2p.Kind_ContactShareMe)
 				So(event.SenderID, ShouldEqual, bob.node.UserID())
 				So(event.ReceiverID, ShouldEqual, alice.node.UserID())
@@ -298,7 +315,7 @@ func Test(t *testing.T) {
 				So(nodeChansLens(alice, bob, eve), ShouldResemble, []int{3, 0, 0, 0, 0, 0})
 			})
 			Convey("Alice sends a ContactShareMe event to Bob", FailureHalts, func() {
-				event := <-alice.node.OutgoingEventsChan()
+				event := <-alice.networkDriver.(*drivermock.Enqueuer).Queue()
 				So(event.SenderID, ShouldEqual, alice.node.UserID())
 				So(event.Kind, ShouldEqual, p2p.Kind_ContactShareMe)
 				So(event.ReceiverID, ShouldEqual, bob.node.UserID())
@@ -315,7 +332,7 @@ func Test(t *testing.T) {
 				So(nodeChansLens(alice, bob, eve), ShouldResemble, []int{2, 0, 1, 1, 0, 0})
 			})
 			Convey("Alice replies an Ack event to Bob's ContactRequestAccepted", FailureHalts, func() {
-				event := <-alice.node.OutgoingEventsChan()
+				event := <-alice.networkDriver.(*drivermock.Enqueuer).Queue()
 				So(event.SenderID, ShouldEqual, alice.node.UserID())
 				So(event.Kind, ShouldEqual, p2p.Kind_Ack)
 				So(event.ReceiverID, ShouldEqual, bob.node.UserID())
@@ -330,7 +347,7 @@ func Test(t *testing.T) {
 				So(nodeChansLens(alice, bob, eve), ShouldResemble, []int{1, 0, 1, 1, 0, 0})
 			})
 			Convey("Alice replies an Ack event to Bob's ContactShareMe", FailureHalts, func() {
-				event := <-alice.node.OutgoingEventsChan()
+				event := <-alice.networkDriver.(*drivermock.Enqueuer).Queue()
 				So(event.SenderID, ShouldEqual, alice.node.UserID())
 				So(event.Kind, ShouldEqual, p2p.Kind_Ack)
 				So(event.ReceiverID, ShouldEqual, bob.node.UserID())
@@ -345,7 +362,7 @@ func Test(t *testing.T) {
 				So(nodeChansLens(alice, bob, eve), ShouldResemble, []int{0, 0, 1, 1, 0, 0})
 			})
 			Convey("Bob replies an Ack event to Alice's ContactShareMe", FailureHalts, func() {
-				event := <-bob.node.OutgoingEventsChan()
+				event := <-bob.networkDriver.(*drivermock.Enqueuer).Queue()
 				So(event.Kind, ShouldEqual, p2p.Kind_Ack)
 				So(event.SenderID, ShouldEqual, bob.node.UserID())
 				So(event.ReceiverID, ShouldEqual, alice.node.UserID())
@@ -464,12 +481,3 @@ func Test(t *testing.T) {
 	})
 
 }
-
-func nodeChansLens(apps ...*AppMock) []int {
-	out := []int{}
-	for _, app := range apps {
-		out = append(out, len(app.node.OutgoingEventsChan()))
-		out = append(out, len(app.node.ClientEventsChan()))
-	}
-	return out
-}

core/test/network_mock.go

@@ -0,0 +1,28 @@
+package test
+
+import (
+	"context"
+	"errors"
+
+	"github.com/berty/berty/core/api/p2p"
+	"github.com/berty/berty/core/network"
+)
+
+type NetworkMock struct {
+	network.Driver
+	apps []*AppMock
+}
+
+func (n *NetworkMock) SendEventToContact(context.Context, string, *p2p.Event) error {
+	return errors.New("not implemented")
+}
+
+func (n *NetworkMock) AddApp(app *AppMock) {
+	n.apps = append(n.apps, app)
+}
+
+func NewNetworkMock() *NetworkMock {
+	return &NetworkMock{
+		apps: make([]*AppMock, 0),
+	}
+}