pss/outbox: Removed interface

Author: aperonam

pss/outbox/mock.go

@@ -14,7 +14,7 @@ var mockForwardFunction = func(msg *message.Message) error {
 	return nil
 }
 
-func NewMock(config *Config) (outboxMock Outbox) {
+func NewMock(config *Config) (outboxMock *Outbox) {
 	if config == nil {
 		config = &Config{
 			NumberSlots: defaultOutboxCapacity,
@@ -28,5 +28,5 @@ func NewMock(config *Config) (outboxMock Outbox) {
 			config.NumberSlots = defaultOutboxCapacity
 		}
 	}
-	return New(config)
+	return NewOutbox(config)
 }

pss/outbox/outbox.go

@@ -7,19 +7,12 @@ import (
 	"github.com/ethersphere/swarm/pss/message"
 )
 
-type Outbox interface {
-	Enqueue(outboxMsg *outboxMsg) error
-	Start()
-	Stop()
-	SetForwardFunction(forwardFunc forwardFunction) // Needed right now to mock forwardFunc in one pss_tests.go test
-}
-
 type Config struct {
 	NumberSlots int
 	Forward     forwardFunction
 }
 
-type outbox struct {
+type Outbox struct {
 	forwardFunc forwardFunction
 	queue       []*outboxMsg
 	slots       chan int
@@ -30,12 +23,8 @@ type forwardFunction func(msg *message.Message) error
 
 var ErrOutboxFull = errors.New("outbox full")
 
-func New(config *Config) Outbox {
-	return newOutbox(config)
-}
-
-func newOutbox(config *Config) *outbox {
-	outbox := outbox{
+func NewOutbox(config *Config) *Outbox {
+	outbox := &Outbox{
 		forwardFunc: config.Forward,
 		queue:       make([]*outboxMsg, config.NumberSlots),
 		slots:       make(chan int, config.NumberSlots),
@@ -45,22 +34,22 @@ func newOutbox(config *Config) *outbox {
 	for i := 0; i < cap(outbox.slots); i++ {
 		outbox.slots <- i
 	}
-	return &outbox
+	return outbox
 }
 
-func (o *outbox) Start() {
+func (o *Outbox) Start() {
 	log.Info("Starting outbox")
 	go o.processOutbox()
 }
 
-func (o *outbox) Stop() {
-	log.Warn("Stopping outbox")
+func (o *Outbox) Stop() {
+	log.Info("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 {
+func (o *Outbox) Enqueue(outboxMsg *outboxMsg) error {
 	// first we try to obtain a slot in the outbox
 	select {
 	case slot := <-o.slots:
@@ -76,7 +65,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)
@@ -96,22 +85,22 @@ func (o *outbox) processOutbox() {
 		}(slot)
 	}
 }
-func (o *outbox) SetForwardFunction(forwardFunc forwardFunction) {
+func (o *Outbox) SetForwardFunction(forwardFunc forwardFunction) {
 	o.forwardFunc = forwardFunc
 }
 
-func (o *outbox) msg(slot int) *outboxMsg {
+func (o *Outbox) msg(slot int) *outboxMsg {
 	return o.queue[slot]
 }
 
-func (o *outbox) free(slot int) {
+func (o *Outbox) free(slot int) {
 	o.slots <- slot
 }
 
-func (o *outbox) requeue(slot int) {
+func (o *Outbox) requeue(slot int) {
 	o.process <- slot
 }
 
-func (o *outbox) len() int {
+func (o *Outbox) len() int {
 	return cap(o.slots) - len(o.slots)
 }

pss/outbox/outbox_test.go

@@ -47,13 +47,11 @@ func TestOutbox(tx *testing.T) {
 	err := testOutbox.Enqueue(testOutboxMessage)
 	t.Ok(err)
 
-	select {
-	case <-successC:
-	case <-time.After(waitTimeout):
-		t.Fatal("timeout waiting for first success forward")
-	}
+	//We wait for the forward function to success
+	<-successC
 
 	forwardFail = true
+
 	err = testOutbox.Enqueue(testOutboxMessage)
 	t.Ok(err)
 
@@ -68,11 +66,7 @@ func TestOutbox(tx *testing.T) {
 	continueC <- struct{}{}
 
 	//We wait for the retry and success
-	select {
-	case <-successC:
-	case <-time.After(waitTimeout):
-		t.Fatal("timeout waiting for second success forward")
-	}
+	<-successC
 }
 
 var testOutboxMessage = outbox.NewOutboxMessage(&message.Message{

pss/outbox/outbox_whitebox_test.go

@@ -7,7 +7,9 @@ import (
 	"time"
 )
 
-func TestOutboxFull(tx *testing.T) {
+const waitTimeout = 2 * time.Second
+
+func TestFullOutbox(tx *testing.T) {
 	t := ut.BeginTest(tx, false)
 	defer t.FinishTest()
 
@@ -18,7 +20,7 @@ func TestOutboxFull(tx *testing.T) {
 		return nil
 	}
 
-	testOutbox := newOutbox(&Config{
+	testOutbox := NewOutbox(&Config{
 		NumberSlots: outboxCapacity,
 		Forward:     successForward,
 	})
@@ -36,13 +38,9 @@ func TestOutboxFull(tx *testing.T) {
 	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")
-	}
 
+	//There should be a slot again in the outbox to enqueue
+	<-testOutbox.slots
 }
 
 var testOutboxMessage = NewOutboxMessage(&message.Message{

pss/pss.go

@@ -132,7 +132,7 @@ type Pss struct {
 	msgTTL          time.Duration
 	paddingByteSize int
 	capstring       string
-	outbox          outbox.Outbox
+	outbox          *outbox.Outbox
 
 	// message handling
 	handlers           map[message.Topic]map[*handler]bool // topic and version based pss payload handlers. See pss.Handle()
@@ -183,7 +183,7 @@ func New(k *network.Kademlia, params *Params) (*Pss, error) {
 			},
 		},
 	}
-	ps.outbox = outbox.New(&outbox.Config{
+	ps.outbox = outbox.NewOutbox(&outbox.Config{
 		NumberSlots: defaultOutboxCapacity,
 		Forward:     ps.forward,
 	})
@@ -220,6 +220,7 @@ func (p *Pss) Start(srv *p2p.Server) error {
 
 	// Forward outbox messages
 	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