Repository: looplab/eventhorizon
Pull request #335 (2021-09-07 10:04:04)

254 / Outbox pattern
### Description

Outbox pattern for handling projections and publishing on the bus with at least once guarantees.

### Affected Components

- Event Bus (MongoDB)
- Outbox, new component

### Related Issues

#254 

### Solution and Design

The outbox is implemented as a separate collection with outbox records. Each record corresponds to one event with all the required handlers marked on it. The outbox collection is being watched and each record is processed, currently each host process only its own records to keep data processing local (command -> event -> projection happening in the same service). After processing, successful handlers are removed from record, and when all handlers are processed the full record is removed. In case of an error the event is retried (currently indefinitely) by periodic sweeps of the outbox; more often for events on the local host, and a bit later for events from all hosts (in case a host is removed).

### Steps to test and verify


coveralls: 
[![Coverage Status](https://coveralls.io/builds/42694433/badge)](https://coveralls.io/builds/42694433)

Coverage decreased (-0.2%) to 69.233% when pulling **3b0b559c0adaf3b80f8ac4cdba213cb00e573375 on maxekman:feature/mongodb-outbox-2** into **e82556fcd0b5c8697599bd099801e061082cf7cb on looplab:main**.

maxekman: @hansbogert Ready for another review, I think all comments are addressed. There is no real performance benefit with using the "watch group" feature, at least not in synthetic benchmarks. I'll still keep the option for the use case of keeping some outbox processing local to logical groups in a larger system (could be seen as some kind of load balancing/direction).
hansbogert: Nice solutions to the things that we're pointed out. 

LGTM

Patch:
diff --git a/Makefile b/Makefile
index 6ee0a85..ec6dcd9 100644
--- a/Makefile
+++ b/Makefile
@@ -75,4 +75,4 @@ clean:
 
 .PHONY: mongodb_shell
 mongodb_shell:
-	docker run -it --network eventhorizon_default --rm mongo:4.4 mongo --host mongodb-rs test
+	docker run -it --network eventhorizon_default --rm mongo:4.4 mongo --host mongodb test
diff --git a/docker-compose.yml b/docker-compose.yml
index 8bdbc63..c4020d7 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -9,41 +9,37 @@ services:
       context: .
       dockerfile: Dockerfile.test
     depends_on:
-      - mongodb
+      - mongodb-docker
       - gpubsub
       - kafka
       - redis
       - nats
     environment:
-      MONGODB_ADDR: mongodb:27017
+      MONGODB_ADDR: mongodb-docker:27017
       PUBSUB_EMULATOR_HOST: gpubsub:8793
       KAFKA_ADDR: kafka:9092
       REDIS_ADDR: redis:6379
       NATS_ADDR: nats:4222
     command: [-c, make test test_integration]
 
-  # MongoDB replica set setup.
-  mongodb:
-    image: mongo:4.4
-    restart: on-failure
-    depends_on:
-      - mongodb-rs
-    networks:
-      default:
-    entrypoint: [bash, /scripts/mongodb-rs-setup.sh]
+  mongodb-docker:
+    image: bitnami/mongodb:4.4
+    ports:
+      - 27017:27017
+    environment:
+      MONGODB_REPLICA_SET_MODE: primary
+      ALLOW_EMPTY_PASSWORD: "yes"
     volumes:
-      - ./scripts/mongodb-rs-setup.sh:/scripts/mongodb-rs-setup.sh
+      - mongodb:/data/db
 
-  mongodb-rs:
-    image: mongo:4.4
-    restart: always
+  mongodb:
+    image: bitnami/mongodb:4.4
     ports:
       - 27017:27017
-    command: [--replSet, rs0, --bind_ip_all]
-    healthcheck:
-      test: test $$(echo "rs.status().ok" | mongo --quiet) -eq 1
-      interval: 10s
-      start_period: 30s
+    environment:
+      MONGODB_REPLICA_SET_MODE: primary
+      MONGODB_ADVERTISED_HOSTNAME: localhost # Needed to connect from localhost.
+      ALLOW_EMPTY_PASSWORD: "yes"
     volumes:
       - mongodb:/data/db
 
diff --git a/eventbus/gcp/eventbus.go b/eventbus/gcp/eventbus.go
index 8174e8a..17a0b8b 100644
--- a/eventbus/gcp/eventbus.go
+++ b/eventbus/gcp/eventbus.go
@@ -208,6 +208,8 @@ func (b *EventBus) AddHandler(ctx context.Context, m eh.EventMatcher, h eh.Event
 	// Register handler.
 	b.registered[h.HandlerType()] = struct{}{}
 
+	b.wg.Add(1)
+
 	// Handle until context is cancelled.
 	go b.handle(m, h, sub)
 
@@ -233,7 +235,6 @@ func (b *EventBus) Close() error {
 
 // Handles all events coming in on the channel.
 func (b *EventBus) handle(m eh.EventMatcher, h eh.EventHandler, sub *pubsub.Subscription) {
-	b.wg.Add(1)
 	defer b.wg.Done()
 
 	for {
diff --git a/eventbus/kafka/eventbus.go b/eventbus/kafka/eventbus.go
index b5d53aa..402b333 100644
--- a/eventbus/kafka/eventbus.go
+++ b/eventbus/kafka/eventbus.go
@@ -234,6 +234,8 @@ func (b *EventBus) AddHandler(ctx context.Context, m eh.EventMatcher, h eh.Event
 	// Register handler.
 	b.registered[h.HandlerType()] = struct{}{}
 
+	b.wg.Add(1)
+
 	// Handle until context is cancelled.
 	go b.handle(m, h, r)
 
@@ -255,7 +257,6 @@ func (b *EventBus) Close() error {
 
 // Handles all events coming in on the channel.
 func (b *EventBus) handle(m eh.EventMatcher, h eh.EventHandler, r *kafka.Reader) {
-	b.wg.Add(1)
 	defer b.wg.Done()
 
 	handler := b.handler(m, h, r)
diff --git a/eventbus/local/eventbus.go b/eventbus/local/eventbus.go
index 4f61aae..190234b 100644
--- a/eventbus/local/eventbus.go
+++ b/eventbus/local/eventbus.go
@@ -123,6 +123,8 @@ func (b *EventBus) AddHandler(ctx context.Context, m eh.EventMatcher, h eh.Event
 	// Register handler.
 	b.registered[h.HandlerType()] = struct{}{}
 
+	b.wg.Add(1)
+
 	// Handle until context is cancelled.
 	go b.handle(m, h, ch)
 
@@ -150,7 +152,6 @@ type evt struct {
 
 // Handles all events coming in on the channel.
 func (b *EventBus) handle(m eh.EventMatcher, h eh.EventHandler, ch <-chan []byte) {
-	b.wg.Add(1)
 	defer b.wg.Done()
 
 	for {
diff --git a/eventbus/nats/eventbus.go b/eventbus/nats/eventbus.go
index a6f4ede..7dede32 100644
--- a/eventbus/nats/eventbus.go
+++ b/eventbus/nats/eventbus.go
@@ -178,6 +178,8 @@ func (b *EventBus) AddHandler(ctx context.Context, m eh.EventMatcher, h eh.Event
 	// Register handler.
 	b.registered[h.HandlerType()] = struct{}{}
 
+	b.wg.Add(1)
+
 	// Handle until context is cancelled.
 	go b.handle(sub)
 
@@ -201,7 +203,6 @@ func (b *EventBus) Close() error {
 
 // Handles all events coming in on the channel.
 func (b *EventBus) handle(sub *nats.Subscription) {
-	b.wg.Add(1)
 	defer b.wg.Done()
 
 	for {
diff --git a/eventbus/redis/eventbus.go b/eventbus/redis/eventbus.go
index 08cbe99..5662979 100644
--- a/eventbus/redis/eventbus.go
+++ b/eventbus/redis/eventbus.go
@@ -177,6 +177,8 @@ func (b *EventBus) AddHandler(ctx context.Context, m eh.EventMatcher, h eh.Event
 	// Register handler.
 	b.registered[h.HandlerType()] = struct{}{}
 
+	b.wg.Add(1)
+
 	// Handle until context is cancelled.
 	go b.handle(m, h, groupName)
 
@@ -198,7 +200,6 @@ func (b *EventBus) Close() error {
 
 // Handles all events coming in on the channel.
 func (b *EventBus) handle(m eh.EventMatcher, h eh.EventHandler, groupName string) {
-	b.wg.Add(1)
 	defer b.wg.Done()
 
 	handler := b.handler(m, h, groupName)
diff --git a/eventstore/mongodb/eventstore.go b/eventstore/mongodb/eventstore.go
index 2ec2df6..6cda628 100644
--- a/eventstore/mongodb/eventstore.go
+++ b/eventstore/mongodb/eventstore.go
@@ -37,9 +37,10 @@ import (
 // collection with one document per aggregate/stream which holds its events
 // as values.
 type EventStore struct {
-	client       *mongo.Client
-	aggregates   *mongo.Collection
-	eventHandler eh.EventHandler
+	client                *mongo.Client
+	aggregates            *mongo.Collection
+	eventHandlerAfterSave eh.EventHandler
+	eventHandlerInTX      eh.EventHandler
 }
 
 // NewEventStore creates a new EventStore with a MongoDB URI: `mongodb://hostname`.
@@ -84,11 +85,39 @@ func NewEventStoreWithClient(client *mongo.Client, db string, options ...Option)
 // Option is an option setter used to configure creation.
 type Option func(*EventStore) error
 
-// WithEventHandler adds an event handler that will be called when saving events.
+// WithEventHandler adds an event handler that will be called after saving events.
 // An example would be to add an event bus to publish events.
 func WithEventHandler(h eh.EventHandler) Option {
 	return func(s *EventStore) error {
-		s.eventHandler = h
+		if s.eventHandlerAfterSave != nil {
+			return fmt.Errorf("another event handler is already set")
+		}
+
+		if s.eventHandlerInTX != nil {
+			return fmt.Errorf("another TX event handler is already set")
+		}
+
+		s.eventHandlerAfterSave = h
+
+		return nil
+	}
+}
+
+// WithEventHandlerInTX adds an event handler that will be called during saving of
+// events. An example would be to add an outbox to further process events.
+// For an outbox to be atomic it needs to use the same transaction as the save
+// operation, which is passed down using the context.
+func WithEventHandlerInTX(h eh.EventHandler) Option {
+	return func(s *EventStore) error {
+		if s.eventHandlerAfterSave != nil {
+			return fmt.Errorf("another event handler is already set")
+		}
+
+		if s.eventHandlerInTX != nil {
+			return fmt.Errorf("another TX event handler is already set")
+		}
+
+		s.eventHandlerInTX = h
 
 		return nil
 	}
@@ -161,17 +190,47 @@ func (s *EventStore) Save(ctx context.Context, events []eh.Event, originalVersio
 		dbEvents[i] = *e
 	}
 
-	// Either insert a new aggregate or append to an existing.
-	if originalVersion == 0 {
-		aggregate := aggregateRecord{
-			AggregateID: id,
-			Version:     len(dbEvents),
-			Events:      dbEvents,
+	// Run the operation in a transaction if using an outbox, otherwise it's not needed.
+	saveEvents := func(ctx mongo.SessionContext) error {
+		// Either insert a new aggregate or append to an existing.
+		if originalVersion == 0 {
+			aggregate := aggregateRecord{
+				AggregateID: id,
+				Version:     len(dbEvents),
+				Events:      dbEvents,
+			}
+			if _, err := s.aggregates.InsertOne(ctx, aggregate); err != nil {
+				return fmt.Errorf("could not insert events (new): %w", err)
+			}
+		} else {
+			// Increment aggregate version on insert of new event record, and
+			// only insert if version of aggregate is matching (ie not changed
+			// since loading the aggregate).
+			if r, err := s.aggregates.UpdateOne(ctx,
+				bson.M{
+					"_id":     id,
+					"version": originalVersion,
+				},
+				bson.M{
+					"$push": bson.M{"events": bson.M{"$each": dbEvents}},
+					"$inc":  bson.M{"version": len(dbEvents)},
+				},
+			); err != nil {
+				return fmt.Errorf("could not insert events (update): %w", err)
+			} else if r.MatchedCount == 0 {
+				return eh.ErrEventConflictFromOtherSave
+			}
 		}
 
-		if _, err := s.aggregates.InsertOne(ctx, aggregate); err != nil {
+		return nil
+	}
+
+	// Run the operation in a transaction if using an outbox, otherwise it's not needed.
+	if s.eventHandlerInTX != nil {
+		sess, err := s.client.StartSession(nil)
+		if err != nil {
 			return &eh.EventStoreError{
-				Err:              fmt.Errorf("could not insert: %w", err),
+				Err:              fmt.Errorf("could not start transaction: %w", err),
 				Op:               eh.EventStoreOpSave,
 				AggregateType:    at,
 				AggregateID:      id,
@@ -179,31 +238,36 @@ func (s *EventStore) Save(ctx context.Context, events []eh.Event, originalVersio
 				Events:           events,
 			}
 		}
-	} else {
-		// Increment aggregate version on insert of new event record, and
-		// only insert if version of aggregate is matching (ie not changed
-		// since loading the aggregate).
-		if r, err := s.aggregates.UpdateOne(ctx,
-			bson.M{
-				"_id":     id,
-				"version": originalVersion,
-			},
-			bson.M{
-				"$push": bson.M{"events": bson.M{"$each": dbEvents}},
-				"$inc":  bson.M{"version": len(dbEvents)},
-			},
-		); err != nil {
+
+		defer sess.EndSession(ctx)
+
+		if _, err := sess.WithTransaction(ctx, func(ctx mongo.SessionContext) (interface{}, error) {
+			if err := saveEvents(ctx); err != nil {
+				return nil, err
+			}
+
+			for _, e := range events {
+				if err := s.eventHandlerInTX.HandleEvent(ctx, e); err != nil {
+					return nil, fmt.Errorf("could not handle event in transaction: %w", err)
+				}
+			}
+
+			return nil, nil
+		}); err != nil {
 			return &eh.EventStoreError{
-				Err:              fmt.Errorf("could not update: %w", err),
+				Err:              err,
 				Op:               eh.EventStoreOpSave,
 				AggregateType:    at,
 				AggregateID:      id,
 				AggregateVersion: originalVersion,
 				Events:           events,
 			}
-		} else if r.MatchedCount == 0 {
+		}
+	} else {
+		dummySessionCtx := mongo.NewSessionContext(ctx, nil)
+		if err := saveEvents(dummySessionCtx); err != nil {
 			return &eh.EventStoreError{
-				Err:              eh.ErrEventConflictFromOtherSave,
+				Err:              err,
 				Op:               eh.EventStoreOpSave,
 				AggregateType:    at,
 				AggregateID:      id,
@@ -214,9 +278,9 @@ func (s *EventStore) Save(ctx context.Context, events []eh.Event, originalVersio
 	}
 
 	// Let the optional event handler handle the events.
-	if s.eventHandler != nil {
+	if s.eventHandlerAfterSave != nil {
 		for _, e := range events {
-			if err := s.eventHandler.HandleEvent(ctx, e); err != nil {
+			if err := s.eventHandlerAfterSave.HandleEvent(ctx, e); err != nil {
 				return &eh.EventHandlerError{
 					Err:   err,
 					Event: e,
diff --git a/eventstore/mongodb_v2/eventstore.go b/eventstore/mongodb_v2/eventstore.go
index 976edce..b6c12da 100644
--- a/eventstore/mongodb_v2/eventstore.go
+++ b/eventstore/mongodb_v2/eventstore.go
@@ -37,10 +37,11 @@ import (
 // for all events and another to keep track of all aggregates/streams. It also
 // keep tracks of the global position of events, stored as metadata.
 type EventStore struct {
-	client       *mongo.Client
-	events       *mongo.Collection
-	streams      *mongo.Collection
-	eventHandler eh.EventHandler
+	client                *mongo.Client
+	events                *mongo.Collection
+	streams               *mongo.Collection
+	eventHandlerAfterSave eh.EventHandler
+	eventHandlerInTX      eh.EventHandler
 }
 
 // NewEventStore creates a new EventStore with a MongoDB URI: `mongodb://hostname`.
@@ -109,11 +110,39 @@ func NewEventStoreWithClient(client *mongo.Client, dbName string, options ...Opt
 // Option is an option setter used to configure creation.
 type Option func(*EventStore) error
 
-// WithEventHandler adds an event handler that will be called when saving events.
+// WithEventHandler adds an event handler that will be called after saving events.
 // An example would be to add an event bus to publish events.
 func WithEventHandler(h eh.EventHandler) Option {
 	return func(s *EventStore) error {
-		s.eventHandler = h
+		if s.eventHandlerAfterSave != nil {
+			return fmt.Errorf("another event handler is already set")
+		}
+
+		if s.eventHandlerInTX != nil {
+			return fmt.Errorf("another TX event handler is already set")
+		}
+
+		s.eventHandlerAfterSave = h
+
+		return nil
+	}
+}
+
+// WithEventHandlerInTX adds an event handler that will be called during saving of
+// events. An example would be to add an outbox to further process events.
+// For an outbox to be atomic it needs to use the same transaction as the save
+// operation, which is passed down using the context.
+func WithEventHandlerInTX(h eh.EventHandler) Option {
+	return func(s *EventStore) error {
+		if s.eventHandlerAfterSave != nil {
+			return fmt.Errorf("another event handler is already set")
+		}
+
+		if s.eventHandlerInTX != nil {
+			return fmt.Errorf("another TX event handler is already set")
+		}
+
+		s.eventHandlerInTX = h
 
 		return nil
 	}
@@ -289,6 +318,14 @@ func (s *EventStore) Save(ctx context.Context, events []eh.Event, originalVersio
 			}
 		}
 
+		if s.eventHandlerInTX != nil {
+			for _, e := range events {
+				if err := s.eventHandlerInTX.HandleEvent(ctx, e); err != nil {
+					return nil, fmt.Errorf("could not handle event in transaction: %w", err)
+				}
+			}
+		}
+
 		return nil, nil
 	}); err != nil {
 		return &eh.EventStoreError{
@@ -302,9 +339,9 @@ func (s *EventStore) Save(ctx context.Context, events []eh.Event, originalVersio
 	}
 
 	// Let the optional event handler handle the events.
-	if s.eventHandler != nil {
+	if s.eventHandlerAfterSave != nil {
 		for _, e := range events {
-			if err := s.eventHandler.HandleEvent(ctx, e); err != nil {
+			if err := s.eventHandlerAfterSave.HandleEvent(ctx, e); err != nil {
 				return &eh.EventHandlerError{
 					Err:   err,
 					Event: e,
diff --git a/examples/guestlist/domains/guestlist/projectors.go b/examples/guestlist/domains/guestlist/projectors.go
index e2048fd..24a9485 100644
--- a/examples/guestlist/domains/guestlist/projectors.go
+++ b/examples/guestlist/domains/guestlist/projectors.go
@@ -135,7 +135,7 @@ func NewGuestListProjector(repo eh.ReadWriteRepo, eventID uuid.UUID) *GuestListP
 
 // HandlerType implements the HandlerType method of the eventhorizon.EventHandler interface.
 func (p *GuestListProjector) HandlerType() eh.EventHandlerType {
-	return eh.EventHandlerType("GuestList")
+	return eh.EventHandlerType("projector_GuestList")
 }
 
 // HandleEvent implements the HandleEvent method of the EventHandler interface.
@@ -176,9 +176,8 @@ func (p *GuestListProjector) HandleEvent(ctx context.Context, event eh.Event) er
 
 	case InviteDeniedEvent:
 		g.NumDenied++
-
 	default:
-		return fmt.Errorf("could not handle event: %s", event)
+		return fmt.Errorf("projector: unsupported event type: %s", event)
 	}
 
 	if err := p.repo.Save(ctx, g); err != nil {
diff --git a/examples/guestlist/domains/guestlist/setup.go b/examples/guestlist/domains/guestlist/setup.go
index 2b32b3c..6b33949 100644
--- a/examples/guestlist/domains/guestlist/setup.go
+++ b/examples/guestlist/domains/guestlist/setup.go
@@ -16,7 +16,7 @@ package guestlist
 
 import (
 	"context"
-	"log"
+	"fmt"
 
 	eh "github.com/looplab/eventhorizon"
 	"github.com/looplab/eventhorizon/aggregatestore/events"
@@ -28,60 +28,80 @@ import (
 	"github.com/looplab/eventhorizon/uuid"
 )
 
+type HandlerAdder interface {
+	AddHandler(context.Context, eh.EventMatcher, eh.EventHandler) error
+}
+
 // Setup configures the guestlist.
 func Setup(
 	ctx context.Context,
 	eventStore eh.EventStore,
-	eventBus eh.EventBus,
+	local, global HandlerAdder,
 	commandBus *bus.CommandHandler,
 	invitationRepo, guestListRepo eh.ReadWriteRepo,
-	eventID uuid.UUID) {
-
-	// Add a logger as an observer.
-	eventBus.AddHandler(ctx, eh.MatchAll{},
-		eh.UseEventHandlerMiddleware(&Logger{}, observer.Middleware))
+	eventID uuid.UUID) error {
 
 	// Create the aggregate repository.
 	aggregateStore, err := events.NewAggregateStore(eventStore)
 	if err != nil {
-		log.Fatalf("could not create aggregate store: %s", err)
+		return fmt.Errorf("could not create aggregate store: %w", err)
 	}
 
 	// Create the aggregate command handler and register the commands it handles.
 	invitationHandler, err := aggregate.NewCommandHandler(InvitationAggregateType, aggregateStore)
 	if err != nil {
-		log.Fatalf("could not create command handler: %s", err)
+		return fmt.Errorf("could not create command handler: %w", err)
 	}
 	commandHandler := eh.UseCommandHandlerMiddleware(invitationHandler, LoggingMiddleware)
-	commandBus.SetHandler(commandHandler, CreateInviteCommand)
-	commandBus.SetHandler(commandHandler, AcceptInviteCommand)
-	commandBus.SetHandler(commandHandler, DeclineInviteCommand)
-	commandBus.SetHandler(commandHandler, ConfirmInviteCommand)
-	commandBus.SetHandler(commandHandler, DenyInviteCommand)
+	for _, cmd := range []eh.CommandType{
+		CreateInviteCommand,
+		AcceptInviteCommand,
+		DeclineInviteCommand,
+		ConfirmInviteCommand,
+		DenyInviteCommand,
+	} {
+		if err := commandBus.SetHandler(commandHandler, cmd); err != nil {
+			return fmt.Errorf("could not add command handler for '%s': %w", cmd, err)
+		}
+	}
 
 	// Create and register a read model for individual invitations.
 	invitationProjector := projector.NewEventHandler(
 		NewInvitationProjector(), invitationRepo)
 	invitationProjector.SetEntityFactory(func() eh.Entity { return &Invitation{} })
-	eventBus.AddHandler(ctx, eh.MatchEvents{
+	if err := local.AddHandler(ctx, eh.MatchEvents{
 		InviteCreatedEvent,
 		InviteAcceptedEvent,
 		InviteDeclinedEvent,
 		InviteConfirmedEvent,
 		InviteDeniedEvent,
-	}, invitationProjector)
+	}, invitationProjector); err != nil {
+		return fmt.Errorf("could not add invitation projector: %w", err)
+	}
 
 	// Create and register a read model for a guest list.
 	guestListProjector := NewGuestListProjector(guestListRepo, eventID)
-	eventBus.AddHandler(ctx, eh.MatchEvents{
+	if err := local.AddHandler(ctx, eh.MatchEvents{
 		InviteAcceptedEvent,
 		InviteDeclinedEvent,
 		InviteConfirmedEvent,
 		InviteDeniedEvent,
-	}, guestListProjector)
+	}, guestListProjector); err != nil {
+		return fmt.Errorf("could not add guest list projector: %w", err)
+	}
 
 	// Setup the saga that responds to the accepted guests and limits the total
 	// amount of guests, responding with a confirmation or denial.
 	responseSaga := saga.NewEventHandler(NewResponseSaga(2), commandBus)
-	eventBus.AddHandler(ctx, eh.MatchEvents{InviteAcceptedEvent}, responseSaga)
+	if err := global.AddHandler(ctx, eh.MatchEvents{InviteAcceptedEvent}, responseSaga); err != nil {
+		return fmt.Errorf("could not add response saga to event bus: %w", err)
+	}
+
+	// Add a logger as an observer.
+	if err := global.AddHandler(ctx, eh.MatchAll{},
+		eh.UseEventHandlerMiddleware(&Logger{}, observer.Middleware)); err != nil {
+		return fmt.Errorf("could not add logger to event bus: %w", err)
+	}
+
+	return nil
 }
diff --git a/examples/guestlist/outbox/outbox.go b/examples/guestlist/outbox/outbox.go
new file mode 100644
index 0000000..e46c1c2
--- /dev/null
+++ b/examples/guestlist/outbox/outbox.go
@@ -0,0 +1,16 @@
+// Copyright (c) 2021 - The Event Horizon authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// Package outbox contains an example of a CQRS/ES app using the outbox pattern.
+package outbox
diff --git a/examples/todomvc/Dockerfile b/examples/todomvc/Dockerfile
index f52ff99..445d8cb 100644
--- a/examples/todomvc/Dockerfile
+++ b/examples/todomvc/Dockerfile
@@ -1,4 +1,4 @@
-FROM golang:1.16-alpine3.13 as builder
+FROM golang:1.17-alpine3.14 as builder
 
 RUN apk -U upgrade && \
     apk add --update ca-certificates tzdata curl gzip
@@ -19,9 +19,9 @@ RUN elm make src/Main.elm --output=elm.js
 
 # Build backend.
 WORKDIR /eventhorizon/examples/todomvc/backend
-RUN GOOS=linux GOARCH=amd64 CGO_ENABLED=0 go build .
+RUN CGO_ENABLED=0 go build .
 
-FROM alpine:3.12
+FROM alpine:3.14
 
 # Import certs and timezone data.
 COPY --from=builder /etc/ssl/certs/ca-certificates.crt /etc/ssl/certs/
diff --git a/examples/todomvc/Makefile b/examples/todomvc/Makefile
index 5e23c38..8eb4705 100644
--- a/examples/todomvc/Makefile
+++ b/examples/todomvc/Makefile
@@ -6,11 +6,11 @@ build:
 
 .PHONY: run
 run:
-	docker-compose up todomvc
+	docker-compose up --build todomvc
 
 .PHONY: run_services
 run_services:
-	docker-compose up -d mongo gpubsub tracing
+	docker-compose up -d mongodb redis tracing
 
 .PHONY: stop
 stop:
diff --git a/examples/todomvc/backend/domains/todo/setup.go b/examples/todomvc/backend/domains/todo/setup.go
index 64c2352..c59687c 100644
--- a/examples/todomvc/backend/domains/todo/setup.go
+++ b/examples/todomvc/backend/domains/todo/setup.go
@@ -13,11 +13,15 @@ import (
 	"github.com/looplab/eventhorizon/repo/mongodb"
 )
 
+type HandlerAdder interface {
+	AddHandler(context.Context, eh.EventMatcher, eh.EventHandler) error
+}
+
 // SetupDomain sets up the Todo domain.
 func SetupDomain(
 	commandBus *bus.CommandHandler,
 	eventStore eh.EventStore,
-	eventBus eh.EventBus,
+	local HandlerAdder,
 	repo eh.ReadWriteRepo,
 ) error {
 	ctx := context.Background()
@@ -33,7 +37,7 @@ func SetupDomain(
 	// Create the read model projector.
 	projector := projector.NewEventHandler(&Projector{}, repo)
 	projector.SetEntityFactory(func() eh.Entity { return &TodoList{} })
-	eventBus.AddHandler(ctx, eh.MatchEvents{
+	local.AddHandler(ctx, eh.MatchEvents{
 		Created,
 		Deleted,
 		ItemAdded,
diff --git a/examples/todomvc/backend/main.go b/examples/todomvc/backend/main.go
index 36fa61f..d64b519 100644
--- a/examples/todomvc/backend/main.go
+++ b/examples/todomvc/backend/main.go
@@ -16,19 +16,24 @@ package main
 
 import (
 	"context"
+	"crypto/rand"
+	"encoding/hex"
 	"log"
 	"net/http"
 	"os"
 	"os/signal"
+	"time"
 
 	eh "github.com/looplab/eventhorizon"
 	"github.com/looplab/eventhorizon/commandhandler/bus"
-	gcpEventBus "github.com/looplab/eventhorizon/eventbus/gcp"
+
+	redisEventBus "github.com/looplab/eventhorizon/eventbus/redis"
 	tracingEventBus "github.com/looplab/eventhorizon/eventbus/tracing"
 	mongoEventStore "github.com/looplab/eventhorizon/eventstore/mongodb"
 	tracingEventStore "github.com/looplab/eventhorizon/eventstore/tracing"
 	"github.com/looplab/eventhorizon/middleware/commandhandler/tracing"
 	"github.com/looplab/eventhorizon/middleware/eventhandler/observer"
+	mongoOutbox "github.com/looplab/eventhorizon/outbox/mongodb"
 	mongoRepo "github.com/looplab/eventhorizon/repo/mongodb"
 	tracingRepo "github.com/looplab/eventhorizon/repo/tracing"
 	"github.com/looplab/eventhorizon/repo/version"
@@ -42,16 +47,16 @@ func main() {
 	log.Println("starting TodoMVC backend")
 
 	// Use MongoDB in Docker with fallback to localhost.
-	addr := os.Getenv("MONGODB_ADDR")
-	if addr == "" {
-		addr = "localhost:27017"
+	mongodbAddr := os.Getenv("MONGODB_ADDR")
+	if mongodbAddr == "" {
+		mongodbAddr = "localhost:27017"
 	}
-	url := "mongodb://" + addr
-	dbPrefix := "todomvc-example"
+	mongodbURI := "mongodb://" + mongodbAddr
 
 	// Connect to localhost if not running inside docker
-	if os.Getenv("PUBSUB_EMULATOR_HOST") == "" {
-		os.Setenv("PUBSUB_EMULATOR_HOST", "localhost:8793")
+	redisAddr := os.Getenv("REDIS_ADDR")
+	if redisAddr == "" {
+		redisAddr = "localhost:6379"
 	}
 
 	// Connect to localhost if not running inside docker
@@ -60,62 +65,83 @@ func main() {
 		tracingURL = "localhost"
 	}
 
-	traceCloser, err := NewTracer("todomvc", tracingURL)
+	// Get a random DB name.
+	b := make([]byte, 4)
+	if _, err := rand.Read(b); err != nil {
+		log.Fatal("could not get random DB name:", err)
+	}
+	db := "todomvc-" + hex.EncodeToString(b)
+	log.Println("using DB:", db)
+
+	traceCloser, err := NewTracer(db, tracingURL)
 	if err != nil {
 		log.Fatal("could not create tracer: ", err)
 	}
 
-	// Create an event bus.
-	commandBus := bus.NewCommandHandler()
-
-	// Create the event bus that distributes events.
-	var eventBus eh.EventBus
-	if eventBus, err = gcpEventBus.NewEventBus("project-id", dbPrefix); err != nil {
-		log.Fatal("could not create event bus: ", err)
+	// Create the outbox that will project and publish events.
+	outbox, err := mongoOutbox.NewOutbox(mongodbURI, db)
+	if err != nil {
+		log.Fatalf("could not create outbox: %s", err)
 	}
 	go func() {
-		for err := range eventBus.Errors() {
-			log.Print("eventbus:", err)
+		for err := range outbox.Errors() {
+			log.Print("outbox:", err)
 		}
 	}()
 
-	// Wrap the event bus to add tracing.
-	eventBus = tracingEventBus.NewEventBus(eventBus)
-
 	// Create the event store.
 	var eventStore eh.EventStore
-	if eventStore, err = mongoEventStore.NewEventStore(url, dbPrefix,
-		mongoEventStore.WithEventHandler(eventBus), // Add the event bus as a handler after save.
+	if eventStore, err = mongoEventStore.NewEventStoreWithClient(
+		outbox.Client(), db,
+		mongoEventStore.WithEventHandlerInTX(outbox),
 	); err != nil {
 		log.Fatal("could not create event store: ", err)
 	}
 	eventStore = tracingEventStore.NewEventStore(eventStore)
 
-	ctx := context.Background()
-
-	// Add an event logger as an observer.
-	eventLogger := &EventLogger{}
-	if err := eventBus.AddHandler(ctx, eh.MatchAll{},
-		eh.UseEventHandlerMiddleware(eventLogger,
-			observer.NewMiddleware(observer.NamedGroup("todomvc")),
-		),
-	); err != nil {
-		log.Fatal("could not add event logger: ", err)
-	}
+	// Create an command bus.
+	commandBus := bus.NewCommandHandler()
 
 	// Create the repository and wrap in a version repository.
 	var todoRepo eh.ReadWriteRepo
-	if todoRepo, err = mongoRepo.NewRepo(url, dbPrefix, "todos"); err != nil {
+	if todoRepo, err = mongoRepo.NewRepo(mongodbURI, db, "todos"); err != nil {
 		log.Fatal("could not create invitation repository: ", err)
 	}
 	todoRepo = version.NewRepo(todoRepo)
 	todoRepo = tracingRepo.NewRepo(todoRepo)
 
 	// Setup the Todo domain.
-	if err := todo.SetupDomain(commandBus, eventStore, eventBus, todoRepo); err != nil {
+	if err := todo.SetupDomain(commandBus, eventStore, outbox, todoRepo); err != nil {
 		log.Fatal("could not setup Todo domain: ", err)
 	}
 
+	ctx := context.Background()
+
+	// Create the event bus that distributes events.
+	var eventBus eh.EventBus
+	if eventBus, err = redisEventBus.NewEventBus(redisAddr, db, "backend"); err != nil {
+		log.Fatal("could not create event bus: ", err)
+	}
+	go func() {
+		for err := range eventBus.Errors() {
+			log.Print("eventbus:", err)
+		}
+	}()
+	eventBus = tracingEventBus.NewEventBus(eventBus)
+	if err := outbox.AddHandler(ctx, eh.MatchAll{}, eventBus); err != nil {
+		log.Fatal("could not add event bus to outbox:", err)
+	}
+
+	// Add an event logger as an observer.
+	eventLogger := &EventLogger{}
+	if err := eventBus.AddHandler(ctx, eh.MatchAll{},
+		eh.UseEventHandlerMiddleware(eventLogger,
+			observer.NewMiddleware(observer.NamedGroup("global")),
+		),
+	); err != nil {
+		log.Fatal("could not add event logger: ", err)
+	}
+
 	// Add tracing middleware to init tracing spans, and the logging middleware.
 	commandHandler := eh.UseCommandHandlerMiddleware(commandBus,
 		tracing.NewMiddleware(),
@@ -127,59 +153,21 @@ func main() {
 	if err != nil {
 		log.Fatal("could not create handler: ", err)
 	}
-
-	log.Println("adding a todo list with a few example items")
-	cmdCtx := context.Background()
-	id := uuid.New()
-	if err := commandHandler.HandleCommand(cmdCtx, &todo.Create{
-		ID: id,
-	}); err != nil {
-		log.Fatal("there should be no error: ", err)
-	}
-
-	// Add some examples and check them off.
-	if err := commandHandler.HandleCommand(cmdCtx, &todo.AddItem{
-		ID:          id,
-		Description: "Build the TodoMVC example",
-	}); err != nil {
-		log.Fatal("there should be no error: ", err)
-	}
-	if err := commandHandler.HandleCommand(cmdCtx, &todo.AddItem{
-		ID:          id,
-		Description: "Run the TodoMVC example",
-	}); err != nil {
-		log.Fatal("there should be no error: ", err)
-	}
-	findCtx, cancelFind := version.NewContextWithMinVersionWait(cmdCtx, 3)
-	if _, err := todoRepo.Find(findCtx, id); err != nil {
-		log.Fatal("could not find created todo list: ", err)
-	}
-	cancelFind()
-	if err := commandHandler.HandleCommand(cmdCtx, &todo.CheckAllItems{
-		ID: id,
-	}); err != nil {
-		log.Fatal("there should be no error: ", err)
+	srv := &http.Server{
+		Addr:    ":8080",
+		Handler: h,
 	}
+	srvClosed := make(chan struct{})
+	go func() {
+		if err := srv.ListenAndServe(); err != http.ErrServerClosed {
+			log.Fatal("could not listen HTTP: ", err)
+		}
+		close(srvClosed)
+	}()
 
-	// Add some more unchecked example items.
-	if err := commandHandler.HandleCommand(cmdCtx, &todo.AddItem{
-		ID:          id,
-		Description: "Learn Go",
-	}); err != nil {
-		log.Fatal("there should be no error: ", err)
-	}
-	if err := commandHandler.HandleCommand(cmdCtx, &todo.AddItem{
-		ID:          id,
-		Description: "Read the Event Horizon source",
-	}); err != nil {
-		log.Fatal("there should be no error: ", err)
-	}
-	if err := commandHandler.HandleCommand(cmdCtx, &todo.AddItem{
-		ID:          id,
-		Description: "Create a PR",
-	}); err != nil {
-		log.Fatal("there should be no error: ", err)
-	}
+	log.Println("adding a todo list with a few example items")
+	time.Sleep(3 * time.Second)
+	seedExample(commandHandler, todoRepo)
 
 	log.Printf(`
 
@@ -189,41 +177,30 @@ func main() {
 
 `)
 
-	srv := &http.Server{
-		Addr:    ":8080",
-		Handler: h,
-	}
-	srvClosed := make(chan struct{})
-	go func() {
-		sigint := make(chan os.Signal, 1)
-		signal.Notify(sigint, os.Interrupt)
-		<-sigint
-		if err := srv.Shutdown(context.Background()); err != nil {
-			log.Print("could not shutdown HTTP server: ", err)
-		}
-		close(srvClosed)
-	}()
+	sigint := make(chan os.Signal, 1)
+	signal.Notify(sigint, os.Interrupt)
+	<-sigint
 
-	log.Println("serving HTTP on :8080")
-	if err := srv.ListenAndServe(); err != http.ErrServerClosed {
-		log.Fatal("could not listen HTTP: ", err)
+	log.Println("waiting for HTTP server to close")
+	if err := srv.Shutdown(context.Background()); err != nil {
+		log.Print("could not shutdown HTTP server: ", err)
 	}
-
-	log.Println("waiting for HTTP request to finish")
 	<-srvClosed
 
 	// Cancel all handlers and wait.
 	log.Println("waiting for handlers to finish")
+	if err := eventBus.Close(); err != nil {
+		log.Print("could not close event bus: ", err)
+	}
+	if err := outbox.Close(); err != nil {
+		log.Print("could not close outbox: ", err)
+	}
 	if err := todoRepo.Close(); err != nil {
 		log.Print("could not close todo repo: ", err)
 	}
 	if err := eventStore.Close(); err != nil {
 		log.Print("could not close event store: ", err)
 	}
-	if err := eventBus.Close(); err != nil {
-		log.Print("could not close event bus: ", err)
-	}
-
 	if err := traceCloser.Close(); err != nil {
 		log.Print("could not close tracer: ", err)
 	}
@@ -252,3 +229,57 @@ func (l *EventLogger) HandleEvent(ctx context.Context, event eh.Event) error {
 	log.Printf("EVENT: %s", event)
 	return nil
 }
+
+func seedExample(h eh.CommandHandler, todoRepo eh.ReadRepo) {
+	cmdCtx := context.Background()
+	id := uuid.New()
+	if err := h.HandleCommand(cmdCtx, &todo.Create{
+		ID: id,
+	}); err != nil {
+		log.Fatal("there should be no error: ", err)
+	}
+
+	// Add some examples and check them off.
+	if err := h.HandleCommand(cmdCtx, &todo.AddItem{
+		ID:          id,
+		Description: "Build the TodoMVC example",
+	}); err != nil {
+		log.Fatal("there should be no error: ", err)
+	}
+	if err := h.HandleCommand(cmdCtx, &todo.AddItem{
+		ID:          id,
+		Description: "Run the TodoMVC example",
+	}); err != nil {
+		log.Fatal("there should be no error: ", err)
+	}
+	findCtx, cancelFind := version.NewContextWithMinVersionWait(cmdCtx, 3)
+	if _, err := todoRepo.Find(findCtx, id); err != nil {
+		log.Fatal("could not find created todo list: ", err)
+	}
+	cancelFind()
+	if err := h.HandleCommand(cmdCtx, &todo.CheckAllItems{
+		ID: id,
+	}); err != nil {
+		log.Fatal("there should be no error: ", err)
+	}
+
+	// Add some more unchecked example items.
+	if err := h.HandleCommand(cmdCtx, &todo.AddItem{
+		ID:          id,
+		Description: "Learn Go",
+	}); err != nil {
+		log.Fatal("there should be no error: ", err)
+	}
+	if err := h.HandleCommand(cmdCtx, &todo.AddItem{
+		ID:          id,
+		Description: "Read the Event Horizon source",
+	}); err != nil {
+		log.Fatal("there should be no error: ", err)
+	}
+	if err := h.HandleCommand(cmdCtx, &todo.AddItem{
+		ID:          id,
+		Description: "Create a PR",
+	}); err != nil {
+		log.Fatal("there should be no error: ", err)
+	}
+}
diff --git a/examples/todomvc/docker-compose.yml b/examples/todomvc/docker-compose.yml
index cd52103..7b81079 100644
--- a/examples/todomvc/docker-compose.yml
+++ b/examples/todomvc/docker-compose.yml
@@ -6,39 +6,45 @@ services:
       context: ../../
       dockerfile: ./examples/todomvc/Dockerfile
     depends_on:
-      - mongodb
-      - gpubsub
+      - mongodb-docker
+      - redis
       - tracing
     ports:
-      - "8080:8080"
+      - 8080:8080
     environment:
-      MONGODB_ADDR: "mongodb:27017"
-      PUBSUB_EMULATOR_HOST: "gpubsub:8793"
-      TRACING_URL: "tracing"
+      MONGODB_ADDR: mongodb-docker:27017
+      REDIS_ADDR: redis:6379
+      TRACING_URL: tracing
+
+  mongodb-docker:
+    image: bitnami/mongodb:4.4
+    ports:
+      - 27017:27017
+    environment:
+      MONGODB_REPLICA_SET_MODE: primary
+      ALLOW_EMPTY_PASSWORD: "yes"
 
   mongodb:
-    image: mongo:4.4
+    image: bitnami/mongodb:4.4
     ports:
-      - "27017:27017"
+      - 27017:27017
+    environment:
+      MONGODB_REPLICA_SET_MODE: primary
+      MONGODB_ADVERTISED_HOSTNAME: localhost # Needed to connect from localhost.
+      ALLOW_EMPTY_PASSWORD: "yes"
 
-  gpubsub:
-    image: google/cloud-sdk:318.0.0
+  redis:
+    image: redis:6.2-alpine3.13
     ports:
-      - "8793:8793"
-    entrypoint:
-      - gcloud
-      - beta
-      - emulators
-      - pubsub
-      - start
-      - "--host-port=0.0.0.0:8793"
+      - 6379:6379
+    entrypoint: [redis-server, --appendonly yes]
 
   tracing:
     image: docker.io/jaegertracing/all-in-one:1.16
     ports:
-      - "9411:9411"
-      - "16686:16686"
-      - "5778:5778"
+      - 9411:9411
+      - 16686:16686
+      - 5778:5778
     environment:
       # Enable Zipkin collector compatability.
       COLLECTOR_ZIPKIN_HTTP_PORT: 9411
diff --git a/mocks/mocks.go b/mocks/mocks.go
index b2df9de..470fd5c 100644
--- a/mocks/mocks.go
+++ b/mocks/mocks.go
@@ -252,6 +252,7 @@ func (m *EventHandler) Reset() {
 	m.Events = []eh.Event{}
 	m.Context = context.Background()
 	m.Time = time.Time{}
+	m.Err = nil
 }
 
 // Wait is a helper to wait some duration until for an event to be handled.
diff --git a/namespace/eventstore.go b/namespace/eventstore.go
index 4cfd34e..c5dd283 100644
--- a/namespace/eventstore.go
+++ b/namespace/eventstore.go
@@ -111,6 +111,8 @@ func (s *EventStore) eventStore(ctx context.Context) (eh.EventStore, error) {
 
 	if !ok {
 		s.eventStoresMu.Lock()
+		defer s.eventStoresMu.Unlock()
+
 		// Perform an additional existence check within the write lock in the
 		// unlikely event that someone else added the event store right before us.
 		if _, ok := s.eventStores[ns]; !ok {
@@ -123,7 +125,6 @@ func (s *EventStore) eventStore(ctx context.Context) (eh.EventStore, error) {
 
 			s.eventStores[ns] = eventStore
 		}
-		s.eventStoresMu.Unlock()
 	}
 
 	return eventStore, nil
diff --git a/namespace/outbox.go b/namespace/outbox.go
new file mode 100644
index 0000000..e2b52ca
--- /dev/null
+++ b/namespace/outbox.go
@@ -0,0 +1,186 @@
+// Copyright (c) 2021 - The Event Horizon authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package namespace
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"strings"
+	"sync"
+
+	eh "github.com/looplab/eventhorizon"
+)
+
+// Outbox is an outbox with support for namespaces passed in the context.
+type Outbox struct {
+	outboxes   map[string]eh.Outbox
+	outboxesMu sync.RWMutex
+	newOutbox  func(ns string) (eh.Outbox, error)
+	errCh      chan error
+	cctx       context.Context
+	cancel     context.CancelFunc
+	wg         sync.WaitGroup
+}
+
+// NewOutbox creates a new outbox which will use the provided factory
+// function to create new outboxes for the provided namespace.
+//
+// Usage:
+//    outbox := NewOutbox(func(ns string) (eh.Outbox, error) {
+//        s, err := mongodb.NewOutbox("mongodb://", ns)
+//        if err != nil {
+//            return nil, err
+//        }
+//        return s, nil
+//    })
+//
+// Usage shared DB client:
+//    client, err := mongo.Connect(ctx)
+//    ...
+//
+//    outbox := NewOutbox(func(ns string) (eh.Outbox, error) {
+//        s, err := mongodb.NewOutboxWithClient(client, ns)
+//        if err != nil {
+//            return nil, err
+//        }
+//        return s, nil
+//    })
+func NewOutbox(factory func(ns string) (eh.Outbox, error)) *Outbox {
+	ctx, cancel := context.WithCancel(context.Background())
+
+	return &Outbox{
+		outboxes:  map[string]eh.Outbox{},
+		newOutbox: factory,
+		errCh:     make(chan error, 100),
+		cctx:      ctx,
+		cancel:    cancel,
+	}
+}
+
+// HandlerType implements the HandlerType method of the eventhorizon.EventHandler interface.
+func (o *Outbox) HandlerType() eh.EventHandlerType {
+	return "outbox"
+}
+
+// AddHandler implements the AddHandler method of the eventhorizon.Outbox interface.
+func (o *Outbox) AddHandler(ctx context.Context, m eh.EventMatcher, h eh.EventHandler) error {
+	ob, err := o.outbox(ctx)
+	if err != nil {
+		return err
+	}
+
+	return ob.AddHandler(ctx, m, h)
+}
+
+// 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)
+	if err != nil {
+		return err
+	}
+
+	return ob.HandleEvent(ctx, event)
+}
+
+// Start implements the Start method of the eventhorizon.Outbox interface.
+func (o *Outbox) Start() {
+	o.outboxesMu.RLock()
+	defer o.outboxesMu.RUnlock()
+
+	for _, ob := range o.outboxes {
+		ob.Start()
+	}
+}
+
+// Close implements the Close method of the eventhorizon.Outbox interface.
+func (o *Outbox) Close() error {
+	o.outboxesMu.RLock()
+	defer o.outboxesMu.RUnlock()
+
+	var errStrs []string
+
+	for _, ob := range o.outboxes {
+		if err := ob.Close(); err != nil {
+			errStrs = append(errStrs, err.Error())
+		}
+	}
+
+	if len(errStrs) > 0 {
+		return fmt.Errorf("multiple errors: %s", strings.Join(errStrs, ", "))
+	}
+
+	// Close all error channel forwarders.
+	o.cancel()
+	o.wg.Wait()
+
+	return nil
+}
+
+// Errors implements the Errors method of the eventhorizon.EventBus interface.
+func (o *Outbox) Errors() <-chan error {
+	return o.errCh
+}
+
+// outbox is a helper that returns or creates an outbox for each namespace.
+func (o *Outbox) outbox(ctx context.Context) (eh.Outbox, error) {
+	ns := FromContext(ctx)
+
+	o.outboxesMu.RLock()
+	ob, ok := o.outboxes[ns]
+	o.outboxesMu.RUnlock()
+
+	if !ok {
+		o.outboxesMu.Lock()
+		defer o.outboxesMu.Unlock()
+
+		// Perform an additional existence check within the write lock in the
+		// unlikely event that someone else added the outbox right before us.
+		if _, ok := o.outboxes[ns]; !ok {
+			var err error
+
+			ob, err = o.newOutbox(ns)
+			if err != nil {
+				return nil, fmt.Errorf("could not create outbox for namespace '%s': %w", ns, err)
+			}
+
+			o.wg.Add(1)
+
+			// Merge error channels for all outboxes.
+			go func(errCh <-chan error) {
+				defer o.wg.Done()
+
+				for {
+					select {
+					case <-o.cctx.Done():
+						return
+					case err := <-errCh:
+						select {
+						case o.errCh <- err:
+						default:
+							log.Printf("eventhorizon: missed error in namespace error forwarding: %s", err)
+						}
+					}
+				}
+			}(ob.Errors())
+
+			ob.Start()
+
+			o.outboxes[ns] = ob
+		}
+	}
+
+	return ob, nil
+}
diff --git a/namespace/repo.go b/namespace/repo.go
index 56c3904..7d56605 100644
--- a/namespace/repo.go
+++ b/namespace/repo.go
@@ -162,6 +162,8 @@ func (r *Repo) repo(ctx context.Context) (eh.ReadWriteRepo, error) {
 
 	if !ok {
 		r.reposMu.Lock()
+		defer r.reposMu.Unlock()
+
 		// Perform an additional existence check within the write lock in the
 		// unlikely event that someone else added the repo right before us.
 		if _, ok := r.repos[ns]; !ok {
@@ -174,8 +176,6 @@ func (r *Repo) repo(ctx context.Context) (eh.ReadWriteRepo, error) {
 
 			r.repos[ns] = repo
 		}
-
-		r.reposMu.Unlock()
 	}
 
 	return repo, nil
diff --git a/outbox.go b/outbox.go
new file mode 100644
index 0000000..df7a341
--- /dev/null
+++ b/outbox.go
@@ -0,0 +1,80 @@
+// Copyright (c) 2021 - The Event Horizon authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package eventhorizon
+
+import (
+	"context"
+)
+
+// Outbox is an outbox for events. It ensures that all handled events get handled
+// by all added handlers. Handling of events must be atomic to storing them for
+// at least once handling of events, which is often provided by storage adapter
+// using a transaction.
+type Outbox interface {
+	EventHandler
+
+	// AddHandler adds a handler for an event. Returns an error if either the
+	// matcher or handler is nil, the handler is already added or there was some
+	// other problem adding the handler.
+	AddHandler(context.Context, EventMatcher, EventHandler) error
+
+	// Start starts processing the outbox until the Close() is cancelled by
+	// handling all events using the added handlers. Should be called after all
+	// handlers are setup but before new events are starting to be handled.
+	Start()
+
+	// Close closes the Outbox and waits for all handlers to finish.
+	Close() error
+
+	// Errors returns an error channel where async handling errors are sent.
+	Errors() <-chan error
+}
+
+// OutboxError is an error in the outbox.
+type OutboxError struct {
+	// Err is the error.
+	Err error
+	// Ctx is the context used when the error happened.
+	Ctx context.Context
+	// Event is the event handeled when the error happened.
+	Event Event
+}
+
+// Error implements the Error method of the errors.Error interface.
+func (e *OutboxError) Error() string {
+	str := "outbox: "
+
+	if e.Err != nil {
+		str += e.Err.Error()
+	} else {
+		str += "unknown error"
+	}
+
+	if e.Event != nil {
+		str += " [" + e.Event.String() + "]"
+	}
+
+	return str
+}
+
+// Unwrap implements the errors.Unwrap method.
+func (e *OutboxError) Unwrap() error {
+	return e.Err
+}
+
+// Cause implements the github.com/pkg/errors Unwrap method.
+func (e *OutboxError) Cause() error {
+	return e.Unwrap()
+}
diff --git a/outbox/memory/graceful.go b/outbox/memory/graceful.go
new file mode 100644
index 0000000..8fac06d
--- /dev/null
+++ b/outbox/memory/graceful.go
@@ -0,0 +1,83 @@
+package memory
+
+import (
+	"context"
+	"time"
+)
+
+// GracefulContext is a context that can be gracefully cancelled after the original
+// context has been canceled, but with a grace period to let operations finish.
+// The KeepAlive method should be used to signal that  operations are still ongoing.
+type GracefulContext struct {
+	context.Context
+
+	doneCh chan struct{}
+}
+
+// NewGracefulContext will create a GracefulContext with an idle time which should be
+// longer than a typical operation cadence, and a "force after" duration which
+// should be long enough to let some amount of operations to finish.
+func NewGracefulContext(ctx context.Context, idle, forceAfter time.Duration) (
+	c context.Context,
+	keepAlive func(),
+	cancel func(),
+) {
+	keepAliveCh := make(chan struct{})
+	stopCh := make(chan struct{})
+	doneCh := make(chan struct{})
+
+	go func() {
+		defer close(doneCh)
+
+		select {
+		case <-ctx.Done():
+			// Will trigger the graceful shutdown.
+		case <-stopCh:
+			// Will skip graceful shutdown.
+			return
+		}
+
+		idleT := time.NewTimer(idle)
+		forceShutdown := time.After(forceAfter)
+
+		for {
+			select {
+			case <-keepAliveCh: // Reset the idle timer on activity.
+				if !idleT.Stop() {
+					<-idleT.C
+				}
+
+				idleT.Reset(idle)
+
+				continue
+			case <-idleT.C: // If there is no more activity, cancel.
+				return
+			case <-forceShutdown: // If there is too much activity, cancel anyway.
+				return
+			}
+		}
+	}()
+
+	c = &GracefulContext{
+		Context: ctx,
+		doneCh:  doneCh,
+	}
+
+	keepAlive = func() {
+		select {
+		case keepAliveCh <- struct{}{}:
+		default:
+		}
+	}
+
+	cancel = func() {
+		close(stopCh)
+	}
+
+	return c, keepAlive, cancel
+}
+
+// Done implements the Done method of the context.Context interface.
+func (g *GracefulContext) Done() <-chan struct{} {
+	return g.doneCh
+}
diff --git a/outbox/memory/outbox.go b/outbox/memory/outbox.go
new file mode 100644
index 0000000..4107ec6
--- /dev/null
+++ b/outbox/memory/outbox.go
@@ -0,0 +1,330 @@
+package memory
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"log"
+	"sync"
+	"time"
+
+	"github.com/jinzhu/copier"
+	bsonCodec "github.com/looplab/eventhorizon/codec/bson"
+	"github.com/looplab/eventhorizon/uuid"
+
+	eh "github.com/looplab/eventhorizon"
+)
+
+var (
+	// Interval in which to do a sweep of various unprocessed events.
+	PeriodicSweepInterval = 15 * time.Second
+
+	// Settings for how old different kind of unprocessed events needs to be
+	// to be processed by the periodic sweep.
+	PeriodicSweepAge   = 15 * time.Second
+	PeriodicCleanupAge = 10 * time.Minute
+)
+
+// Outbox implements an eventhorizon.Outbox for MongoDB.
+type Outbox struct {
+	db             map[uuid.UUID]*outboxDoc
+	dbMu           sync.RWMutex
+	handlers       []*matcherHandler
+	handlersByType map[eh.EventHandlerType]*matcherHandler
+	handlersMu     sync.RWMutex
+	watchCh        chan *outboxDoc
+	errCh          chan error
+	processingMu   sync.Mutex
+	cctx           context.Context
+	cancel         context.CancelFunc
+	wg             sync.WaitGroup
+	codec          eh.EventCodec
+}
+
+type matcherHandler struct {
+	eh.EventMatcher
+	eh.EventHandler
+}
+
+// NewOutbox creates a new Outbox with a MongoDB URI: `mongodb://hostname`.
+func NewOutbox() (*Outbox, error) {
+	ctx, cancel := context.WithCancel(context.Background())
+
+	o := &Outbox{
+		db:             map[uuid.UUID]*outboxDoc{},
+		handlersByType: map[eh.EventHandlerType]*matcherHandler{},
+		watchCh:        make(chan *outboxDoc, 100),
+		errCh:          make(chan error, 100),
+		cctx:           ctx,
+		cancel:         cancel,
+		codec:          &bsonCodec.EventCodec{},
+	}
+
+	return o, nil
+}
+
+// HandlerType implements the HandlerType method of the eventhorizon.EventHandler interface.
+func (o *Outbox) HandlerType() eh.EventHandlerType {
+	return "outbox"
+}
+
+// AddHandler implements the AddHandler method of the eventhorizon.Outbox interface.
+func (o *Outbox) AddHandler(ctx context.Context, m eh.EventMatcher, h eh.EventHandler) error {
+	if m == nil {
+		return eh.ErrMissingMatcher
+	}
+
+	if h == nil {
+		return eh.ErrMissingHandler
+	}
+
+	o.handlersMu.Lock()
+	defer o.handlersMu.Unlock()
+
+	if _, ok := o.handlersByType[h.HandlerType()]; ok {
+		return eh.ErrHandlerAlreadyAdded
+	}
+
+	mh := &matcherHandler{m, h}
+	o.handlers = append(o.handlers, mh)
+	o.handlersByType[h.HandlerType()] = mh
+
+	return nil
+}
+
+// Returns an added handler and matcher for a handler type.
+func (o *Outbox) handler(handlerType string) (*matcherHandler, bool) {
+	o.handlersMu.RLock()
+	defer o.handlersMu.RUnlock()
+
+	mh, ok := o.handlersByType[eh.EventHandlerType(handlerType)]
+
+	return mh, ok
+}
+
+// outboxDoc is the DB representation of an outbox entry.
+type outboxDoc struct {
+	ID        uuid.UUID
+	Event     eh.Event
+	Ctx       context.Context
+	Handlers  []string
+	CreatedAt time.Time
+	TakenAt   time.Time
+}
+
+// HandleEvent implements the HandleEvent method of the eventhorizon.EventHandler interface.
+func (o *Outbox) HandleEvent(ctx context.Context, event eh.Event) error {
+	o.handlersMu.RLock()
+	defer o.handlersMu.RUnlock()
+
+	o.dbMu.Lock()
+	defer o.dbMu.Unlock()
+
+	var handlerNames []string
+
+	for _, h := range o.handlers {
+		if h.Match(event) {
+			handlerNames = append(handlerNames, h.HandlerType().String())
+		}
+	}
+
+	// Create the event record with timestamp.
+	e, err := copyEvent(ctx, event)
+	if err != nil {
+		return fmt.Errorf("could not copy event: %w", err)
+	}
+
+	r := &outboxDoc{
+		ID:        uuid.New(),
+		Event:     e,
+		Ctx:       ctx,
+		Handlers:  handlerNames,
+		CreatedAt: time.Now(),
+	}
+
+	o.db[r.ID] = r
+
+	select {
+	case o.watchCh <- r:
+	default:
+		// TODO: Error.
+	}
+
+	return nil
+}
+
+// Start implements the Start method of the eventhorizon.Outbox interface.
+func (o *Outbox) Start() {
+	o.wg.Add(2)
+
+	go o.runPeriodicallyUntilCancelled(o.processWithWatch, time.Second)
+	go o.runPeriodicallyUntilCancelled(o.processFullOutbox, PeriodicSweepInterval)
+}
+
+// Close implements the Close method of the eventhorizon.EventBus interface.
+func (o *Outbox) Close() error {
+	o.cancel()
+	o.wg.Wait()
+
+	return nil
+}
+
+// Errors implements the Errors method of the eventhorizon.EventBus interface.
+func (o *Outbox) Errors() <-chan error {
+	return o.errCh
+}
+
+func (o *Outbox) runPeriodicallyUntilCancelled(f func(context.Context) error, d time.Duration) {
+	defer o.wg.Done()
+
+	for {
+		if err := f(o.cctx); err != nil {
+			if errors.Is(err, context.Canceled) {
+				return
+			}
+
+			select {
+			case o.errCh <- &eh.OutboxError{Err: err}:
+			default:
+				log.Printf("eventhorizon: missed error in MongoDB outbox processing: %s", err)
+			}
+		}
+
+		// Wait until next full run or cancelled.
+		select {
+		case <-time.After(d):
+		case <-o.cctx.Done():
+			return
+		}
+	}
+}
+
+func (o *Outbox) processWithWatch(ctx context.Context) error {
+	for {
+		select {
+		case <-ctx.Done():
+			return nil
+		case r := <-o.watchCh:
+			o.dbMu.Lock()
+
+			// Use a new context to let processing finish when canceled.
+			if err := o.processOutboxEvent(context.Background(), r, time.Now()); err != nil {
+				err = fmt.Errorf("could not process outbox event: %w", err)
+				select {
+				case o.errCh <- &eh.OutboxError{Err: err}:
+				default:
+					log.Printf("outbox: missed error in MongoDB outbox: %s", err)
+				}
+			}
+
+			o.dbMu.Unlock()
+		}
+	}
+}
+
+func (o *Outbox) processFullOutbox(ctx context.Context) error {
+	o.processingMu.Lock()
+	defer o.processingMu.Unlock()
+
+	o.dbMu.Lock()
+	defer o.dbMu.Unlock()
+
+	now := time.Now()
+
+	for _, r := range o.db {
+		// Take started but non-finished events after 15 sec,
+		// or non-started events after 10 min.
+		if !(r.TakenAt.Before(now.Add(-PeriodicSweepAge)) ||
+			(r.TakenAt.IsZero() && r.CreatedAt.Before(now.Add(-PeriodicCleanupAge)))) {
+			continue
+		}
+
+		// Use a new context to let processing finish when canceled.
+		if err := o.processOutboxEvent(context.Background(), r, now); err != nil {
+			return fmt.Errorf("could not process outbox event: %w", err)
+		}
+	}
+
+	return nil
+}
+
+func (o *Outbox) processOutboxEvent(ctx context.Context, r *outboxDoc, now time.Time) error {
+	event := r.Event
+
+	if r.TakenAt.IsZero() ||
+		r.TakenAt.Before(now.Add(-PeriodicSweepAge)) ||
+		(r.TakenAt.IsZero() && r.CreatedAt.Before(now.Add(-PeriodicCleanupAge))) {
+		r.TakenAt = now
+	} else {
+		return nil
+	}
+
+	var processedHandlers []interface{}
+
+	// Process all handlers without returning handler errors.
+	for _, handlerType := range r.Handlers {
+		mh, ok := o.handler(handlerType)
+		if !ok {
+			continue
+		}
+
+		if !mh.Match(event) {
+			continue
+		}
+
+		if err := mh.HandleEvent(r.Ctx, event); err != nil {
+			err = fmt.Errorf("could not handle event (%s): %w", mh.HandlerType(), err)
+			select {
+			case o.errCh <- &eh.OutboxError{Err: err, Ctx: ctx, Event: event}:
+			default:
+				log.Printf("outbox: missed error in MongoDB outbox: %s", err)
+			}
+		} else {
+			processedHandlers = append(processedHandlers, handlerType)
+		}
+	}
+
+	if len(processedHandlers) == len(r.Handlers) {
+		delete(o.db, r.ID)
+	} else if len(processedHandlers) > 0 {
+		for _, ph := range processedHandlers {
+			j := 0
+			for _, h := range r.Handlers {
+				if h != ph {
+					r.Handlers[j] = h
+					j++
+				}
+			}
+			r.Handlers = r.Handlers[:j]
+		}
+	}
+
+	return nil
+}
+
+// copyEvent duplicates an event.
+func copyEvent(ctx context.Context, event eh.Event) (eh.Event, error) {
+	var data eh.EventData
+
+	// Copy data if there is any.
+	if event.Data() != nil {
+		var err error
+		if data, err = eh.CreateEventData(event.EventType()); err != nil {
+			return nil, fmt.Errorf("could not create event data: %w", err)
+		}
+
+		copier.Copy(data, event.Data())
+	}
+
+	return eh.NewEvent(
+		event.EventType(),
+		data,
+		event.Timestamp(),
+		eh.ForAggregate(
+			event.AggregateType(),
+			event.AggregateID(),
+			event.Version(),
+		),
+		eh.WithMetadata(event.Metadata()),
+	), nil
+}
diff --git a/outbox/mongodb/graceful.go b/outbox/mongodb/graceful.go
new file mode 100644
index 0000000..cdac525
--- /dev/null
+++ b/outbox/mongodb/graceful.go
@@ -0,0 +1,83 @@
+package mongodb
+
+import (
+	"context"
+	"time"
+)
+
+// GracefulContext is a context that can be gracefully cancelled after the original
+// context has been canceled, but with a grace period to let operations finish.
+// The KeepAlive method should be used to signal that  operations are still ongoing.
+type GracefulContext struct {
+	context.Context
+
+	doneCh chan struct{}
+}
+
+// NewGracefulContext will create a GracefulContext with an idle time which should be
+// longer than a typical operation cadence, and a "force after" duration which
+// should be long enough to let some amount of operations to finish.
+func NewGracefulContext(ctx context.Context, idle, forceAfter time.Duration) (
+	c context.Context,
+	keepAlive func(),
+	cancel func(),
+) {
+	keepAliveCh := make(chan struct{})
+	stopCh := make(chan struct{})
+	doneCh := make(chan struct{})
+
+	go func() {
+		defer close(doneCh)
+
+		select {
+		case <-ctx.Done():
+			// Will trigger the graceful shutdown.
+		case <-stopCh:
+			// Will skip graceful shutdown.
+			return
+		}
+
+		idleT := time.NewTimer(idle)
+		forceShutdown := time.After(forceAfter)
+
+		for {
+			select {
+			case <-keepAliveCh: // Reset the idle timer on activity.
+				if !idleT.Stop() {
+					<-idleT.C
+				}
+
+				idleT.Reset(idle)
+
+				continue
+			case <-idleT.C: // If there is no more activity, cancel.
+				return
+			case <-forceShutdown: // If there is too much activity, cancel anyway.
+				return
+			}
+		}
+	}()
+
+	c = &GracefulContext{
+		Context: ctx,
+		doneCh:  doneCh,
+	}
+
+	keepAlive = func() {
+		select {
+		case keepAliveCh <- struct{}{}:
+		default:
+		}
+	}
+
+	cancel = func() {
+		close(stopCh)
+	}
+
+	return c, keepAlive, cancel
+}
+
+// Done implements the Done method of the context.Context interface.
+func (g *GracefulContext) Done() <-chan struct{} {
+	return g.doneCh
+}
diff --git a/outbox/mongodb/outbox.go b/outbox/mongodb/outbox.go
new file mode 100644
index 0000000..faffb0e
--- /dev/null
+++ b/outbox/mongodb/outbox.go
@@ -0,0 +1,440 @@
+package mongodb
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"log"
+	"sync"
+	"time"
+
+	bsonCodec "github.com/looplab/eventhorizon/codec/bson"
+	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/bson/primitive"
+	"go.mongodb.org/mongo-driver/mongo"
+	"go.mongodb.org/mongo-driver/mongo/options"
+	mongoOptions "go.mongodb.org/mongo-driver/mongo/options"
+	"go.mongodb.org/mongo-driver/mongo/readconcern"
+	"go.mongodb.org/mongo-driver/mongo/readpref"
+	"go.mongodb.org/mongo-driver/mongo/writeconcern"
+
+	eh "github.com/looplab/eventhorizon"
+)
+
+var (
+	// Interval in which to do a sweep of various unprocessed events.
+	PeriodicSweepInterval = 15 * time.Second
+
+	// Settings for how old different kind of unprocessed events needs to be
+	// to be processed by the periodic sweep.
+	PeriodicSweepAge   = 15 * time.Second
+	PeriodicCleanupAge = 10 * time.Minute
+)
+
+// Outbox implements an eventhorizon.Outbox for MongoDB.
+type Outbox struct {
+	client         *mongo.Client
+	outbox         *mongo.Collection
+	handlers       []*matcherHandler
+	handlersByType map[eh.EventHandlerType]*matcherHandler
+	handlersMu     sync.RWMutex
+	errCh          chan error
+	watchToken     string
+	resumeToken    bson.Raw
+	processingMu   sync.Mutex
+	cctx           context.Context
+	cancel         context.CancelFunc
+	wg             sync.WaitGroup
+	codec          eh.EventCodec
+}
+
+type matcherHandler struct {
+	eh.EventMatcher
+	eh.EventHandler
+}
+
+// NewOutbox creates a new Outbox with a MongoDB URI: `mongodb://hostname`.
+func NewOutbox(uri, db string, options ...Option) (*Outbox, error) {
+	opts := mongoOptions.Client().ApplyURI(uri)
+	opts.SetWriteConcern(writeconcern.New(writeconcern.WMajority()))
+	opts.SetReadConcern(readconcern.Majority())
+	opts.SetReadPreference(readpref.Primary())
+
+	client, err := mongo.Connect(context.TODO(), opts)
+	if err != nil {
+		return nil, fmt.Errorf("could not connect to DB: %w", err)
+	}
+
+	return NewOutboxWithClient(client, db, options...)
+}
+
+// NewOutboxWithClient creates a new Outbox with a client.
+func NewOutboxWithClient(client *mongo.Client, db string, options ...Option) (*Outbox, error) {
+	if client == nil {
+		return nil, fmt.Errorf("missing DB client")
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+
+	o := &Outbox{
+		client:         client,
+		outbox:         client.Database(db).Collection("outbox"),
+		handlersByType: map[eh.EventHandlerType]*matcherHandler{},
+		errCh:          make(chan error, 100),
+		cctx:           ctx,
+		cancel:         cancel,
+		codec:          &bsonCodec.EventCodec{},
+	}
+
+	for _, option := range options {
+		if err := option(o); err != nil {
+			return nil, fmt.Errorf("error while applying option: %w", err)
+		}
+	}
+
+	if err := o.client.Ping(context.Background(), readpref.Primary()); err != nil {
+		return nil, fmt.Errorf("could not connect to MongoDB: %w", err)
+	}
+
+	return o, nil
+}
+
+// Option is an option setter used to configure creation.
+type Option func(*Outbox) error
+
+// WithWatchToken sets a token, used for watching outbox events stored with the
+// same token. This can be used to let parts of a system watch their own events
+// by setting a common token, for example a service name. If each host should
+// watch their local events the hostname can often be used.
+func WithWatchToken(token string) Option {
+	return func(o *Outbox) error {
+		o.watchToken = token
+
+		return nil
+	}
+}
+
+// Client returns the MongoDB client used by the outbox. To use the outbox with
+// the EventStore it needs to be created with the same client.
+func (o *Outbox) Client() *mongo.Client {
+	return o.client
+}
+
+// HandlerType implements the HandlerType method of the eventhorizon.EventHandler interface.
+func (o *Outbox) HandlerType() eh.EventHandlerType {
+	return "outbox"
+}
+
+// AddHandler implements the AddHandler method of the eventhorizon.Outbox interface.
+func (o *Outbox) AddHandler(ctx context.Context, m eh.EventMatcher, h eh.EventHandler) error {
+	if m == nil {
+		return eh.ErrMissingMatcher
+	}
+
+	if h == nil {
+		return eh.ErrMissingHandler
+	}
+
+	o.handlersMu.Lock()
+	defer o.handlersMu.Unlock()
+
+	if _, ok := o.handlersByType[h.HandlerType()]; ok {
+		return eh.ErrHandlerAlreadyAdded
+	}
+
+	mh := &matcherHandler{m, h}
+	o.handlers = append(o.handlers, mh)
+	o.handlersByType[h.HandlerType()] = mh
+
+	return nil
+}
+
+// Returns an added handler and matcher for a handler type.
+func (o *Outbox) handler(handlerType string) (*matcherHandler, bool) {
+	o.handlersMu.RLock()
+	defer o.handlersMu.RUnlock()
+
+	mh, ok := o.handlersByType[eh.EventHandlerType(handlerType)]
+
+	return mh, ok
+}
+
+// outboxDoc is the DB representation of an outbox entry.
+type outboxDoc struct {
+	ID         primitive.ObjectID `bson:"_id,omitempty"`
+	Event      bson.Raw           `bson:"event"`
+	Handlers   []string           `bson:"handlers"`
+	WatchToken string             `bson:"watch_token,omitempty"`
+	CreatedAt  time.Time          `bson:"created_at"`
+	TakenAt    time.Time          `bson:"taken_at,omitempty"`
+}
+
+// HandleEvent implements the HandleEvent method of the eventhorizon.EventHandler interface.
+func (o *Outbox) HandleEvent(ctx context.Context, event eh.Event) error {
+	o.handlersMu.RLock()
+	defer o.handlersMu.RUnlock()
+
+	e, err := o.codec.MarshalEvent(ctx, event)
+	if err != nil {
+		return fmt.Errorf("could not marshal event: %w", err)
+	}
+
+	var handlerNames []string
+
+	for _, h := range o.handlers {
+		if h.Match(event) {
+			handlerNames = append(handlerNames, h.HandlerType().String())
+		}
+	}
+
+	r := &outboxDoc{
+		Event:     e,
+		Handlers:  handlerNames,
+		CreatedAt: time.Now(),
+	}
+	if o.watchToken != "" {
+		r.WatchToken = o.watchToken
+	}
+
+	if _, err := o.outbox.InsertOne(ctx, r); err != nil {
+		return fmt.Errorf("could not queue event: %w", err)
+	}
+
+	return nil
+}
+
+// Start implements the Start method of the eventhorizon.Outbox interface.
+func (o *Outbox) Start() {
+	o.wg.Add(2)
+
+	go o.runPeriodicallyUntilCancelled(o.processWithWatch, time.Second)
+	go o.runPeriodicallyUntilCancelled(o.processFullOutbox, PeriodicSweepInterval)
+}
+
+// Close implements the Close method of the eventhorizon.EventBus interface.
+func (o *Outbox) Close() error {
+	o.cancel()
+	o.wg.Wait()
+
+	return nil
+}
+
+// Errors implements the Errors method of the eventhorizon.EventBus interface.
+func (o *Outbox) Errors() <-chan error {
+	return o.errCh
+}
+
+func (o *Outbox) runPeriodicallyUntilCancelled(f func(context.Context) error, d time.Duration) {
+	defer o.wg.Done()
+
+	for {
+		if err := f(o.cctx); err != nil {
+			if errors.Is(err, context.Canceled) {
+				return
+			}
+
+			select {
+			case o.errCh <- &eh.OutboxError{Err: err}:
+			default:
+				log.Printf("eventhorizon: missed error in MongoDB outbox processing: %s", err)
+			}
+		}
+
+		// Wait until next full run or cancelled.
+		select {
+		case <-time.After(d):
+		case <-o.cctx.Done():
+			return
+		}
+	}
+}
+
+func (o *Outbox) processWithWatch(ctx context.Context) error {
+	opts := options.ChangeStream().
+		SetBatchSize(1)
+	if o.resumeToken != nil {
+		opts = opts.SetResumeAfter(o.resumeToken)
+	}
+
+	match := bson.M{"operationType": "insert"}
+
+	// Match only documents with the same group token if that option is set.
+	if o.watchToken != "" {
+		match["fullDocument.watch_token"] = o.watchToken
+	}
+
+	stream, err := o.outbox.Watch(ctx, mongo.Pipeline{bson.D{{"$match", match}}}, opts)
+	if err != nil {
+		return fmt.Errorf("could not watch outbox: %w", err)
+	}
+
+	// Graceful cancel handling.
+	gracefulCtx, keepAlive, cancel := NewGracefulContext(ctx, 500*time.Millisecond, 5*time.Second)
+
+	// Return handling.
+	defer func() {
+		cancel()
+
+		o.resumeToken = stream.ResumeToken()
+	}()
+
+	// Watch loop.
+	for stream.Next(gracefulCtx) {
+		if err := o.processStreamEvent(stream.Current); err != nil {
+			select {
+			case o.errCh <- &eh.OutboxError{Err: err}:
+			default:
+				log.Printf("outbox: missed error in MongoDB outbox: %s", err)
+			}
+		}
+
+		// Ping the graceful exit handling, before exiting it's a no-op.
+		keepAlive()
+	}
+
+	if err := stream.Err(); err != nil {
+		return fmt.Errorf("watch error: %w", err)
+	}
+
+	return nil
+}
+
+func (o *Outbox) processStreamEvent(streamEvent bson.Raw) error {
+	o.processingMu.Lock()
+	defer o.processingMu.Unlock()
+
+	doc, err := streamEvent.LookupErr("fullDocument")
+	if err != nil {
+		return fmt.Errorf("missing outbox event: %w", err)
+	}
+
+	var r outboxDoc
+	if err := doc.Unmarshal(&r); err != nil {
+		return fmt.Errorf("could not unmarshal outbox event: %w", err)
+	}
+
+	// Use a new context to let processing finish when canceled.
+	if err := o.processOutboxEvent(context.Background(), &r, time.Now()); err != nil {
+		return fmt.Errorf("could not process outbox event: %w", err)
+	}
+
+	return nil
+}
+
+func (o *Outbox) processFullOutbox(ctx context.Context) error {
+	o.processingMu.Lock()
+	defer o.processingMu.Unlock()
+
+	// Keep track of the query time to avoid data races later when taking items.
+	now := time.Now()
+
+	// Take started but non-finished events after 15 sec,
+	// or non-started events after 10 min.
+	cur, err := o.outbox.Find(ctx, bson.M{"$or": bson.A{
+		bson.M{"taken_at": bson.M{"$lt": now.Add(-PeriodicSweepAge)}},
+		bson.M{"taken_at": nil, "created_at": bson.M{"$lt": now.Add(-PeriodicCleanupAge)}},
+	}})
+	if err != nil {
+		return fmt.Errorf("could not find outbox event: %w", err)
+	}
+
+	for cur.Next(ctx) {
+		var r outboxDoc
+		if err := cur.Decode(&r); err != nil {
+			return fmt.Errorf("could not unmarshal outbox event: %w", err)
+		}
+
+		// Use a new context to let processing finish when canceled.
+		if err := o.processOutboxEvent(context.Background(), &r, now); err != nil {
+			return fmt.Errorf("could not process outbox event: %w", err)
+		}
+	}
+
+	return cur.Close(ctx)
+}
+
+// The current time is passed to avoid data races between concurrent sweeps where
+// the fetch time and taken time could differ.
+func (o *Outbox) processOutboxEvent(ctx context.Context, r *outboxDoc, now time.Time) error {
+	event, ctx, err := o.codec.UnmarshalEvent(ctx, r.Event)
+	if err != nil {
+		return &eh.OutboxError{
+			Err: fmt.Errorf("could not unmarshal event: %w", err),
+			Ctx: ctx,
+		}
+	}
+
+	if res, err := o.outbox.UpdateOne(ctx, bson.M{
+		"_id": r.ID,
+		"$or": bson.A{
+			bson.M{"taken_at": nil},
+			bson.M{"taken_at": bson.M{"$lt": now.Add(-PeriodicSweepAge)}},
+			bson.M{"taken_at": nil, "created_at": bson.M{"$lt": now.Add(-PeriodicCleanupAge)}},
+		}},
+		bson.M{"$set": bson.M{"taken_at": now}},
+	); err != nil {
+		return &eh.OutboxError{
+			Err:   fmt.Errorf("could not take event for handling: %w", err),
+			Ctx:   ctx,
+			Event: event,
+		}
+	} else if res.MatchedCount == 0 {
+		return nil
+	}
+
+	var processedHandlers []interface{}
+
+	// Process all handlers without returning handler errors.
+	for _, handlerType := range r.Handlers {
+		mh, ok := o.handler(handlerType)
+		if !ok {
+			continue
+		}
+
+		if !mh.Match(event) {
+			continue
+		}
+
+		if err := mh.HandleEvent(ctx, event); err != nil {
+			err = fmt.Errorf("could not handle event (%s): %w", mh.HandlerType(), err)
+			select {
+			case o.errCh <- &eh.OutboxError{Err: err, Ctx: ctx, Event: event}:
+			default:
+				log.Printf("outbox: missed error in MongoDB outbox: %s", err)
+			}
+		} else {
+			processedHandlers = append(processedHandlers, handlerType)
+		}
+	}
+
+	if len(processedHandlers) == len(r.Handlers) {
+		if _, err := o.outbox.DeleteOne(ctx,
+			bson.M{"_id": r.ID},
+		); err != nil {
+			return &eh.OutboxError{
+				Err:   fmt.Errorf("could not delete outbox event: %w", err),
+				Ctx:   ctx,
+				Event: event,
+			}
+		}
+	} else if len(processedHandlers) > 0 {
+		if res, err := o.outbox.UpdateOne(ctx,
+			bson.M{"_id": r.ID},
+			bson.M{"$pullAll": bson.M{"handlers": bson.A(processedHandlers)}},
+		); err != nil {
+			return &eh.OutboxError{
+				Err:   fmt.Errorf("could not set outbox event as hadeled: %w", err),
+				Ctx:   ctx,
+				Event: event,
+			}
+		} else if res.MatchedCount == 0 {
+			return &eh.OutboxError{
+				Err:   fmt.Errorf("could not find outbox event to set as handled: %w", err),
+				Ctx:   ctx,
+				Event: event,
+			}
+		}
+	}
+
+	return nil
+}
diff --git a/repo/mongodb/repo.go b/repo/mongodb/repo.go
index 2a48a7e..673958c 100644
--- a/repo/mongodb/repo.go
+++ b/repo/mongodb/repo.go
@@ -34,11 +34,12 @@ import (
 	"github.com/looplab/eventhorizon/uuid"
 )
 
-// ErrModelNotSet is when a model factory is not set on the Repo.
-var ErrModelNotSet = errors.New("model not set")
-
-// ErrNoCursor is when a provided callback function returns a nil cursor.
-var ErrNoCursor = errors.New("no cursor")
+var (
+	// ErrModelNotSet is when a model factory is not set on the Repo.
+	ErrModelNotSet = errors.New("model not set")
+	// ErrNoCursor is when a provided callback function returns a nil cursor.
+	ErrNoCursor = errors.New("no cursor")
+)
 
 // Repo implements an MongoDB repository for entities.
 type Repo struct {
diff --git a/scripts/mongodb-rs-setup.sh b/scripts/mongodb-rs-setup.sh
deleted file mode 100644
index b9d7e14..0000000
--- a/scripts/mongodb-rs-setup.sh
+++ /dev/null
@@ -1,30 +0,0 @@
-#!/bin/bash
-
-MONGODB1=mongodb-rs
-
-echo "Waiting for MongoDB startup..."
-until curl http://${MONGODB1}:27017/serverStatus\?text\=1 2>&1 | grep uptime | head -1; do
-  printf '.'
-  sleep 1
-done
-
-# check if replica set is already initiated
-RS_STATUS=$( mongo --quiet --host ${MONGODB1}:27017 --eval "rs.status().ok" )
-if [[ $RS_STATUS != 1 ]]
-then
-  echo "[INFO] Replication set config invalid. Reconfiguring now."
-  RS_CONFIG_STATUS=$( mongo --quiet --host ${MONGODB1}:27017 -u --eval "rs.status().codeName" )
-  if [[ $RS_CONFIG_STATUS == 'InvalidReplicaSetConfig' ]]
-  then
-    mongo --quiet --host ${MONGODB1}:27017 <<EOF
-config = rs.config()
-config.members[0].host = 'localhost:27017' # Here is important to set the host name of the DB instance
-rs.reconfig(config, {force: true})
-EOF
-  else
-    echo "[INFO] MongoDB setup finished. Initiating replicata set."
-    mongo --quiet --host ${MONGODB1}:27017 --eval "rs.initiate({_id:'rs0', members:[{_id:0, host:'localhost:27017'}]})" > /dev/null
-  fi
-else
-  echo "[INFO] Replication set already initiated."
-fi


Test patch:
diff --git a/eventbus/kafka/eventbus_test.go b/eventbus/kafka/eventbus_test.go
index 6877683..b19a819 100644
--- a/eventbus/kafka/eventbus_test.go
+++ b/eventbus/kafka/eventbus_test.go
@@ -71,7 +71,7 @@ func TestEventBusIntegration(t *testing.T) {
 
 	t.Logf("using topic: %s_events", appID)
 
-	eventbus.AcceptanceTest(t, bus1, bus2, time.Second)
+	eventbus.AcceptanceTest(t, bus1, bus2, 3*time.Second)
 }
 
 func TestEventBusLoadtest(t *testing.T) {
diff --git a/eventhandler_test.go b/eventhandler_test.go
index db9b4e2..6f76b22 100644
--- a/eventhandler_test.go
+++ b/eventhandler_test.go
@@ -28,12 +28,14 @@ func TestEventHandlerFunc(t *testing.T) {
 
 		return nil
 	})
+
 	if h.HandlerType() != EventHandlerType("eventhorizon-TestEventHandlerFunc-func1") {
 		t.Error("the handler type should be correct:", h.HandlerType())
 	}
 
 	e := NewEvent("test", nil, time.Now())
 	h.HandleEvent(context.Background(), e)
+
 	if !reflect.DeepEqual(events, []Event{e}) {
 		t.Error("the events should be correct")
 		t.Log(events)
diff --git a/examples/guestlist/memory/memory_test.go b/examples/guestlist/memory/memory_test.go
index ab07ee5..ab0bc99 100644
--- a/examples/guestlist/memory/memory_test.go
+++ b/examples/guestlist/memory/memory_test.go
@@ -77,6 +77,7 @@ func Example() {
 	guestlist.Setup(
 		ctx,
 		eventStore,
+		eventBus, // Use the event bus both as local and global handler.
 		eventBus,
 		commandBus,
 		invitationRepo, guestListRepo,
@@ -163,6 +164,9 @@ func Example() {
 			l.NumGuests, l.NumAccepted, l.NumDeclined, l.NumConfirmed, l.NumDenied)
 	}
 
+	if err := eventBus.Close(); err != nil {
+		log.Println("error closing event bus:", err)
+	}
 	if err := invitationRepo.Close(); err != nil {
 		log.Println("error closing invitation repo:", err)
 	}
@@ -172,9 +176,6 @@ func Example() {
 	if err := eventStore.Close(); err != nil {
 		log.Println("error closing event store:", err)
 	}
-	if err := eventBus.Close(); err != nil {
-		log.Println("error closing event bus:", err)
-	}
 
 	// Output:
 	// invitation: Athena - confirmed
diff --git a/examples/guestlist/mongodb/mongodb_test.go b/examples/guestlist/mongodb/mongodb_test.go
index 19b93d4..2f722ea 100644
--- a/examples/guestlist/mongodb/mongodb_test.go
+++ b/examples/guestlist/mongodb/mongodb_test.go
@@ -52,7 +52,7 @@ guest list: 4 invited - 3 accepted, 1 declined - 2 confirmed, 1 denied`)
 		addr = "localhost:27017"
 	}
 	url := "mongodb://" + addr
-	dbPrefix := "guestlist-example"
+	dbPrefix := "guestlist"
 
 	// Create the event bus that distributes events.
 	eventBus := localEventBus.NewEventBus(nil)
@@ -90,32 +90,41 @@ guest list: 4 invited - 3 accepted, 1 declined - 2 confirmed, 1 denied`)
 
 	ctx := context.Background()
 
-	// Setup a test utility waiter that waits for all 11 events to occur before
-	// evaluating results.
-	var wg sync.WaitGroup
-	wg.Add(11)
-	eventBus.AddHandler(ctx, eh.MatchAll{}, eh.EventHandlerFunc(
-		func(ctx context.Context, e eh.Event) error {
-			wg.Done()
-			return nil
-		},
-	))
-
 	// Setup the guestlist.
 	eventID := uuid.New()
 	guestlist.Setup(
 		ctx,
 		eventStore,
+		eventBus, // Use the event bus both as local and global handler.
 		eventBus,
 		commandBus,
 		invitationVersionRepo, guestListRepo,
 		eventID,
 	)
 
+	// Setup a test utility waiter that waits for all 11 events to occur before
+	// evaluating results.
+	var wg sync.WaitGroup
+	wg.Add(11)
+	if err := eventBus.AddHandler(ctx, eh.MatchAll{}, eh.EventHandlerFunc(
+		func(ctx context.Context, e eh.Event) error {
+			wg.Done()
+			return nil
+		},
+	)); err != nil {
+		log.Fatal("could not add test sync to event bus:", err)
+	}
+
 	// Clear DB collections.
-	eventStore.Clear(ctx)
-	invitationRepo.Clear(ctx)
-	guestListRepo.Clear(ctx)
+	if err := eventStore.Clear(ctx); err != nil {
+		log.Fatal("could not clear event store:", err)
+	}
+	if err := invitationRepo.Clear(ctx); err != nil {
+		log.Fatal("could not clear invitation repo:", err)
+	}
+	if err := guestListRepo.Clear(ctx); err != nil {
+		log.Fatal("could not clear guest list repo:", err)
+	}
 
 	// --- Execute commands on the domain --------------------------------------
 
@@ -197,6 +206,9 @@ guest list: 4 invited - 3 accepted, 1 declined - 2 confirmed, 1 denied`)
 			l.NumGuests, l.NumAccepted, l.NumDeclined, l.NumConfirmed, l.NumDenied)
 	}
 
+	if err := eventBus.Close(); err != nil {
+		log.Println("error closing event bus:", err)
+	}
 	if err := invitationRepo.Close(); err != nil {
 		log.Println("error closing invitation repo:", err)
 	}
@@ -206,9 +218,6 @@ guest list: 4 invited - 3 accepted, 1 declined - 2 confirmed, 1 denied`)
 	if err := eventStore.Close(); err != nil {
 		log.Println("error closing event store:", err)
 	}
-	if err := eventBus.Close(); err != nil {
-		log.Println("error closing event bus:", err)
-	}
 
 	// Output:
 	// invitation: Athena - confirmed
diff --git a/examples/guestlist/outbox/outbox_test.go b/examples/guestlist/outbox/outbox_test.go
new file mode 100644
index 0000000..766963e
--- /dev/null
+++ b/examples/guestlist/outbox/outbox_test.go
@@ -0,0 +1,254 @@
+// Copyright (c) 2021 - The Event Horizon authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package outbox
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"os"
+	"sort"
+	"sync"
+	"testing"
+	"time"
+
+	eh "github.com/looplab/eventhorizon"
+	"github.com/looplab/eventhorizon/commandhandler/bus"
+	localEventBus "github.com/looplab/eventhorizon/eventbus/local"
+	mongoEventStore "github.com/looplab/eventhorizon/eventstore/mongodb"
+	mongoOutbox "github.com/looplab/eventhorizon/outbox/mongodb"
+	"github.com/looplab/eventhorizon/repo/mongodb"
+	"github.com/looplab/eventhorizon/repo/version"
+	"github.com/looplab/eventhorizon/uuid"
+
+	"github.com/looplab/eventhorizon/examples/guestlist/domains/guestlist"
+)
+
+func ExampleIntegration() {
+	if testing.Short() {
+		// Skip test when not running integration, fake success by printing.
+		fmt.Println(`invitation: Athena - confirmed
+invitation: Hades - confirmed
+invitation: Poseidon - denied
+invitation: Zeus - declined
+guest list: 4 invited - 3 accepted, 1 declined - 2 confirmed, 1 denied`)
+		return
+	}
+
+	// Use MongoDB in Docker with fallback to localhost.
+	addr := os.Getenv("MONGODB_ADDR")
+	if addr == "" {
+		addr = "localhost:27017"
+	}
+	url := "mongodb://" + addr
+	dbPrefix := "guestlist_outbox"
+
+	// Create the outbox that will project and publish events.
+	outbox, err := mongoOutbox.NewOutbox(url, dbPrefix)
+	if err != nil {
+		log.Fatalf("could not create outbox: %s", err)
+	}
+	go func() {
+		for e := range outbox.Errors() {
+			log.Printf("outbox: %s", e.Error())
+		}
+	}()
+
+	// Create the event store.
+	eventStore, err := mongoEventStore.NewEventStoreWithClient(
+		outbox.Client(), dbPrefix,
+		mongoEventStore.WithEventHandlerInTX(outbox),
+	)
+	if err != nil {
+		log.Fatalf("could not create event store: %s", err)
+	}
+
+	// Create the event bus that distributes events.
+	eventBus := localEventBus.NewEventBus(nil)
+	go func() {
+		for e := range eventBus.Errors() {
+			log.Printf("eventbus: %s", e.Error())
+		}
+	}()
+
+	// Create the command bus.
+	commandBus := bus.NewCommandHandler()
+
+	// Create the read repositories.
+	invitationRepo, err := mongodb.NewRepo(url, dbPrefix, "invitations")
+	if err != nil {
+		log.Fatalf("could not create invitation repository: %s", err)
+	}
+	invitationRepo.SetEntityFactory(func() eh.Entity { return &guestlist.Invitation{} })
+	// A version repo is needed for the projector to handle eventual consistency.
+	invitationVersionRepo := version.NewRepo(invitationRepo)
+
+	guestListRepo, err := mongodb.NewRepo(url, dbPrefix, "guest_lists")
+	if err != nil {
+		log.Fatalf("could not create guest list repository: %s", err)
+	}
+	guestListRepo.SetEntityFactory(func() eh.Entity { return &guestlist.GuestList{} })
+
+	ctx := context.Background()
+
+	// Setup the guestlist.
+	eventID := uuid.New()
+	if err := guestlist.Setup(
+		ctx,
+		eventStore,
+		outbox,
+		eventBus,
+		commandBus,
+		invitationVersionRepo, guestListRepo,
+		eventID,
+	); err != nil {
+		log.Fatal("could not setup domain:", err)
+	}
+
+	// Add the event bus as the last handler of the outbox.
+	if err := outbox.AddHandler(ctx, eh.MatchAll{}, eventBus); err != nil {
+		log.Fatal("could not add event bus to outbox:", err)
+	}
+
+	// Setup a test utility waiter that waits for all 11 events to occur before
+	// evaluating results.
+	var wg sync.WaitGroup
+	wg.Add(11)
+	if err := eventBus.AddHandler(ctx, eh.MatchAll{}, eh.EventHandlerFunc(
+		func(ctx context.Context, e eh.Event) error {
+			wg.Done()
+			return nil
+		},
+	)); err != nil {
+		log.Fatal("could not add test sync to event bus:", err)
+	}
+
+	// Clear DB collections.
+	if err := eventStore.Clear(ctx); err != nil {
+		log.Fatal("could not clear event store:", err)
+	}
+	if err := invitationRepo.Clear(ctx); err != nil {
+		log.Fatal("could not clear invitation repo:", err)
+	}
+	if err := guestListRepo.Clear(ctx); err != nil {
+		log.Fatal("could not clear guest list repo:", err)
+	}
+
+	outbox.Start()
+
+	// --- Execute commands on the domain --------------------------------------
+
+	// IDs for all the guests.
+	athenaID := uuid.New()
+	hadesID := uuid.New()
+	zeusID := uuid.New()
+	poseidonID := uuid.New()
+
+	// Issue some invitations and responses. Error checking omitted here.
+	if err := commandBus.HandleCommand(ctx, &guestlist.CreateInvite{ID: athenaID, Name: "Athena", Age: 42}); err != nil {
+		log.Println("error:", err)
+	}
+	if err := commandBus.HandleCommand(ctx, &guestlist.CreateInvite{ID: hadesID, Name: "Hades"}); err != nil {
+		log.Println("error:", err)
+	}
+	if err := commandBus.HandleCommand(ctx, &guestlist.CreateInvite{ID: zeusID, Name: "Zeus"}); err != nil {
+		log.Println("error:", err)
+	}
+	if err := commandBus.HandleCommand(ctx, &guestlist.CreateInvite{ID: poseidonID, Name: "Poseidon"}); err != nil {
+		log.Println("error:", err)
+	}
+
+	// The invited guests accept and decline the event.
+	// Note that Athena tries to decline the event after first accepting, but
+	// that is not allowed by the domain logic in InvitationAggregate. The
+	// result is that she is still accepted.
+	if err := commandBus.HandleCommand(ctx, &guestlist.AcceptInvite{ID: athenaID}); err != nil {
+		log.Println("error:", err)
+	}
+	if err = commandBus.HandleCommand(ctx, &guestlist.DeclineInvite{ID: athenaID}); err != nil {
+		// NOTE: This error is supposed to be printed!
+		log.Println("error:", err)
+	}
+	if err := commandBus.HandleCommand(ctx, &guestlist.AcceptInvite{ID: hadesID}); err != nil {
+		log.Println("error:", err)
+	}
+	if err := commandBus.HandleCommand(ctx, &guestlist.DeclineInvite{ID: zeusID}); err != nil {
+		log.Println("error:", err)
+	}
+
+	// Poseidon is a bit late to the party, will not be accepted...
+	if err := commandBus.HandleCommand(ctx, &guestlist.AcceptInvite{ID: poseidonID}); err != nil {
+		log.Println("error:", err)
+	}
+
+	// Wait for simulated eventual consistency before reading.
+	wg.Wait()
+	time.Sleep(1000 * time.Millisecond)
+
+	// Read all invites.
+	invitationStrs := []string{}
+	invitations, err := invitationRepo.FindAll(ctx)
+	if err != nil {
+		log.Println("error:", err)
+	}
+	for _, i := range invitations {
+		if i, ok := i.(*guestlist.Invitation); ok {
+			invitationStrs = append(invitationStrs, fmt.Sprintf("%s - %s", i.Name, i.Status))
+		}
+	}
+
+	// Sort the output to be able to compare test results.
+	sort.Strings(invitationStrs)
+	for _, s := range invitationStrs {
+		log.Printf("invitation: %s\n", s)
+		fmt.Printf("invitation: %s\n", s)
+	}
+
+	// Read the guest list.
+	l, err := guestListRepo.Find(ctx, eventID)
+	if err != nil {
+		log.Println("error:", err)
+	}
+	if l, ok := l.(*guestlist.GuestList); ok {
+		log.Printf("guest list: %d invited - %d accepted, %d declined - %d confirmed, %d denied\n",
+			l.NumGuests, l.NumAccepted, l.NumDeclined, l.NumConfirmed, l.NumDenied)
+		fmt.Printf("guest list: %d invited - %d accepted, %d declined - %d confirmed, %d denied\n",
+			l.NumGuests, l.NumAccepted, l.NumDeclined, l.NumConfirmed, l.NumDenied)
+	}
+
+	// Cancel all handlers and wait.
+	if err := eventBus.Close(); err != nil {
+		log.Println("error closing event bus:", err)
+	}
+	if err := outbox.Close(); err != nil {
+		log.Println("error closing outbox:", err)
+	}
+	if err := invitationRepo.Close(); err != nil {
+		log.Println("error closing invitation repo:", err)
+	}
+	if err := guestListRepo.Close(); err != nil {
+		log.Println("error closing guest list repo:", err)
+	}
+	if err := eventStore.Close(); err != nil {
+		log.Println("error closing event store:", err)
+	}
+
+	// Output:
+	// invitation: Athena - confirmed
+	// invitation: Hades - confirmed
+	// invitation: Poseidon - denied
+	// invitation: Zeus - declined
+	// guest list: 4 invited - 3 accepted, 1 declined - 2 confirmed, 1 denied
+}
diff --git a/matcher_test.go b/matcher_test.go
index 9856d66..e82c373 100644
--- a/matcher_test.go
+++ b/matcher_test.go
@@ -24,6 +24,7 @@ import (
 func TestMatchEvents(t *testing.T) {
 	et := EventType("test")
 	m := MatchEvents{et}
+
 	if m.Match(nil) {
 		t.Error("match event should not match nil event")
 	}
diff --git a/middleware_test.go b/middleware_test.go
index 3128e3c..b610b43 100644
--- a/middleware_test.go
+++ b/middleware_test.go
@@ -43,6 +43,7 @@ func TestCommandHandlerMiddleware(t *testing.T) {
 		middleware("third"),
 	)
 	h.HandleCommand(context.Background(), MiddlewareTestCommand{})
+
 	if !reflect.DeepEqual(order, []string{"first", "second", "third"}) {
 		t.Error("the order of middleware should be correct")
 		t.Log(order)
@@ -77,6 +78,7 @@ func TestEventHandlerMiddleware(t *testing.T) {
 		middleware("third"),
 	)
 	h.HandleEvent(context.Background(), NewEvent("test", nil, time.Now()))
+
 	if !reflect.DeepEqual(order, []string{"first", "second", "third"}) {
 		t.Error("the order of middleware should be correct")
 		t.Log(order)
diff --git a/namespace/eventstore_test.go b/namespace/eventstore_test.go
index f8e46e9..d4b497f 100644
--- a/namespace/eventstore_test.go
+++ b/namespace/eventstore_test.go
@@ -40,11 +40,12 @@ func TestEventStore(t *testing.T) {
 		t.Fatal("there should be a store")
 	}
 
-	// Default namespace.
+	t.Log("testing default namespace")
 	eventstore.AcceptanceTest(t, store, context.Background())
 
-	// Other namespace.
 	ctx := NewContext(context.Background(), "other")
+
+	t.Log("testing other namespace")
 	eventstore.AcceptanceTest(t, store, ctx)
 
 	if _, ok := usedNamespaces["default"]; !ok {
diff --git a/namespace/outbox_test.go b/namespace/outbox_test.go
new file mode 100644
index 0000000..e7c8541
--- /dev/null
+++ b/namespace/outbox_test.go
@@ -0,0 +1,59 @@
+package namespace
+
+import (
+	"context"
+	"math/rand"
+	"testing"
+	"time"
+
+	eh "github.com/looplab/eventhorizon"
+	"github.com/looplab/eventhorizon/outbox"
+	"github.com/looplab/eventhorizon/outbox/memory"
+)
+
+func init() {
+	rand.Seed(time.Now().Unix())
+}
+
+func TestOutbox(t *testing.T) {
+	usedNamespaces := map[string]struct{}{}
+
+	// Shorter sweeps for testing
+	memory.PeriodicSweepInterval = 2 * time.Second
+	memory.PeriodicSweepAge = 2 * time.Second
+
+	o := NewOutbox(func(ns string) (eh.Outbox, error) {
+		usedNamespaces[ns] = struct{}{}
+		o, err := memory.NewOutbox()
+		if err != nil {
+			return nil, err
+		}
+
+		return o, nil
+	})
+	if o == nil {
+		t.Fatal("there should be an outbox")
+	}
+
+	o.Start()
+
+	t.Log("testing default namespace")
+	outbox.AcceptanceTest(t, o, context.Background())
+
+	ctx := NewContext(context.Background(), "other")
+
+	t.Log("testing other namespace")
+	outbox.AcceptanceTest(t, o, ctx)
+
+	if _, ok := usedNamespaces["default"]; !ok {
+		t.Error("the default namespace should have been used")
+	}
+
+	if _, ok := usedNamespaces["other"]; !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 95c4a37..d0c141c 100644
--- a/namespace/repo_test.go
+++ b/namespace/repo_test.go
@@ -68,7 +68,9 @@ func TestReadRepo(t *testing.T) {
 		t.Error("the other repo should be correct")
 	}
 
-	// Test both namespaces.
+	t.Log("testing default namespace")
 	repo.AcceptanceTest(t, r, defaultCtx)
+
+	t.Log("testing other namespace")
 	repo.AcceptanceTest(t, r, otherCtx)
 }
diff --git a/outbox/acceptance_testing.go b/outbox/acceptance_testing.go
new file mode 100644
index 0000000..4686601
--- /dev/null
+++ b/outbox/acceptance_testing.go
@@ -0,0 +1,231 @@
+// Copyright (c) 2021 - The Event Horizon authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package outbox
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/kr/pretty"
+	eh "github.com/looplab/eventhorizon"
+	"github.com/looplab/eventhorizon/mocks"
+	"github.com/looplab/eventhorizon/uuid"
+)
+
+func TestAddHandler(t *testing.T, o eh.Outbox, ctx context.Context) {
+	// Error on nil matcher.
+	if err := o.AddHandler(ctx, nil, mocks.NewEventHandler("no-matcher")); err != eh.ErrMissingMatcher {
+		t.Error("the error should be correct:", err)
+	}
+
+	// Error on nil handler.
+	if err := o.AddHandler(ctx, eh.MatchAll{}, nil); err != eh.ErrMissingHandler {
+		t.Error("the error should be correct:", err)
+	}
+
+	// Error on multiple registrations.
+	if err := o.AddHandler(ctx, eh.MatchAll{}, mocks.NewEventHandler("multi")); err != nil {
+		t.Fatal("there should be no error:", err)
+	}
+
+	if err := o.AddHandler(ctx, eh.MatchAll{}, mocks.NewEventHandler("multi")); err != eh.ErrHandlerAlreadyAdded {
+		t.Error("the error should be correct:", err)
+	}
+}
+
+// AcceptanceTest is the acceptance test that all implementations of Outbox
+// should pass. It should manually be called from a test case in each
+// implementation:
+//
+//   func TestOutbox(t *testing.T) {
+//       o := NewOutbox()
+//       outbox.AcceptanceTest(t, o, context.Background())
+//   }
+//
+func AcceptanceTest(t *testing.T, o eh.Outbox, ctx context.Context) {
+	ctx = mocks.WithContextOne(ctx, "testval")
+
+	// Without handler.
+	id := uuid.New()
+	timestamp := time.Date(2009, time.November, 10, 23, 0, 0, 0, time.UTC)
+	event1 := eh.NewEvent(mocks.EventType, &mocks.EventData{Content: "event1"}, timestamp,
+		eh.ForAggregate(mocks.AggregateType, id, 1),
+		eh.WithMetadata(map[string]interface{}{"meta": "data", "num": 42.0}),
+	)
+
+	if err := o.HandleEvent(ctx, event1); err != nil {
+		t.Error("there should be no error:", err)
+	}
+
+	// Event without data (tested in its own handler).
+	otherHandler := mocks.NewEventHandler("other-handler")
+	if err := o.AddHandler(ctx, eh.MatchEvents{mocks.EventOtherType}, otherHandler); err != nil {
+		t.Fatal("there should be no error:", err)
+	}
+
+	eventWithoutData := eh.NewEvent(mocks.EventOtherType, nil, timestamp,
+		eh.ForAggregate(mocks.AggregateType, uuid.New(), 1))
+
+	if err := o.HandleEvent(ctx, eventWithoutData); err != nil {
+		t.Error("there should be no error:", err)
+	}
+
+	expectedEvents := []eh.Event{eventWithoutData}
+
+	if !otherHandler.Wait(time.Second) {
+		t.Error("did not receive event in time")
+	}
+
+	otherHandler.Lock()
+
+	if !eh.CompareEventSlices(otherHandler.Events, expectedEvents) {
+		t.Error("the events were incorrect:")
+		t.Log(otherHandler.Events)
+
+		if len(otherHandler.Events) == 1 {
+			t.Log(pretty.Sprint(otherHandler.Events[0]))
+		}
+	}
+
+	otherHandler.Unlock()
+
+	const (
+		handlerName = "handler"
+	)
+
+	// Add handlers and observers.
+	handler := mocks.NewEventHandler(handlerName)
+	if err := o.AddHandler(ctx, eh.MatchEvents{mocks.EventType}, handler); err != nil {
+		t.Fatal("there should be no error:", err)
+	}
+
+	anotherHandler := mocks.NewEventHandler("another_handler")
+	if err := o.AddHandler(ctx, eh.MatchEvents{mocks.EventType}, anotherHandler); err != nil {
+		t.Fatal("there should be no error:", err)
+	}
+
+	// Event with data.
+	event2 := eh.NewEvent(mocks.EventType, &mocks.EventData{Content: "event2"}, timestamp,
+		eh.ForAggregate(mocks.AggregateType, id, 2),
+		eh.WithMetadata(map[string]interface{}{"meta": "data", "num": 42.0}),
+	)
+	if err := o.HandleEvent(ctx, event2); err != nil {
+		t.Error("there should be no error:", err)
+	}
+
+	// Check for correct event in handler.
+	expectedEvents = []eh.Event{event2}
+
+	if !handler.Wait(time.Second) {
+		t.Error("did not receive event in time")
+	}
+
+	handler.Lock()
+
+	if !eh.CompareEventSlices(handler.Events, expectedEvents) {
+		t.Error("the events were incorrect:")
+		t.Log(handler.Events)
+	}
+
+	if val, ok := mocks.ContextOne(handler.Context); !ok || val != "testval" {
+		t.Error("the context should be correct:", handler.Context)
+	}
+
+	handler.Unlock()
+
+	// Check the other handler.
+	if !anotherHandler.Wait(time.Second) {
+		t.Error("did not receive event in time")
+	}
+
+	anotherHandler.Lock()
+
+	if !eh.CompareEventSlices(anotherHandler.Events, expectedEvents) {
+		t.Error("the events were incorrect:")
+		t.Log(anotherHandler.Events)
+	}
+
+	if val, ok := mocks.ContextOne(anotherHandler.Context); !ok || val != "testval" {
+		t.Error("the context should be correct:", anotherHandler.Context)
+	}
+
+	anotherHandler.Unlock()
+
+	// Check and clear all errors before the error tests.
+	checkOutboxErrors(t, o)
+
+	// Test async errors from handlers.
+	errorHandler := mocks.NewEventHandler("error_handler")
+	errorHandler.Err = errors.New("handler error")
+
+	if err := o.AddHandler(ctx, eh.MatchAll{}, errorHandler); err != nil {
+		t.Fatal("there should be no error:", err)
+	}
+
+	event3 := eh.NewEvent(mocks.EventType, &mocks.EventData{Content: "event3"}, timestamp,
+		eh.ForAggregate(mocks.AggregateType, id, 3),
+		eh.WithMetadata(map[string]interface{}{"meta": "data", "num": 42.0}),
+	)
+	if err := o.HandleEvent(ctx, event3); err != nil {
+		t.Error("there should be no error:", err)
+	}
+
+	select {
+	case <-time.After(time.Second):
+		t.Error("there should be an async error")
+	case err := <-o.Errors():
+		// Good case.
+		if err.Error() != "outbox: could not handle event (error_handler): handler error [Event("+id.String()+", v3)]" {
+			t.Error("incorrect error sent on outbox:", err)
+		}
+	}
+
+	// Reset error and let the periodic sweep process it.
+	errorHandler.Reset()
+
+	expectedEvents = []eh.Event{event3}
+
+	if !errorHandler.Wait(5 * time.Second) {
+		t.Error("did not receive event in time")
+	}
+
+	errorHandler.Lock()
+
+	if !eh.CompareEventSlices(errorHandler.Events, expectedEvents) {
+		t.Error("the events were incorrect:")
+		t.Log(errorHandler.Events)
+	}
+
+	errorHandler.Unlock()
+
+	checkOutboxErrors(t, o)
+}
+
+func checkOutboxErrors(t *testing.T, outbox eh.Outbox) {
+	t.Helper()
+
+	for {
+		select {
+		case err := <-outbox.Errors():
+			if err != nil {
+				t.Error("there should be no previous error:", err)
+			}
+		default:
+			return
+		}
+	}
+}
diff --git a/outbox/memory/outbox_test.go b/outbox/memory/outbox_test.go
new file mode 100644
index 0000000..dd2cdb0
--- /dev/null
+++ b/outbox/memory/outbox_test.go
@@ -0,0 +1,62 @@
+package memory
+
+import (
+	"context"
+	"math/rand"
+	"testing"
+	"time"
+
+	"github.com/looplab/eventhorizon/outbox"
+)
+
+func init() {
+	rand.Seed(time.Now().Unix())
+}
+
+func TestOutboxAddHandler(t *testing.T) {
+	o, err := NewOutbox()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	outbox.TestAddHandler(t, o, context.Background())
+}
+
+func TestOutbox(t *testing.T) {
+	// Shorter sweeps for testing
+	PeriodicSweepInterval = 2 * time.Second
+	PeriodicSweepAge = 2 * time.Second
+
+	o, err := NewOutbox()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	o.Start()
+
+	outbox.AcceptanceTest(t, o, context.Background())
+
+	if err := o.Close(); err != nil {
+		t.Error("there should be no error:", err)
+	}
+}
+
+func BenchmarkOutbox(b *testing.B) {
+	// Shorter sweeps for testing.
+	PeriodicSweepInterval = 1 * time.Second
+	PeriodicSweepAge = 1 * time.Second
+	PeriodicCleanupAge = 5 * time.Second
+
+	o, err := NewOutbox()
+	if err != nil {
+		b.Fatal(err)
+	}
+
+	o.Start()
+
+	outbox.Benchmark(b, o)
+
+	if err := o.Close(); err != nil {
+		b.Error("there should be no error:", err)
+	}
+}
diff --git a/outbox/mongodb/outbox_test.go b/outbox/mongodb/outbox_test.go
new file mode 100644
index 0000000..a02e304
--- /dev/null
+++ b/outbox/mongodb/outbox_test.go
@@ -0,0 +1,126 @@
+package mongodb
+
+import (
+	"context"
+	"encoding/hex"
+	"math/rand"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/looplab/eventhorizon/outbox"
+)
+
+func init() {
+	rand.Seed(time.Now().Unix())
+}
+
+func TestOutboxAddHandler(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test")
+	}
+
+	// Use MongoDB in Docker with fallback to localhost.
+	url := os.Getenv("MONGODB_ADDR")
+	if url == "" {
+		url = "localhost:27017"
+	}
+
+	url = "mongodb://" + url
+
+	// Get a random DB name.
+	bs := make([]byte, 4)
+	if _, err := rand.Read(bs); err != nil {
+		t.Fatal(err)
+	}
+
+	db := "test-" + hex.EncodeToString(bs)
+
+	t.Log("using DB:", db)
+
+	o, err := NewOutbox(url, db)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	outbox.TestAddHandler(t, o, context.Background())
+}
+
+func TestOutboxIntegration(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test")
+	}
+
+	// Use MongoDB in Docker with fallback to localhost.
+	url := os.Getenv("MONGODB_ADDR")
+	if url == "" {
+		url = "localhost:27017"
+	}
+
+	url = "mongodb://" + url
+
+	// Get a random DB name.
+	bs := make([]byte, 4)
+	if _, err := rand.Read(bs); err != nil {
+		t.Fatal(err)
+	}
+
+	db := "test-" + hex.EncodeToString(bs)
+
+	t.Log("using DB:", db)
+
+	// Shorter sweeps for testing
+	PeriodicSweepInterval = 2 * time.Second
+	PeriodicSweepAge = 2 * time.Second
+
+	o, err := NewOutbox(url, db)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	o.Start()
+
+	outbox.AcceptanceTest(t, o, context.Background())
+
+	if err := o.Close(); err != nil {
+		t.Error("there should be no error:", err)
+	}
+}
+
+func BenchmarkOutbox(b *testing.B) {
+	// Use MongoDB in Docker with fallback to localhost.
+	url := os.Getenv("MONGODB_ADDR")
+	if url == "" {
+		url = "localhost:27017"
+	}
+
+	url = "mongodb://" + url
+
+	// Get a random DB name.
+	bs := make([]byte, 4)
+	if _, err := rand.Read(bs); err != nil {
+		b.Fatal(err)
+	}
+
+	db := "test-" + hex.EncodeToString(bs)
+
+	b.Log("using DB:", db)
+
+	// Shorter sweeps for testing.
+	PeriodicSweepInterval = 1 * time.Second
+	PeriodicSweepAge = 1 * time.Second
+	PeriodicCleanupAge = 5 * time.Second
+
+	o, err := NewOutbox(url, db)
+	if err != nil {
+		b.Fatal(err)
+	}
+
+	o.Start()
+
+	outbox.Benchmark(b, o)
+
+	if err := o.Close(); err != nil {
+		b.Error("there should be no error:", err)
+	}
+}
diff --git a/outbox/performance_testing.go b/outbox/performance_testing.go
new file mode 100644
index 0000000..7ccdf8b
--- /dev/null
+++ b/outbox/performance_testing.go
@@ -0,0 +1,128 @@
+// Copyright (c) 2021 - The Event Horizon authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package outbox
+
+import (
+	"context"
+	"fmt"
+	"math/rand"
+	"sync"
+	"testing"
+	"time"
+
+	eh "github.com/looplab/eventhorizon"
+	"github.com/looplab/eventhorizon/mocks"
+	"github.com/looplab/eventhorizon/uuid"
+)
+
+func Benchmark(b *testing.B, o eh.Outbox) {
+	numAggregates := 50
+	numHandlers := 3
+	numEvents := b.N
+
+	b.Log("num iterations:", b.N)
+
+	ctx, cancel := context.WithCancel(context.Background())
+
+	handlers := make([]*mocks.EventHandler, numHandlers)
+	for i := range handlers {
+		h := mocks.NewEventHandler(fmt.Sprintf("handler-%d", i))
+
+		if err := o.AddHandler(context.Background(), eh.MatchAll{}, h); err != nil {
+			b.Fatal("there should be no error:", err)
+		}
+
+		handlers[i] = h
+	}
+
+	var wg sync.WaitGroup
+	for _, h := range handlers {
+		wg.Add(1)
+
+		go func(h *mocks.EventHandler) {
+			defer wg.Done()
+
+			numEventsReceived := 0
+
+			for {
+				if numEventsReceived == numEvents {
+					return
+				}
+				select {
+				case <-h.Recv:
+					numEventsReceived++
+
+					continue
+				case <-ctx.Done():
+					return
+				case <-time.After(30 * time.Second):
+					b.Error("did not receive message within timeout")
+
+					return
+				}
+			}
+		}(h)
+	}
+
+	var aggregates []*struct {
+		id      uuid.UUID
+		version int
+	}
+
+	for i := 0; i < numAggregates; i++ {
+		aggregates = append(aggregates, &struct {
+			id      uuid.UUID
+			version int
+		}{uuid.New(), 0})
+	}
+
+	b.Log("setup complete")
+	b.ResetTimer()
+
+	for n := 0; n < numEvents; n++ {
+		a := aggregates[rand.Intn(len(aggregates))]
+		a.version++
+
+		timestamp := time.Date(2009, time.November, 10, 23, n, 0, 0, time.UTC)
+		e := eh.NewEvent(
+			mocks.EventType, &mocks.EventData{Content: fmt.Sprintf("event%d", n)}, timestamp,
+			eh.ForAggregate(mocks.AggregateType, a.id, a.version))
+
+		if err := o.HandleEvent(context.Background(), e); err != nil {
+			b.Error("could not handle event:", err)
+		}
+	}
+
+	wg.Wait() // Wait for all events to be handled.
+	b.StopTimer()
+	cancel() // Stop handler goroutines.
+
+	checkOutboxBenchErrors(b, o)
+}
+
+func checkOutboxBenchErrors(b *testing.B, o eh.Outbox) {
+	b.Helper()
+
+	for {
+		select {
+		case err := <-o.Errors():
+			if err != nil {
+				b.Error("there should be no previous error:", err)
+			}
+		default:
+			return
+		}
+	}
+}