pss/outbox: Removed outbox.msg. Changed how outbox stops

Author: aperonam

pss/outbox/outbox.go

@@ -17,6 +17,7 @@ type Outbox struct {
 	queue       []*outboxMsg
 	slots       chan int
 	process     chan int
+	stopC       chan struct{}
 }
 
 type forwardFunction func(msg *message.Message) error
@@ -29,6 +30,7 @@ func NewOutbox(config *Config) *Outbox {
 		queue:       make([]*outboxMsg, config.NumberSlots),
 		slots:       make(chan int, config.NumberSlots),
 		process:     make(chan int),
+		stopC:       make(chan struct{}),
 	}
 	// fill up outbox slots
 	for i := 0; i < cap(outbox.slots); i++ {
@@ -44,7 +46,7 @@ func (o *Outbox) Start() {
 
 func (o *Outbox) Stop() {
 	log.Info("Stopping outbox")
-	close(o.process)
+	close(o.stopC)
 }
 
 // Enqueue a new element in the outbox if there is any slot available.
@@ -56,7 +58,10 @@ 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
-		o.process <- slot
+		select {
+		case <-o.stopC:
+		case o.process <- slot:
+		}
 		return nil
 	default:
 		metrics.GetOrRegisterCounter("pss.enqueue.outbox.full", nil).Inc(1)
@@ -68,11 +73,10 @@ func (o *Outbox) Enqueue(outboxMsg *outboxMsg) error {
 func (o *Outbox) processOutbox() {
 	for slot := range o.process {
 		go func(slot int) {
-			msg := o.msg(slot)
+			msg := o.queue[slot]
 			metrics.GetOrRegisterResettingTimer("pss.handle.outbox", nil).UpdateSince(msg.startedAt)
 			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())
 				// requeue the message for processing
 				o.requeue(slot)
@@ -86,18 +90,20 @@ func (o *Outbox) processOutbox() {
 	}
 }
 
-func (o *Outbox) msg(slot int) *outboxMsg {
-	return o.queue[slot]
-}
-
 func (o *Outbox) free(slot int) {
-	o.slots <- slot
+	select {
+	case <-o.stopC:
+	case o.slots <- slot:
+	}
+
 }
 
 func (o *Outbox) requeue(slot int) {
-	o.process <- slot
+	select {
+	case <-o.stopC:
+	case o.process <- slot:
+	}
 }
-
 func (o *Outbox) len() int {
 	return cap(o.slots) - len(o.slots)
 }

pss/outbox/outbox_test.go

@@ -59,7 +59,7 @@ func TestOutbox(tx *testing.T) {
 	<-failedC
 
 	failedMessages := len(failed)
-	t.Assert(failedMessages == 1, "incorrect number of failed messages, expected 1 got %v", failedMessages)
+	t.Equals(failedMessages, 1)
 
 	// The message will be retried once we send to continueC, so first, we change the forward function
 	forwardFail = false

pss/outbox/outbox_whitebox_test.go

@@ -20,7 +20,7 @@ func TestFullOutbox(tx *testing.T) {
 		return nil
 	}
 
-	testOutbox := NewOutbox(&Config{
+	testOutbox := NewMock(&Config{
 		NumberSlots: outboxCapacity,
 		Forward:     successForward,
 	})