Fixed mongo usage

Author: vgarvardt

cmd/goengine/main.go

@@ -11,9 +11,6 @@ import (
 	"github.com/hellofresh/goengine/rabbit"
 	"github.com/mongodb/mongo-go-driver/mongo"
 	"github.com/mongodb/mongo-go-driver/mongo/options"
-	"github.com/mongodb/mongo-go-driver/mongo/readconcern"
-	"github.com/mongodb/mongo-go-driver/mongo/readpref"
-	"github.com/mongodb/mongo-go-driver/mongo/writeconcern"
 )
 
 func main() {
@@ -31,11 +28,7 @@ func main() {
 	goengine.Log("Connecting to the database", map[string]interface{}{"dsn": mongoDSN}, nil)
 	mongoClient, err := mongo.NewClientWithOptions(
 		mongoDSN,
-		options.Client().
-			SetAppName("goengine").
-			SetReadConcern(readconcern.Linearizable()).
-			SetReadPreference(readpref.Nearest()).
-			SetWriteConcern(writeconcern.New(writeconcern.WMajority())),
+		options.Client().SetAppName("goengine"),
 	)
 	if err != nil {
 		goengine.Log("Failed to create new Mongo mongoClient", nil, err)
@@ -68,11 +61,11 @@ func main() {
 	bus := rabbit.NewEventBus(brokerDSN, "events", "events")
 
 	goengine.Log("Setting up the event store", nil, nil)
-	es := mongodb.NewEventStore(mongoClient.Database("event_store"), registry)
+	eventStore := mongodb.NewEventStore(mongoClient.Database("event_store"), registry)
 
 	eventDispatcher := goengine.NewVersionedEventDispatchManager(bus, registry)
 	eventDispatcher.RegisterEventHandler(&RecipeCreated{}, func(event *goengine.DomainMessage) error {
-		goengine.Log("Event received", nil, nil)
+		goengine.Log("Event received", map[string]interface{}{"event": event}, nil)
 		return nil
 	})
 
@@ -82,12 +75,15 @@ func main() {
 	goengine.Log("Creating a recipe", nil, nil)
 	aggregateRoot := CreateScenario()
 
-	repository := goengine.NewPublisherRepository(es, bus)
-	repository.Save(aggregateRoot, streamName)
+	repository := goengine.NewPublisherRepository(eventStore, bus)
+	if err := repository.Save(aggregateRoot, streamName); err != nil {
+		goengine.Log("Failed to save aggregate to stream", nil, err)
+		panic(err)
+	}
 
-	_, err = NewRecipeFromHisotry(aggregateRoot.ID, streamName, repository)
+	_, err = NewRecipeFromHistory(aggregateRoot.ID, streamName, repository)
 	if err != nil {
-		goengine.Log("Failed to connect to Mongo", nil, err)
+		goengine.Log("Failed get a recipe from history", nil, err)
 		panic(err)
 	}
 

cmd/goengine/model.go

@@ -7,21 +7,21 @@ import (
 )
 
 type RecipeCreated struct {
-	ocurredOn time.Time
-	Name      string
+	occurredOn time.Time
+	Name       string
 }
 
-func (e RecipeCreated) OcurredOn() time.Time {
-	return e.ocurredOn
+func (e RecipeCreated) OccurredOn() time.Time {
+	return e.occurredOn
 }
 
 type RecipeRated struct {
-	ocurredOn time.Time
-	Rating    int
+	occurredOn time.Time
+	Rating     int
 }
 
-func (e RecipeRated) OcurredOn() time.Time {
-	return e.ocurredOn
+func (e RecipeRated) OccurredOn() time.Time {
+	return e.occurredOn
 }
 
 type Recipe struct {
@@ -38,7 +38,7 @@ func NewRecipe(name string) *Recipe {
 	return recipe
 }
 
-func NewRecipeFromHisotry(id string, streamName goengine.StreamName, repo goengine.AggregateRepository) (*Recipe, error) {
+func NewRecipeFromHistory(id string, streamName goengine.StreamName, repo goengine.AggregateRepository) (*Recipe, error) {
 	recipe := new(Recipe)
 	recipe.AggregateRootBased = goengine.NewEventSourceBasedWithID(recipe, id)
 	err := repo.Reconstitute(id, recipe, streamName)

mongodb/eventstore.go

@@ -10,6 +10,7 @@ import (
 	"github.com/mongodb/mongo-go-driver/bson"
 	"github.com/mongodb/mongo-go-driver/mongo"
 	"github.com/mongodb/mongo-go-driver/mongo/options"
+	"github.com/mongodb/mongo-go-driver/x/bsonx"
 )
 
 // MongoEvent represents an event on mongodb
@@ -62,7 +63,7 @@ func (s *MongoDBEventStore) Append(events *goengine.EventStream) error {
 
 // GetEventsFor gets events for an id on the specified stream
 func (s *MongoDBEventStore) GetEventsFor(streamName goengine.StreamName, id string) (*goengine.EventStream, error) {
-	var mongoEvents []*MongoEvent
+	var mongoEvents []MongoEvent
 	coll := s.mongoDB.Collection(string(streamName))
 
 	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
@@ -74,8 +75,8 @@ func (s *MongoDBEventStore) GetEventsFor(streamName goengine.StreamName, id stri
 	}
 
 	for cur.Next(ctx) {
-		var mongoEvent *MongoEvent
-		err := cur.Decode(mongoEvent)
+		var mongoEvent MongoEvent
+		err := cur.Decode(&mongoEvent)
 		if err != nil {
 			return nil, err
 		}
@@ -102,7 +103,7 @@ func (s *MongoDBEventStore) GetEventsFor(streamName goengine.StreamName, id stri
 
 // FromVersion gets events for an id and version on the specified stream
 func (s *MongoDBEventStore) FromVersion(streamName goengine.StreamName, id string, version int) (*goengine.EventStream, error) {
-	var mongoEvents []*MongoEvent
+	var mongoEvents []MongoEvent
 	coll := s.mongoDB.Collection(string(streamName))
 
 	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
@@ -121,7 +122,7 @@ func (s *MongoDBEventStore) FromVersion(streamName goengine.StreamName, id strin
 	}
 
 	for cur.Next(ctx) {
-		var mongoEvent *MongoEvent
+		var mongoEvent MongoEvent
 		err := cur.Decode(mongoEvent)
 		if err != nil {
 			return nil, err
@@ -162,7 +163,7 @@ func (s *MongoDBEventStore) createIndexes(c *mongo.Collection) error {
 	_, err := c.Indexes().CreateOne(
 		ctx,
 		mongo.IndexModel{
-			Keys:    []string{"aggregate_id", "version"},
+			Keys:    bsonx.Doc{{"aggregate_id", bsonx.Int32(1)}, {"version", bsonx.Int32(-1)}},
 			Options: options.Index().SetUnique(true).SetBackground(true),
 		},
 	)
@@ -185,7 +186,7 @@ func (s *MongoDBEventStore) toMongoEvent(event *goengine.DomainMessage) (*MongoE
 	}, nil
 }
 
-func (s *MongoDBEventStore) fromMongoEvent(mongoEvent *MongoEvent) (*goengine.DomainMessage, error) {
+func (s *MongoDBEventStore) fromMongoEvent(mongoEvent MongoEvent) (*goengine.DomainMessage, error) {
 	event, err := s.registry.Get(mongoEvent.Type)
 	if nil != err {
 		return nil, err