feat: realtime events function for one or multiple aggregate types

README.md

@@ -223,9 +223,18 @@ type Encoder interface {
 }
 ```
 
-### Realtime Event Subscription
+### Realtime Event
 
-For now the real time event subscription has been removed as I'm not satisfied with the exported API. Please fill an issue if you want it back.
+`SetSaveHook` allows your application to register a callback that is triggered synchronously whenever events are saved for specific aggregates. This is useful for cases where you need real-time reactions to domain events.
+The hook runs synchronously during the Save operation. Keep its execution fast and non-blocking to avoid slowing down the save process.
+
+```go
+aggregate.SetSaveHook(func(events []eventsourcing.Event) {
+    for _, evt := range events {
+        log.Printf("New event saved: %v", evt)
+    }
+}, &OrderAggregate{}, &UserAggregate{})
+```
 
 ## Snapshot
 

aggregate/aggregate.go

@@ -13,7 +13,10 @@ import (
 
 type RegisterFunc = func(events ...interface{})
 
-// Aggregate interface to use the aggregate root specific methods
+// register holding functions that is triggered when events for an aggregate is saved.
+var saveHookMap = make(map[string][]func(events []eventsourcing.Event))
+
+// aggregate interface to use the aggregate root specific methods
 type aggregate interface {
 	root() *Root
 	Transition(event eventsourcing.Event)
@@ -78,20 +81,52 @@ func Save(es core.EventStore, a aggregate) error {
 	if err != nil {
 		return err
 	}
-	// update the global version on the aggregate
+	// set the global version
 	root.globalVersion = globalVersion
 
-	// set internal properties and reset the events slice
+	// set the  internal version
 	lastEvent := root.events[len(root.events)-1]
 	root.version = lastEvent.Version()
-	root.events = []eventsourcing.Event{}
 
+	// run save hook functions
+	for _, f := range saveHookMap[aggregateType(a)] {
+		f(root.events)
+	}
+
+	// clear the events
+	root.events = []eventsourcing.Event{}
 	return nil
 }
 
 // Register registers the aggregate and its events
 func Register(a aggregate) {
 	internal.GlobalRegister.Register(a)
+
+	// only create new map if the aggregate has not been registered before
+	if _, ok := saveHookMap[aggregateType(a)]; ok {
+		return
+	}
+	saveHookMap[aggregateType(a)] = []func(events []eventsourcing.Event){}
+}
+
+// ResetRegister reset the internal aggregate registers
+// This is mostly used in internal tests to make sure the registers are cleared
+func ResetRegister() {
+	internal.ResetRegister()
+	saveHookMap = make(map[string][]func(events []eventsourcing.Event))
+}
+
+// SetSaveHook enables the application to react in realtime when events are saved from specific aggregates.
+// Note that the function is ran in sync with the Save method and should return as fast as possible.
+// It return error if an aggregate is not registered via the aggregate.Register function.
+func SetSaveHook(f func(events []eventsourcing.Event), aggregates ...aggregate) error {
+	for _, a := range aggregates {
+		if !internal.GlobalRegister.AggregateRegistered(a) {
+			return fmt.Errorf("%s %w when calling the SetSaveHook", aggregateType(a), eventsourcing.ErrAggregateNotRegistered)
+		}
+		saveHookMap[aggregateType(a)] = append(saveHookMap[aggregateType(a)], f)
+	}
+	return nil
 }
 
 // Save events to the event store

aggregate/aggregate_test.go

@@ -2,6 +2,7 @@ package aggregate_test
 
 import (
 	"context"
+	"errors"
 	"testing"
 
 	"github.com/hallgren/eventsourcing"
@@ -49,7 +50,6 @@ func TestLoadAggregateFromSnapshot(t *testing.T) {
 	es := memory.Create()
 	ss := ss.Create()
 	aggregate.Register(&Person{})
-
 	person, err := CreatePerson("kalle")
 	if err != nil {
 		t.Fatal(err)
@@ -68,6 +68,9 @@ func TestLoadAggregateFromSnapshot(t *testing.T) {
 	// add one more event to the person aggregate
 	person.GrowOlder()
 	err = aggregate.Save(es, person)
+	if err != nil {
+		t.Fatal(err)
+	}
 
 	// load person to person2 from snaphost and events
 	person2 := &Person{}
@@ -90,3 +93,59 @@ func TestLoadNoneExistingAggregate(t *testing.T) {
 		t.Fatal("could not get aggregate")
 	}
 }
+
+func TestSaveHookAggregateNotRegistered(t *testing.T) {
+	aggregate.ResetRegister()
+	err := aggregate.SetSaveHook(func(events []eventsourcing.Event) {}, &Person{})
+	if !errors.Is(err, eventsourcing.ErrAggregateNotRegistered) {
+		t.Fatalf("expected error eventsourcing.ErrAggregateNotRegistered got %v", err)
+	}
+}
+
+func TestSaveHook(t *testing.T) {
+	var trigger bool
+	var event eventsourcing.Event
+	es := memory.Create()
+	aggregate.Register(&Person{})
+
+	person, err := CreatePerson("kalle")
+	if err != nil {
+		t.Fatal(err)
+	}
+	err = aggregate.Save(es, person)
+	if err != nil {
+		t.Fatalf("could not save aggregate, err: %v", err)
+	}
+	if trigger {
+		t.Fatal("post trigger should not be activated")
+	}
+
+	// set post save trigger functions
+	err = aggregate.SetSaveHook(func(events []eventsourcing.Event) {
+		trigger = true
+	}, &Person{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	err = aggregate.SetSaveHook(func(events []eventsourcing.Event) {
+		event = events[0]
+	}, &Person{})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// make sure double register does not affect save hook
+	aggregate.Register(&Person{})
+
+	person.GrowOlder()
+	err = aggregate.Save(es, person)
+	if err != nil {
+		t.Fatalf("could not save aggregate, err: %v", err)
+	}
+	if !trigger {
+		t.Fatal("post trigger should be activated")
+	}
+	if event.Reason() != "AgedOneYear" {
+		t.Fatalf("expected AgedOneYear got %v", event.Reason())
+	}
+}

eventsourcing.go

@@ -10,7 +10,7 @@ var (
 	// ErrAggregateNotFound returns if events not found for aggregate or aggregate was not based on snapshot from the outside
 	ErrAggregateNotFound = errors.New("aggregate not found")
 
-	// ErrAggregateNotRegistered when saving aggregate when it's not registered in the repository
+	// ErrAggregateNotRegistered when aggregate is not registered via the aggregate.Register method and can be returned by aggregate.Save or aggregate.SetSaveHook
 	ErrAggregateNotRegistered = errors.New("aggregate not registered")
 
 	// ErrEventNotRegistered when saving aggregate and one event is not registered in the repository

example/go.mod

@@ -6,4 +6,4 @@ require github.com/hallgren/eventsourcing v0.8.1
 
 require github.com/hallgren/eventsourcing/core v0.4.0 // indirect
 
-// replace github.com/hallgren/eventsourcing => ../.
+replace github.com/hallgren/eventsourcing => ../.

example/tictactoe/cmd/realtime_events/main.go

@@ -0,0 +1,125 @@
+// This realtime events example demonstrates capturing and analyzing game events such as player moves and results
+// by using save hooks in the event sourcing framework. The program outputs aggregated statistics
+// after running a set of games.
+package main
+
+import (
+	"fmt"
+	"math/rand"
+	"sync"
+	"time"
+
+	"github.com/hallgren/eventsourcing"
+	"github.com/hallgren/eventsourcing/aggregate"
+	"github.com/hallgren/eventsourcing/eventstore/memory"
+	"github.com/hallgren/eventsourcing/example/tictactoe"
+)
+
+func main() {
+	// Buffered channels to receive move and result events
+	chanMoved := make(chan string, 100)
+	chanResult := make(chan string, 100)
+
+	// Aggregated counts of moves and end results
+	moveResult := map[string]int{
+		"XMoved": 0,
+		"OMoved": 0,
+	}
+	endResults := map[string]int{
+		"Draw": 0,
+		"XWon": 0,
+		"OWon": 0,
+	}
+	// In-memory event store
+	es := memory.Create()
+
+	// Register the TicTacToe Game aggregate
+	aggregate.Register(&tictactoe.Game{})
+
+	// Hook to capture final game result events
+	err := aggregate.SetSaveHook(func(events []eventsourcing.Event) {
+		lastEvent := events[len(events)-1]
+		chanResult <- lastEvent.Reason()
+	}, &tictactoe.Game{})
+	if err != nil {
+		panic(err)
+	}
+
+	// Hook to capture move events (XMoved/OMoved)
+	err = aggregate.SetSaveHook(func(events []eventsourcing.Event) {
+		for _, event := range events {
+			switch event.Data().(type) {
+			case *tictactoe.XMoved, *tictactoe.OMoved:
+				chanMoved <- event.Reason()
+			}
+		}
+	}, &tictactoe.Game{})
+	if err != nil {
+		panic(err)
+	}
+	// waitgroup to sync the finish of the async workers
+	wg := sync.WaitGroup{}
+	wg.Add(2)
+
+	// Start async workers for processing move and result events
+	go movesWorker(chanMoved, moveResult, &wg)
+	go resultWorker(chanResult, endResults, &wg)
+
+	// Simulate and save 10 TicTacToe games
+	for i := 0; i < 10; i++ {
+		game := PlayGame()
+		aggregate.Save(es, game)
+	}
+
+	// Close channels to signal no more incoming data
+	close(chanMoved)
+	close(chanResult)
+	fmt.Println("Events are saved, wait for workers to finish.")
+	wg.Wait()
+
+	// Print out aggregated move and result statistics
+	fmt.Println(moveResult, endResults)
+}
+
+func resultWorker(c chan string, m map[string]int, wg *sync.WaitGroup) {
+	for {
+		select {
+		case result, ok := <-c:
+			// no more results
+			if !ok {
+				wg.Done()
+				fmt.Println("results worker finsished")
+				return
+			}
+			time.Sleep(50 * time.Millisecond)
+			m[result]++
+		}
+	}
+}
+
+// do some time consuming operations in the go routine
+func movesWorker(c chan string, m map[string]int, wg *sync.WaitGroup) {
+	for {
+		select {
+		case move, ok := <-c:
+			// no more moves
+			if !ok {
+				wg.Done()
+				fmt.Println("moves worker finsished")
+				return
+			}
+			time.Sleep(50 * time.Millisecond)
+			m[move]++
+		}
+	}
+}
+
+func PlayGame() *tictactoe.Game {
+	game := tictactoe.NewGame()
+	for !game.Done() {
+		x := rand.Intn(3)
+		y := rand.Intn(3)
+		game.PlayMove(x, y)
+	}
+	return game
+}

projections_test.go

@@ -12,7 +12,6 @@ import (
 	"github.com/hallgren/eventsourcing/aggregate"
 	"github.com/hallgren/eventsourcing/core"
 	"github.com/hallgren/eventsourcing/eventstore/memory"
-	"github.com/hallgren/eventsourcing/internal"
 )
 
 // Person aggregate
@@ -386,7 +385,7 @@ func TestErrorFromCallback(t *testing.T) {
 func TestStrict(t *testing.T) {
 	// setup
 	es := memory.Create()
-	internal.ResetRegister()
+	aggregate.ResetRegister()
 
 	// We do not register the Person aggregate with the Born event attached
 	err := createPersonEvent(es, "kalle", 1)