Remove useless []*Changed

Author: boekkooi-fresh

aggregate/aggregate.go

@@ -39,7 +39,7 @@ type (
 	}
 
 	eventSourced interface {
-		replay(aggregate EventApplier, historyEvents []*Changed)
+		replay(aggregate EventApplier, historyEvents goengine.EventStream) error
 		recordThat(aggregate EventApplier, event *Changed)
 	}
 )

aggregate/base_root.go

@@ -2,6 +2,8 @@ package aggregate
 
 import (
 	"sync"
+
+	"github.com/hellofresh/goengine"
 )
 
 var (
@@ -49,13 +51,32 @@ func (b *BaseRoot) popRecordedEvents() []*Changed {
 	return pendingEvents
 }
 
-func (b *BaseRoot) replay(aggregate EventApplier, historyEvents []*Changed) {
+func (b *BaseRoot) replay(aggregate EventApplier, streamEvents goengine.EventStream) error {
 	b.Lock()
 	defer b.Unlock()
 
-	for _, pastEvent := range historyEvents {
-		b.version = pastEvent.Version()
+	for streamEvents.Next() {
+		msg, _, err := streamEvents.Message()
+		if err != nil {
+			return err
+		}
+
+		changedEvent, ok := msg.(*Changed)
+		if !ok {
+			return ErrUnexpectedMessageType
+		}
+
+		b.version = changedEvent.Version()
+		aggregate.Apply(changedEvent)
+	}
 
-		aggregate.Apply(pastEvent)
+	if err := streamEvents.Err(); err != nil {
+		return err
 	}
+
+	if b.version == 0 {
+		return ErrEmptyEventStream
+	}
+
+	return nil
 }

aggregate/repository.go

@@ -94,32 +94,11 @@ func (r *Repository) GetAggregateRoot(ctx context.Context, aggregateID ID) (Root
 	}
 	defer streamEvents.Close()
 
-	var changedStream []*Changed
-	for streamEvents.Next() {
-		msg, _, err := streamEvents.Message()
-		if err != nil {
-			return nil, err
-		}
-
-		changedEvent, ok := msg.(*Changed)
-		if !ok {
-			return nil, ErrUnexpectedMessageType
-		}
-
-		changedStream = append(changedStream, changedEvent)
-	}
-
-	if err := streamEvents.Err(); err != nil {
+	root := r.aggregateType.CreateInstance()
+	if err = root.replay(root, streamEvents); err != nil {
 		return nil, err
 	}
 
-	if len(changedStream) == 0 {
-		return nil, ErrEmptyEventStream
-	}
-
-	root := r.aggregateType.CreateInstance()
-	root.replay(root, changedStream)
-
 	return root, nil
 }