storage/mem: implement message cap for #88

- Move message cap tests into storage test suite.
- Update change log.

pkg/storage/file/fstore_test.go

@@ -22,8 +22,8 @@ import (
 
 // TestSuite runs storage package test suite on file store.
 func TestSuite(t *testing.T) {
-	test.StoreSuite(t, func() (storage.Store, func(), error) {
-		ds, _ := setupDataStore(config.Storage{})
+	test.StoreSuite(t, func(conf config.Storage) (storage.Store, func(), error) {
+		ds, _ := setupDataStore(conf)
 		destroy := func() {
 			teardownDataStore(ds)
 		}
@@ -144,78 +144,6 @@ func TestFSMissing(t *testing.T) {
 	}
 }
 
-// Test delivering several messages to the same mailbox, see if message cap works
-func TestFSMessageCap(t *testing.T) {
-	mbCap := 10
-	ds, logbuf := setupDataStore(config.Storage{MailboxMsgCap: mbCap})
-	defer teardownDataStore(ds)
-
-	mbName := "captain"
-	for i := 0; i < 20; i++ {
-		// Add a message
-		subj := fmt.Sprintf("subject %v", i)
-		deliverMessage(ds, mbName, subj, time.Now())
-		t.Logf("Delivered %q", subj)
-
-		// Check number of messages
-		msgs, err := ds.GetMessages(mbName)
-		if err != nil {
-			t.Fatalf("Failed to GetMessages for %q: %v", mbName, err)
-		}
-		if len(msgs) > mbCap {
-			t.Errorf("Mailbox should be capped at %v messages, but has %v", mbCap, len(msgs))
-		}
-
-		// Check that the first message is correct
-		first := i - mbCap + 1
-		if first < 0 {
-			first = 0
-		}
-		firstSubj := fmt.Sprintf("subject %v", first)
-		if firstSubj != msgs[0].Subject() {
-			t.Errorf("Expected first subject to be %q, got %q", firstSubj, msgs[0].Subject())
-		}
-	}
-
-	if t.Failed() {
-		// Wait for handler to finish logging
-		time.Sleep(2 * time.Second)
-		// Dump buffered log data if there was a failure
-		_, _ = io.Copy(os.Stderr, logbuf)
-	}
-}
-
-// Test delivering several messages to the same mailbox, see if no message cap works
-func TestFSNoMessageCap(t *testing.T) {
-	mbCap := 0
-	ds, logbuf := setupDataStore(config.Storage{MailboxMsgCap: mbCap})
-	defer teardownDataStore(ds)
-
-	mbName := "captain"
-	for i := 0; i < 20; i++ {
-		// Add a message
-		subj := fmt.Sprintf("subject %v", i)
-		deliverMessage(ds, mbName, subj, time.Now())
-		t.Logf("Delivered %q", subj)
-
-		// Check number of messages
-		msgs, err := ds.GetMessages(mbName)
-		if err != nil {
-			t.Fatalf("Failed to GetMessages for %q: %v", mbName, err)
-		}
-		if len(msgs) != i+1 {
-			t.Errorf("Expected %v messages, got %v", i+1, len(msgs))
-		}
-	}
-
-	if t.Failed() {
-		// Wait for handler to finish logging
-		time.Sleep(2 * time.Second)
-		// Dump buffered log data if there was a failure
-		_, _ = io.Copy(os.Stderr, logbuf)
-	}
-}
-
 // Test Get the latest message
 func TestGetLatestMessage(t *testing.T) {
 	ds, logbuf := setupDataStore(config.Storage{})

pkg/storage/mem/store.go

@@ -14,12 +14,14 @@ import (
 type Store struct {
 	sync.Mutex
 	boxes map[string]*mbox
+	cap   int
 }
 
 type mbox struct {
 	sync.RWMutex
 	name     string
-	counter  int
+	last     int
+	first    int
 	messages map[string]*Message
 }
 
@@ -29,6 +31,7 @@ var _ storage.Store = &Store{}
 func New(cfg config.Storage) (storage.Store, error) {
 	return &Store{
 		boxes: make(map[string]*mbox),
+		cap:   cfg.MailboxMsgCap,
 	}, nil
 }
 
@@ -46,10 +49,10 @@ func (s *Store) AddMessage(message storage.Message) (id string, err error) {
 			return
 		}
 		// Generate message ID.
-		mb.counter++
-		id = strconv.Itoa(mb.counter)
+		mb.last++
+		id = strconv.Itoa(mb.last)
 		m := &Message{
-			index:   mb.counter,
+			index:   mb.last,
 			mailbox: message.Mailbox(),
 			id:      id,
 			from:    message.From(),
@@ -59,6 +62,13 @@ func (s *Store) AddMessage(message storage.Message) (id string, err error) {
 			source:  source,
 		}
 		mb.messages[id] = m
+		if s.cap > 0 {
+			// Enforce cap.
+			for len(mb.messages) > s.cap {
+				delete(mb.messages, strconv.Itoa(mb.first))
+				mb.first++
+			}
+		}
 	})
 	return id, err
 }
@@ -121,7 +131,7 @@ func (s *Store) VisitMailboxes(f func([]storage.Message) (cont bool)) error {
 }
 
 // withMailbox gets or creates a mailbox, locks it, then calls f.
-func (s *Store) withMailbox(mailbox string, rw bool, f func(mb *mbox)) {
+func (s *Store) withMailbox(mailbox string, writeLock bool, f func(mb *mbox)) {
 	s.Lock()
 	mb, ok := s.boxes[mailbox]
 	if !ok {
@@ -133,13 +143,13 @@ func (s *Store) withMailbox(mailbox string, rw bool, f func(mb *mbox)) {
 		s.boxes[mailbox] = mb
 	}
 	s.Unlock()
-	if rw {
+	if writeLock {
 		mb.Lock()
 	} else {
 		mb.RLock()
 	}
 	defer func() {
-		if rw {
+		if writeLock {
 			mb.Unlock()
 		} else {
 			mb.RUnlock()

pkg/storage/mem/store_test.go

@@ -10,8 +10,8 @@ import (
 
 // TestSuite runs storage package test suite on file store.
 func TestSuite(t *testing.T) {
-	test.StoreSuite(t, func() (storage.Store, func(), error) {
-		s, _ := New(config.Storage{})
+	test.StoreSuite(t, func(conf config.Storage) (storage.Store, func(), error) {
+		s, _ := New(conf)
 		destroy := func() {}
 		return s, destroy, nil
 	})