pss/outbox: Split tests in white and black box. Added Start and Stop to interface

Author: aperonam

pss/outbox/mock.go

@@ -14,23 +14,19 @@ var mockForwardFunction = func(msg *message.Message) error {
 	return nil
 }
 
-func NewOutboxMock(config *Config) (outbox *outbox) {
+func NewMock(config *Config) (outboxMock Outbox) {
 	if config == nil {
 		config = &Config{
 			NumberSlots: defaultOutboxCapacity,
-			QuitC:       make(chan struct{}),
 			Forward:     mockForwardFunction,
 		}
 	} else {
 		if config.Forward == nil {
 			config.Forward = mockForwardFunction
 		}
-		if &config.NumberSlots == nil {
+		if config.NumberSlots == 0 {
 			config.NumberSlots = defaultOutboxCapacity
 		}
-		if config.QuitC == nil {
-			config.QuitC = make(chan struct{})
-		}
 	}
-	return NewOutbox(config)
+	return New(config)
 }

pss/outbox/outbox.go

@@ -9,13 +9,13 @@ import (
 
 type Outbox interface {
 	Enqueue(outboxMsg *outboxMsg) error
-	ProcessOutbox()
-	SetForwardFunction(forwardFunc forwardFunction) // Needed right now to mock forwardFunc in one pss_tests test
+	Start()
+	Stop()
+	SetForwardFunction(forwardFunc forwardFunction) // Needed right now to mock forwardFunc in one pss_tests.go test
 }
 
 type Config struct {
 	NumberSlots int
-	QuitC       chan struct{}
 	Forward     forwardFunction
 }
 
@@ -24,20 +24,22 @@ type outbox struct {
 	queue       []*outboxMsg
 	slots       chan int
 	process     chan int
-	quitC       chan struct{}
 }
 
 type forwardFunction func(msg *message.Message) error
 
 var ErrOutboxFull = errors.New("outbox full")
 
-func NewOutbox(config *Config) *outbox {
+func New(config *Config) Outbox {
+	return newOutbox(config)
+}
+
+func newOutbox(config *Config) *outbox {
 	outbox := outbox{
 		forwardFunc: config.Forward,
 		queue:       make([]*outboxMsg, config.NumberSlots),
 		slots:       make(chan int, config.NumberSlots),
 		process:     make(chan int),
-		quitC:       config.QuitC,
 	}
 	// fill up outbox slots
 	for i := 0; i < cap(outbox.slots); i++ {
@@ -46,6 +48,16 @@ func NewOutbox(config *Config) *outbox {
 	return &outbox
 }
 
+func (o *outbox) Start() {
+	log.Info("Starting outbox")
+	go o.processOutbox()
+}
+
+func (o *outbox) Stop() {
+	log.Warn("Stopping outbox")
+	close(o.process)
+}
+
 // Enqueue a new element in the outbox if there is any slot available.
 // Then send it to process. This method is blocking in the process channel!
 func (o *outbox) Enqueue(outboxMsg *outboxMsg) error {
@@ -55,10 +67,7 @@ func (o *outbox) Enqueue(outboxMsg *outboxMsg) error {
 		o.queue[slot] = outboxMsg
 		metrics.GetOrRegisterGauge("pss.outbox.len", nil).Update(int64(o.len()))
 		// we send this message slot to process
-		select {
-		case o.process <- slot:
-		case <-o.quitC:
-		}
+		o.process <- slot
 		return nil
 	default:
 		metrics.GetOrRegisterCounter("pss.enqueue.outbox.full", nil).Inc(1)
@@ -67,7 +76,7 @@ func (o *outbox) Enqueue(outboxMsg *outboxMsg) error {
 }
 
 //ProcessOutbox starts a routine that tries to forward messages present in the outbox queue
-func (o *outbox) ProcessOutbox() {
+func (o *outbox) processOutbox() {
 	for slot := range o.process {
 		go func(slot int) {
 			msg := o.msg(slot)
@@ -100,10 +109,7 @@ func (o *outbox) free(slot int) {
 }
 
 func (o *outbox) requeue(slot int) {
-	select {
-	case o.process <- slot:
-	case <-o.quitC:
-	}
+	o.process <- slot
 }
 
 func (o *outbox) len() int {

pss/outbox/outbox_test.go

@@ -1,7 +1,8 @@
-package outbox
+package outbox_test
 
 import (
 	"errors"
+	"github.com/ethersphere/swarm/pss/outbox"
 	"testing"
 	"time"
 
@@ -16,104 +17,65 @@ func TestOutbox(tx *testing.T) {
 	defer t.FinishTest()
 
 	outboxCapacity := 2
+	failed := make([]interface{}, 0)
+	failedC := make(chan struct{})
 	successC := make(chan struct{})
+	continueC := make(chan struct{})
+
+	forwardFail := false
+
 	mockForwardFunction := func(msg *message.Message) error {
-		successC <- struct{}{}
-		return nil
+		if !forwardFail {
+			successC <- struct{}{}
+			return nil
+		} else {
+			failed = append(failed, msg)
+			failedC <- struct{}{}
+			<-continueC
+			return errors.New("forced test error forwarding message")
+		}
 	}
 
-	testOutbox := NewOutboxMock(&Config{
+	testOutbox := outbox.NewMock(&outbox.Config{
 		NumberSlots: outboxCapacity,
-		QuitC:       nil,
 		Forward:     mockForwardFunction,
 	})
-	go testOutbox.ProcessOutbox()
+
+	testOutbox.Start()
+	defer testOutbox.Stop()
 
 	err := testOutbox.Enqueue(testOutboxMessage)
 	t.Ok(err)
 
-	usedSlots := testOutbox.len()
-	t.Assert(usedSlots == 1, "incorrect outbox length. expected 1, got %v", usedSlots)
-
 	select {
 	case <-successC:
 	case <-time.After(waitTimeout):
-		t.Fatal("timeout waiting for success forward")
+		t.Fatal("timeout waiting for first success forward")
 	}
 
-	failed := make([]interface{}, 0)
-	failedC := make(chan struct{})
-	continueC := make(chan struct{})
-	failedForward := func(msg *message.Message) error {
-		failed = append(failed, msg)
-		failedC <- struct{}{}
-		<-continueC
-		return errors.New("forced test error forwarding message")
-	}
-	testOutbox.forwardFunc = failedForward
-
+	forwardFail = true
 	err = testOutbox.Enqueue(testOutboxMessage)
 	t.Ok(err)
 
-	select {
-	case <-failedC:
-	case <-time.After(waitTimeout):
-		t.Fatal("timeout waiting for failing forward")
-	}
+	//We wait for the forward function to fail
+	<-failedC
 
 	failedMessages := len(failed)
 	t.Assert(failedMessages == 1, "incorrect number of failed messages, expected 1 got %v", failedMessages)
 
 	// The message will be retried once we send to continueC, so first, we change the forward function
-	testOutbox.forwardFunc = mockForwardFunction
+	forwardFail = false
 	continueC <- struct{}{}
 
+	//We wait for the retry and success
 	select {
 	case <-successC:
 	case <-time.After(waitTimeout):
 		t.Fatal("timeout waiting for second success forward")
 	}
 }
 
-func TestOutboxFull(tx *testing.T) {
-	t := ut.BeginTest(tx, false)
-	defer t.FinishTest()
-
-	outboxCapacity := 2
-	processC := make(chan struct{})
-	successForward := func(msg *message.Message) error {
-		<-processC
-		return nil
-	}
-
-	testOutbox := NewOutboxMock(&Config{
-		NumberSlots: outboxCapacity,
-		QuitC:       nil,
-		Forward:     successForward,
-	})
-	go testOutbox.ProcessOutbox()
-
-	err := testOutbox.Enqueue(testOutboxMessage)
-	t.Ok(err)
-
-	err = testOutbox.Enqueue(testOutboxMessage)
-	t.Ok(err)
-
-	//As we haven't signaled processC, the messages are still in the outbox
-	err = testOutbox.Enqueue(testOutboxMessage)
-	t.MustFailWith(err, ErrOutboxFull)
-
-	processC <- struct{}{}
-
-	//There should be a slot again in the outbox
-	select {
-	case <-testOutbox.slots:
-	case <-time.After(waitTimeout):
-		t.Fatalf("timeout waiting for a free slot")
-	}
-}
-
-var testOutboxMessage = NewOutboxMessage(&message.Message{
+var testOutboxMessage = outbox.NewOutboxMessage(&message.Message{
 	To:      nil,
 	Flags:   message.Flags{},
 	Expire:  0,

pss/outbox/outbox_whitebox_test.go

@@ -0,0 +1,54 @@
+package outbox
+
+import (
+	"github.com/epiclabs-io/ut"
+	"github.com/ethersphere/swarm/pss/message"
+	"testing"
+	"time"
+)
+
+func TestOutboxFull(tx *testing.T) {
+	t := ut.BeginTest(tx, false)
+	defer t.FinishTest()
+
+	outboxCapacity := 2
+	processC := make(chan struct{})
+	successForward := func(msg *message.Message) error {
+		<-processC
+		return nil
+	}
+
+	testOutbox := newOutbox(&Config{
+		NumberSlots: outboxCapacity,
+		Forward:     successForward,
+	})
+	testOutbox.Start()
+	defer testOutbox.Stop()
+
+	err := testOutbox.Enqueue(testOutboxMessage)
+	t.Ok(err)
+
+	err = testOutbox.Enqueue(testOutboxMessage)
+	t.Ok(err)
+
+	//As we haven't signaled processC, the messages are still in the outbox
+	err = testOutbox.Enqueue(testOutboxMessage)
+	t.MustFailWith(err, ErrOutboxFull)
+
+	processC <- struct{}{}
+	//There should be a slot again in the outbox to enqueue
+	select {
+	case <-testOutbox.slots:
+	case <-time.After(2 * time.Second):
+		t.Fatalf("timeout waiting for a free slot")
+	}
+
+}
+
+var testOutboxMessage = NewOutboxMessage(&message.Message{
+	To:      nil,
+	Flags:   message.Flags{},
+	Expire:  0,
+	Topic:   message.Topic{},
+	Payload: nil,
+})

pss/pss.go

@@ -183,9 +183,8 @@ func New(k *network.Kademlia, params *Params) (*Pss, error) {
 			},
 		},
 	}
-	ps.outbox = outbox.NewOutbox(&outbox.Config{
+	ps.outbox = outbox.New(&outbox.Config{
 		NumberSlots: defaultOutboxCapacity,
-		QuitC:       ps.quitC,
 		Forward:     ps.forward,
 	})
 
@@ -220,8 +219,7 @@ func (p *Pss) Start(srv *p2p.Server) error {
 	}()
 
 	// Forward outbox messages
-	go p.outbox.ProcessOutbox()
-
+	p.outbox.Start()
 	log.Info("Started Pss")
 	log.Info("Loaded EC keys", "pubkey", hex.EncodeToString(p.Crypto.SerializePublicKey(p.PublicKey())), "secp256", hex.EncodeToString(p.Crypto.CompressPublicKey(p.PublicKey())))
 	return nil
@@ -230,6 +228,7 @@ func (p *Pss) Start(srv *p2p.Server) error {
 func (p *Pss) Stop() error {
 	log.Info("Pss shutting down")
 	close(p.quitC)
+	p.outbox.Stop()
 	return nil
 }
 

pss/pss_test.go

@@ -269,7 +269,7 @@ func TestAddressMatchProx(t *testing.T) {
 		}
 
 	ps.outbox.SetForwardFunction(mockForwardFunc)
-	go ps.outbox.ProcessOutbox()
+	ps.outbox.Start()
 
 	// create kademlia peers, so we have peers both inside and outside minproxlimit
 	var peers []*network.Peer