pss: Made some changes due to rebase

Author: aperonam

pss/outbox/message.go

@@ -1,15 +1,16 @@
 package outbox
 
 import (
+	"github.com/ethersphere/swarm/pss/message"
 	"time"
 )
 
 type outboxMsg struct {
-	msg       interface{}
+	msg       *message.Message
 	startedAt time.Time
 }
 
-func NewOutboxMessage(msg interface{}) *outboxMsg {
+func NewOutboxMessage(msg *message.Message) *outboxMsg {
 	return &outboxMsg{
 		msg:       msg,
 		startedAt: time.Now(),

pss/outbox/mock.go

@@ -2,21 +2,35 @@ package outbox
 
 import (
 	"github.com/ethersphere/swarm/log"
+	"github.com/ethersphere/swarm/pss/message"
 )
 
+const (
+	defaultOutboxCapacity = 100000
+)
+
+var mockForwardFunction = func(msg *message.Message) error {
+	log.Debug("Forwarded message", "msg", msg)
+	return nil
+}
+
 func NewOutboxMock(config *Config) (outbox *outbox) {
-	if config.Forward == nil {
-		config.Forward = func(msg interface{}) error {
-			log.Debug("Forwarded message", "msg", msg)
-			return nil
+	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 {
+			config.NumberSlots = defaultOutboxCapacity
+		}
+		if config.QuitC == nil {
+			config.QuitC = make(chan struct{})
 		}
-	}
-
-	if &config.NumberSlots == nil {
-		config.NumberSlots = defaultOutboxCapacity
-	}
-	if config.QuitC == nil {
-		config.QuitC = make(chan struct{})
 	}
 	return NewOutbox(config)
 }

pss/outbox/outbox.go

@@ -4,37 +4,38 @@ import (
 	"errors"
 	"github.com/ethereum/go-ethereum/metrics"
 	"github.com/ethersphere/swarm/log"
-)
-
-const (
-	defaultOutboxCapacity = 100000
+	"github.com/ethersphere/swarm/pss/message"
 )
 
 type Outbox interface {
-	Enqueue(outboxmsg *outboxMsg) error
+	Enqueue(outboxMsg *outboxMsg) error
 	ProcessOutbox()
-	SetForwardFunction(forwardFunc func(msg interface{}) error) //Eventually this will not be needed, when pss has a newMock that creates a mock outbox, this is used just for a test outside of outbox package
+	SetForwardFunction(forwardFunc forwardFunction) // Needed right now to mock forwardFunc in one pss_tests test
 }
 
 type Config struct {
 	NumberSlots int
 	QuitC       chan struct{}
-	Forward     func(msg interface{}) error
+	Forward     forwardFunction
 }
 
 type outbox struct {
-	Config
-	queue   []*outboxMsg
-	slots   chan int
-	process chan int
+	forwardFunc forwardFunction
+	queue       []*outboxMsg
+	slots       chan int
+	process     chan int
+	quitC       chan struct{}
 }
 
+type forwardFunction func(msg *message.Message) error
+
 func NewOutbox(config *Config) *outbox {
 	outbox := outbox{
-		Config:  *config,
-		process: make(chan int),
-		slots:   make(chan int, config.NumberSlots),
-		queue:   make([]*outboxMsg, config.NumberSlots),
+		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++ {
@@ -43,18 +44,18 @@ func NewOutbox(config *Config) *outbox {
 	return &outbox
 }
 
-// enqueue a new element in the outbox if there is any slot available.
+// 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:
-		o.queue[slot] = outboxmsg
+		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:
+		case <-o.quitC:
 		}
 		return nil
 	default:
@@ -63,29 +64,29 @@ 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() {
 	for slot := range o.process {
 		go func(slot int) {
 			msg := o.msg(slot)
 			metrics.GetOrRegisterResettingTimer("pss.handle.outbox", nil).UpdateSince(msg.startedAt)
-			if err := o.Forward(msg.msg); err != nil {
+			if err := o.forwardFunc(msg.msg); err != nil {
 				metrics.GetOrRegisterCounter("pss.forward.err", nil).Inc(1)
 				// if we failed to forward, re-insert message in the queue
 				log.Debug(err.Error())
-				// reenqueue the message for processing
-				o.reenqueue(slot)
-				log.Debug("Message re-enqued", "slot", slot)
+				// requeue the message for processing
+				o.requeue(slot)
+				log.Debug("Message requeued", "slot", slot)
 				return
 			}
-			// free the outbox slot
+			//message processed, free the outbox slot
 			o.free(slot)
 			metrics.GetOrRegisterGauge("pss.outbox.len", nil).Update(int64(o.len()))
 		}(slot)
 	}
 }
-
-func (o *outbox) SetForwardFunction(forwardFunc func(msg interface{}) error) {
-	o.Forward = forwardFunc
+func (o *outbox) SetForwardFunction(forwardFunc forwardFunction) {
+	o.forwardFunc = forwardFunc
 }
 
 func (o *outbox) msg(slot int) *outboxMsg {
@@ -96,10 +97,10 @@ func (o *outbox) free(slot int) {
 	o.slots <- slot
 }
 
-func (o *outbox) reenqueue(slot int) {
+func (o *outbox) requeue(slot int) {
 	select {
 	case o.process <- slot:
-	case <-o.QuitC:
+	case <-o.quitC:
 	}
 }
 

pss/outbox/outbox_test.go

@@ -2,43 +2,37 @@ package outbox
 
 import (
 	"errors"
-	"github.com/ethersphere/swarm/log"
+	"github.com/ethersphere/swarm/pss/message"
 	"testing"
 	"time"
 )
 
-type testMsg struct {
-}
-
 func TestMessageOutbox(t *testing.T) {
 
 	outboxCapacity := 2
 	successC := make(chan struct{})
-	forward := func(msg interface{}) error {
+	mockForwardFunction := func(msg *message.Message) error {
 		successC <- struct{}{}
 		return nil
 	}
 
-	outbox := NewOutboxMock(&Config{
+	testOutbox := NewOutboxMock(&Config{
 		NumberSlots: outboxCapacity,
 		QuitC:       nil,
-		Forward:     forward,
+		Forward:     mockForwardFunction,
 	})
 
-	outboxMessage := NewOutboxMessage(testMsg{})
-
-	go outbox.ProcessOutbox()
+	go testOutbox.ProcessOutbox()
 
-	err := outbox.Enqueue(outboxMessage)
+	err := testOutbox.Enqueue(testOutboxMessage)
 	if err != nil {
 		t.Fatalf("expected no error, got %v", err)
 	}
 
-	usedSlots := outbox.len()
+	usedSlots := testOutbox.len()
 	if usedSlots != 1 {
 		t.Fatalf("incorrect outbox length. expected 1, got %v", usedSlots)
 	}
-	t.Log("Message enqueued", "Outbox len", outbox.len())
 
 	select {
 	case <-successC:
@@ -49,16 +43,16 @@ func TestMessageOutbox(t *testing.T) {
 	failed := make([]interface{}, 0)
 	failedC := make(chan struct{})
 	continueC := make(chan struct{})
-	failedForward := func(msg interface{}) error {
+	failedForward := func(msg *message.Message) error {
 		failed = append(failed, msg)
 		failedC <- struct{}{}
 		<-continueC
 		return errors.New("forced test error forwarding message")
 	}
 
-	outbox.Forward = failedForward
+	testOutbox.forwardFunc = failedForward
 
-	err = outbox.Enqueue(outboxMessage)
+	err = testOutbox.Enqueue(testOutboxMessage)
 	if err != nil {
 		t.Fatalf("Expected no error enqueing, got %v", err.Error())
 	}
@@ -73,7 +67,7 @@ func TestMessageOutbox(t *testing.T) {
 		t.Fatal("Incorrect number of failed messages, expected 1 got 0")
 	}
 	// The message will be retried once we send to continueC, so first, we change the forward function
-	outbox.Forward = forward
+	testOutbox.forwardFunc = mockForwardFunction
 	continueC <- struct{}{}
 	select {
 	case <-successC:
@@ -84,50 +78,48 @@ func TestMessageOutbox(t *testing.T) {
 
 func TestOutboxFull(t *testing.T) {
 
-	addr := make([]byte, 32)
-	addr[0] = 0x01
 	outboxCapacity := 2
-
 	procChan := make(chan struct{})
-	succesForward := func(msg interface{}) error {
+	successForward := func(msg *message.Message) error {
 		<-procChan
-		log.Info("Message processed")
 		return nil
 	}
 
-	outbox := NewOutboxMock(&Config{
+	testOutbox := NewOutboxMock(&Config{
 		NumberSlots: outboxCapacity,
 		QuitC:       nil,
-		Forward:     succesForward,
+		Forward:     successForward,
 	})
-	go outbox.ProcessOutbox()
 
-	err := outbox.Enqueue(outboxTestMessage())
+	go testOutbox.ProcessOutbox()
+
+	err := testOutbox.Enqueue(testOutboxMessage)
 	if err != nil {
 		t.Fatalf("expected no error enqueing first message, got %v", err)
 	}
-	err = outbox.Enqueue(outboxTestMessage())
+	err = testOutbox.Enqueue(testOutboxMessage)
 	if err != nil {
 		t.Fatalf("expected no error enqueing second message, got %v", err)
 	}
-	//As we haven't signaled procChan, the messages are still in the outbox
 
-	err = outbox.Enqueue(outboxTestMessage())
+	//As we haven't signaled procChan, the messages are still in the outbox
+	err = testOutbox.Enqueue(testOutboxMessage)
 	if err == nil {
 		t.Fatalf("expected error enqueing third message, instead got nil")
 	}
 	procChan <- struct{}{}
 	//There should be a slot again in the outbox
 	select {
-	case <-outbox.slots:
+	case <-testOutbox.slots:
 	case <-time.After(2 * time.Second):
 		t.Fatalf("timeout waiting for a free slot")
 	}
 }
 
-func outboxTestMessage() *outboxMsg {
-	return &outboxMsg{
-		msg:       nil,
-		startedAt: time.Time{},
-	}
-}
+var testOutboxMessage = NewOutboxMessage(&message.Message{
+	To:      nil,
+	Flags:   message.Flags{},
+	Expire:  0,
+	Topic:   message.Topic{},
+	Payload: nil,
+})

pss/pss.go

@@ -540,8 +540,8 @@ func (p *Pss) isSelfPossibleRecipient(msg *message.Message, prox bool) bool {
 func (p *Pss) enqueue(msg *message.Message) error {
 	defer metrics.GetOrRegisterResettingTimer("pss.enqueue", nil).UpdateSince(time.Now())
 
-	outboxmsg := outbox.NewOutboxMessage(msg)
-	return p.outbox.Enqueue(outboxmsg)
+	outboxMsg := outbox.NewOutboxMessage(msg)
+	return p.outbox.Enqueue(outboxMsg)
 }
 
 // Send a raw message (any encryption is responsibility of calling client)

pss/pss_test.go

@@ -278,7 +278,7 @@ func TestAddressMatchProx(t *testing.T) {
 	}
 	// enqueue method now is blocking, so we need always somebody processing the outbox
 	mockForwardFunc :=
-		func(msg interface{}) error {
+		func(msg *message.Message) error {
 			return nil
 		}