Repository: looplab/eventhorizon
Pull request #363 (2021-11-17 10:18:26)

Add PreRegisterNamespace()
### Description

Adds the PreRegisterNamespace() method to the namespace Outbox and EventStore.

This method is useful to pre-populate with all namespaces known upfront at service start.

### Affected Components

- Namespace, Outbox and EventStore

### Related Issues

### Solution and Design

### Steps to test and verify



Patch:
diff --git a/namespace/eventstore.go b/namespace/eventstore.go
index c5dd283..4580528 100644
--- a/namespace/eventstore.go
+++ b/namespace/eventstore.go
@@ -61,6 +61,18 @@ func NewEventStore(factory func(ns string) (eh.EventStore, error)) *EventStore {
 	}
 }
 
+// PreRegisterNamespace will make sure that a namespace exists in the eventstore.
+// In normal cases the eventstore for a namespace is created when an event for
+// that namespace is first seen.
+func (s *EventStore) PreRegisterNamespace(ns string) error {
+	ctx := NewContext(context.Background(), ns)
+
+	// This creates the eventstore for a namespace in case id did not exist.
+	_, err := s.eventStore(ctx)
+
+	return err
+}
+
 // Save implements the Save method of the eventhorizon.EventStore interface.
 func (s *EventStore) Save(ctx context.Context, events []eh.Event, originalVersion int) error {
 	store, err := s.eventStore(ctx)
diff --git a/namespace/outbox.go b/namespace/outbox.go
index d861ac5..352e9cc 100644
--- a/namespace/outbox.go
+++ b/namespace/outbox.go
@@ -115,6 +115,18 @@ func (o *Outbox) AddHandler(ctx context.Context, m eh.EventMatcher, h eh.EventHa
 	return nil
 }
 
+// PreRegisterNamespace will make sure that a namespace exists in the outbox
+// and that processing of that namespace is active. In normal cases the outbox
+// for a namespace is started when an event for that namespace is first seen.
+func (o *Outbox) PreRegisterNamespace(ns string) error {
+	ctx := NewContext(context.Background(), ns)
+
+	// This starts the outbox for a namespace in case it was not already started.
+	_, err := o.outbox(ctx)
+
+	return err
+}
+
 // HandleEvent implements the HandleEvent method of the eventhorizon.Outbox interface.
 func (o *Outbox) HandleEvent(ctx context.Context, event eh.Event) error {
 	ob, err := o.outbox(ctx)


Test patch:
diff --git a/eventstore/memory/eventstore_test.go b/eventstore/memory/eventstore_test.go
index d8a3126..0822cbe 100644
--- a/eventstore/memory/eventstore_test.go
+++ b/eventstore/memory/eventstore_test.go
@@ -36,6 +36,10 @@ func TestEventStore(t *testing.T) {
 	}
 
 	eventstore.AcceptanceTest(t, store, context.Background())
+
+	if err := store.Close(); err != nil {
+		t.Error("there should be no error:", err)
+	}
 }
 
 func TestWithEventHandler(t *testing.T) {
diff --git a/eventstore/mongodb/eventstore_test.go b/eventstore/mongodb/eventstore_test.go
index c73e1e8..64c5ed6 100644
--- a/eventstore/mongodb/eventstore_test.go
+++ b/eventstore/mongodb/eventstore_test.go
@@ -60,9 +60,11 @@ func TestEventStoreIntegration(t *testing.T) {
 		t.Fatal("there should be a store")
 	}
 
-	defer store.Close()
-
 	eventstore.AcceptanceTest(t, store, context.Background())
+
+	if err := store.Close(); err != nil {
+		t.Error("there should be no error:", err)
+	}
 }
 
 func TestWithCollectionNameIntegration(t *testing.T) {
diff --git a/eventstore/mongodb_v2/eventstore_test.go b/eventstore/mongodb_v2/eventstore_test.go
index f739cde..a519b0a 100644
--- a/eventstore/mongodb_v2/eventstore_test.go
+++ b/eventstore/mongodb_v2/eventstore_test.go
@@ -60,9 +60,11 @@ func TestEventStoreIntegration(t *testing.T) {
 		t.Fatal("there should be a store")
 	}
 
-	defer store.Close()
-
 	eventstore.AcceptanceTest(t, store, context.Background())
+
+	if err := store.Close(); err != nil {
+		t.Error("there should be no error:", err)
+	}
 }
 
 func TestWithCollectionNamesIntegration(t *testing.T) {
diff --git a/eventstore/recorder/eventstore_test.go b/eventstore/recorder/eventstore_test.go
index bf1e798..7f1de84 100644
--- a/eventstore/recorder/eventstore_test.go
+++ b/eventstore/recorder/eventstore_test.go
@@ -100,4 +100,8 @@ func TestEventStore(t *testing.T) {
 			t.Error("the event version should be correct:", event, event.Version())
 		}
 	}
+
+	if err := store.Close(); err != nil {
+		t.Error("there should be no error:", err)
+	}
 }
diff --git a/namespace/eventstore_test.go b/namespace/eventstore_test.go
index d4b497f..6630a99 100644
--- a/namespace/eventstore_test.go
+++ b/namespace/eventstore_test.go
@@ -16,6 +16,7 @@ package namespace
 
 import (
 	"context"
+	"sync"
 	"testing"
 
 	eh "github.com/looplab/eventhorizon"
@@ -27,6 +28,10 @@ import (
 func TestEventStore(t *testing.T) {
 	usedNamespaces := map[string]struct{}{}
 
+	var storeCreated sync.WaitGroup
+
+	storeCreated.Add(2)
+
 	store := NewEventStore(func(ns string) (eh.EventStore, error) {
 		usedNamespaces[ns] = struct{}{}
 		s, err := memory.NewEventStore()
@@ -34,19 +39,25 @@ func TestEventStore(t *testing.T) {
 			return nil, err
 		}
 
+		storeCreated.Done()
+
 		return s, nil
 	})
 	if store == nil {
 		t.Fatal("there should be a store")
 	}
 
-	t.Log("testing default namespace")
-	eventstore.AcceptanceTest(t, store, context.Background())
+	if err := store.PreRegisterNamespace(DefaultNamespace); err != nil {
+		t.Error("there should be no error:", err)
+	}
 
-	ctx := NewContext(context.Background(), "other")
+	ns := "other"
+	if err := store.PreRegisterNamespace(ns); err != nil {
+		t.Error("there should be no error:", err)
+	}
 
-	t.Log("testing other namespace")
-	eventstore.AcceptanceTest(t, store, ctx)
+	// Check that both event stores has been created.
+	storeCreated.Wait()
 
 	if _, ok := usedNamespaces["default"]; !ok {
 		t.Error("the default namespace should have been used")
@@ -55,4 +66,16 @@ func TestEventStore(t *testing.T) {
 	if _, ok := usedNamespaces["other"]; !ok {
 		t.Error("the other namespace should have been used")
 	}
+
+	t.Log("testing default namespace")
+	eventstore.AcceptanceTest(t, store, context.Background())
+
+	ctx := NewContext(context.Background(), ns)
+
+	t.Log("testing other namespace")
+	eventstore.AcceptanceTest(t, store, ctx)
+
+	if err := store.Close(); err != nil {
+		t.Error("there should be no error:", err)
+	}
 }
diff --git a/namespace/outbox_test.go b/namespace/outbox_test.go
index e5a743b..a6f546a 100644
--- a/namespace/outbox_test.go
+++ b/namespace/outbox_test.go
@@ -3,6 +3,7 @@ package namespace
 import (
 	"context"
 	"math/rand"
+	"sync"
 	"testing"
 	"time"
 
@@ -23,6 +24,10 @@ func TestOutbox(t *testing.T) {
 	memory.PeriodicSweepInterval = 2 * time.Second
 	memory.PeriodicSweepAge = 2 * time.Second
 
+	var outboxCreated sync.WaitGroup
+
+	outboxCreated.Add(2)
+
 	o := NewOutbox(func(ns string) (eh.Outbox, error) {
 		usedNamespaces[ns] = struct{}{}
 		o, err := memory.NewOutbox()
@@ -30,6 +35,8 @@ func TestOutbox(t *testing.T) {
 			return nil, err
 		}
 
+		outboxCreated.Done()
+
 		return o, nil
 	})
 	if o == nil {
@@ -43,10 +50,29 @@ func TestOutbox(t *testing.T) {
 		t.Fatal("there should be no error:", err)
 	}
 
+	if err := o.PreRegisterNamespace(DefaultNamespace); err != nil {
+		t.Error("there should be no error:", err)
+	}
+
+	ns := "other"
+	if err := o.PreRegisterNamespace(ns); err != nil {
+		t.Error("there should be no error:", err)
+	}
+
+	// Check that both outboxes have been created.
+	outboxCreated.Wait()
+
+	if _, ok := usedNamespaces[DefaultNamespace]; !ok {
+		t.Error("the default namespace should have been used")
+	}
+
+	if _, ok := usedNamespaces[ns]; !ok {
+		t.Error("the other namespace should have been used")
+	}
+
 	t.Log("testing default namespace")
 	outbox.AcceptanceTest(t, o, context.Background(), DefaultNamespace)
 
-	ns := "other"
 	ctx := NewContext(context.Background(), ns)
 
 	t.Log("testing other namespace")
@@ -64,14 +90,6 @@ func TestOutbox(t *testing.T) {
 
 	handlerAddedBefore.Unlock()
 
-	if _, ok := usedNamespaces[DefaultNamespace]; !ok {
-		t.Error("the default namespace should have been used")
-	}
-
-	if _, ok := usedNamespaces[ns]; !ok {
-		t.Error("the other namespace should have been used")
-	}
-
 	if err := o.Close(); err != nil {
 		t.Error("there should be no error:", err)
 	}
diff --git a/namespace/repo_test.go b/namespace/repo_test.go
index d0c141c..51b84a5 100644
--- a/namespace/repo_test.go
+++ b/namespace/repo_test.go
@@ -73,4 +73,8 @@ func TestReadRepo(t *testing.T) {
 
 	t.Log("testing other namespace")
 	repo.AcceptanceTest(t, r, otherCtx)
+
+	if err := r.Close(); err != nil {
+		t.Error("there should be no error:", err)
+	}
 }
diff --git a/repo/cache/repo_test.go b/repo/cache/repo_test.go
index f49c6fc..399979d 100644
--- a/repo/cache/repo_test.go
+++ b/repo/cache/repo_test.go
@@ -41,6 +41,10 @@ func TestReadRepo(t *testing.T) {
 	}
 
 	repo.AcceptanceTest(t, r, context.Background())
+
+	if err := r.Close(); err != nil {
+		t.Error("there should be no error:", err)
+	}
 }
 
 func TestIntoRepo(t *testing.T) {
diff --git a/repo/memory/repo_test.go b/repo/memory/repo_test.go
index 3650dc2..aa295a0 100644
--- a/repo/memory/repo_test.go
+++ b/repo/memory/repo_test.go
@@ -39,6 +39,10 @@ func TestReadRepo(t *testing.T) {
 	}
 
 	repo.AcceptanceTest(t, r, context.Background())
+
+	if err := r.Close(); err != nil {
+		t.Error("there should be no error:", err)
+	}
 }
 
 func TestIntoRepo(t *testing.T) {
diff --git a/repo/mongodb/repo_test.go b/repo/mongodb/repo_test.go
index 7bd88ba..aec7555 100644
--- a/repo/mongodb/repo_test.go
+++ b/repo/mongodb/repo_test.go
@@ -77,6 +77,10 @@ func TestReadRepoIntegration(t *testing.T) {
 
 	repo.AcceptanceTest(t, r, context.Background())
 	extraRepoTests(t, r)
+
+	if err := r.Close(); err != nil {
+		t.Error("there should be no error:", err)
+	}
 }
 
 func extraRepoTests(t *testing.T, r *Repo) {
diff --git a/repo/tracing/repo_test.go b/repo/tracing/repo_test.go
index 4319f5b..8042248 100644
--- a/repo/tracing/repo_test.go
+++ b/repo/tracing/repo_test.go
@@ -41,6 +41,10 @@ func TestReadRepo(t *testing.T) {
 	}
 
 	repo.AcceptanceTest(t, r, context.Background())
+
+	if err := r.Close(); err != nil {
+		t.Error("there should be no error:", err)
+	}
 }
 
 func TestIntoRepo(t *testing.T) {
diff --git a/repo/version/repo_test.go b/repo/version/repo_test.go
index 5729d1a..7db81b4 100644
--- a/repo/version/repo_test.go
+++ b/repo/version/repo_test.go
@@ -46,6 +46,10 @@ func TestReadRepo(t *testing.T) {
 
 	repo.AcceptanceTest(t, r, context.Background())
 	extraRepoTests(t, r, baseRepo)
+
+	if err := r.Close(); err != nil {
+		t.Error("there should be no error:", err)
+	}
 }
 
 func extraRepoTests(t *testing.T, r *Repo, baseRepo *memory.Repo) {