Reorganize packages, closes #79

- All packages go into either cmd or pkg directories
- Most packages renamed
- Server packages moved into pkg/server
- sanitize moved into webui, as that's the only place it's used
- filestore moved into pkg/storage/file
- Makefile updated, and PKG variable use fixed

pkg/msghub/hub_test.go

@@ -0,0 +1,255 @@
+package msghub
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+)
+
+// testListener implements the Listener interface, mock for unit tests
+type testListener struct {
+	messages     []*Message // received messages
+	wantMessages int        // how many messages this listener wants to receive
+	errorAfter   int        // when != 0, messages until Receive() begins returning error
+
+	done     chan struct{} // closed once we have received wantMessages
+	overflow chan struct{} // closed if we receive wantMessages+1
+}
+
+func newTestListener(want int) *testListener {
+	l := &testListener{
+		messages:     make([]*Message, 0, want*2),
+		wantMessages: want,
+		done:         make(chan struct{}),
+		overflow:     make(chan struct{}),
+	}
+	if want == 0 {
+		close(l.done)
+	}
+	return l
+}
+
+// Receive a Message, store it in the messages slice, close applicable channels, and return an error
+// if instructed
+func (l *testListener) Receive(msg Message) error {
+	l.messages = append(l.messages, &msg)
+	if len(l.messages) == l.wantMessages {
+		close(l.done)
+	}
+	if len(l.messages) == l.wantMessages+1 {
+		close(l.overflow)
+	}
+	if l.errorAfter > 0 && len(l.messages) > l.errorAfter {
+		return fmt.Errorf("Too many messages")
+	}
+	return nil
+}
+
+// String formats the got vs wanted message counts
+func (l *testListener) String() string {
+	return fmt.Sprintf("got %v messages, wanted %v", len(l.messages), l.wantMessages)
+}
+
+func TestHubNew(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	hub := New(ctx, 5)
+	if hub == nil {
+		t.Fatal("New() == nil, expected a new Hub")
+	}
+}
+
+func TestHubZeroLen(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	hub := New(ctx, 0)
+	m := Message{}
+	for i := 0; i < 100; i++ {
+		hub.Dispatch(m)
+	}
+	// Just making sure Hub doesn't panic
+}
+
+func TestHubZeroListeners(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	hub := New(ctx, 5)
+	m := Message{}
+	for i := 0; i < 100; i++ {
+		hub.Dispatch(m)
+	}
+	// Just making sure Hub doesn't panic
+}
+
+func TestHubOneListener(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	hub := New(ctx, 5)
+	m := Message{}
+	l := newTestListener(1)
+
+	hub.AddListener(l)
+	hub.Dispatch(m)
+
+	// Wait for messages
+	select {
+	case <-l.done:
+	case <-time.After(time.Second):
+		t.Error("Timeout:", l)
+	}
+}
+
+func TestHubRemoveListener(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	hub := New(ctx, 5)
+	m := Message{}
+	l := newTestListener(1)
+
+	hub.AddListener(l)
+	hub.Dispatch(m)
+	hub.RemoveListener(l)
+	hub.Dispatch(m)
+	hub.Sync()
+
+	// Wait for messages
+	select {
+	case <-l.overflow:
+		t.Error(l)
+	case <-time.After(50 * time.Millisecond):
+		// Expected result, no overflow
+	}
+}
+
+func TestHubRemoveListenerOnError(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	hub := New(ctx, 5)
+	m := Message{}
+
+	// error after 1 means listener should receive 2 messages before being removed
+	l := newTestListener(2)
+	l.errorAfter = 1
+
+	hub.AddListener(l)
+	hub.Dispatch(m)
+	hub.Dispatch(m)
+	hub.Dispatch(m)
+	hub.Dispatch(m)
+	hub.Sync()
+
+	// Wait for messages
+	select {
+	case <-l.overflow:
+		t.Error(l)
+	case <-time.After(50 * time.Millisecond):
+		// Expected result, no overflow
+	}
+}
+
+func TestHubHistoryReplay(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	hub := New(ctx, 100)
+	l1 := newTestListener(3)
+	hub.AddListener(l1)
+
+	// Broadcast 3 messages with no listeners
+	msgs := make([]Message, 3)
+	for i := 0; i < len(msgs); i++ {
+		msgs[i] = Message{
+			Subject: fmt.Sprintf("subj %v", i),
+		}
+		hub.Dispatch(msgs[i])
+	}
+
+	// Wait for messages (live)
+	select {
+	case <-l1.done:
+	case <-time.After(time.Second):
+		t.Fatal("Timeout:", l1)
+	}
+
+	// Add a new listener
+	l2 := newTestListener(3)
+	hub.AddListener(l2)
+
+	// Wait for messages (history)
+	select {
+	case <-l2.done:
+	case <-time.After(time.Second):
+		t.Fatal("Timeout:", l2)
+	}
+
+	for i := 0; i < len(msgs); i++ {
+		got := l2.messages[i].Subject
+		want := msgs[i].Subject
+		if got != want {
+			t.Errorf("msg[%v].Subject == %q, want %q", i, got, want)
+		}
+	}
+}
+
+func TestHubHistoryReplayWrap(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	hub := New(ctx, 5)
+	l1 := newTestListener(20)
+	hub.AddListener(l1)
+
+	// Broadcast more messages than the hub can hold
+	msgs := make([]Message, 20)
+	for i := 0; i < len(msgs); i++ {
+		msgs[i] = Message{
+			Subject: fmt.Sprintf("subj %v", i),
+		}
+		hub.Dispatch(msgs[i])
+	}
+
+	// Wait for messages (live)
+	select {
+	case <-l1.done:
+	case <-time.After(time.Second):
+		t.Fatal("Timeout:", l1)
+	}
+
+	// Add a new listener
+	l2 := newTestListener(5)
+	hub.AddListener(l2)
+
+	// Wait for messages (history)
+	select {
+	case <-l2.done:
+	case <-time.After(time.Second):
+		t.Fatal("Timeout:", l2)
+	}
+
+	for i := 0; i < 5; i++ {
+		got := l2.messages[i].Subject
+		want := msgs[i+15].Subject
+		if got != want {
+			t.Errorf("msg[%v].Subject == %q, want %q", i, got, want)
+		}
+	}
+}
+
+func TestHubContextCancel(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	hub := New(ctx, 5)
+	m := Message{}
+	l := newTestListener(1)
+
+	hub.AddListener(l)
+	hub.Dispatch(m)
+	hub.Sync()
+	cancel()
+
+	// Wait for messages
+	select {
+	case <-l.overflow:
+		t.Error(l)
+	case <-time.After(50 * time.Millisecond):
+		// Expected result, no overflow
+	}
+}