diff --git a/README.md b/README.md index 3765ff49..606b51c9 100644 --- a/README.md +++ b/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 diff --git a/aggregate/aggregate.go b/aggregate/aggregate.go index 8d9222ea..adc9842d 100644 --- a/aggregate/aggregate.go +++ b/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 diff --git a/aggregate/aggregate_test.go b/aggregate/aggregate_test.go index 9e06e0fa..c9a836d9 100644 --- a/aggregate/aggregate_test.go +++ b/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()) + } +} diff --git a/eventsourcing.go b/eventsourcing.go index 1f7d5766..ae1a71b1 100644 --- a/eventsourcing.go +++ b/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 diff --git a/example/go.mod b/example/go.mod index 6c7081d0..843b4eb7 100644 --- a/example/go.mod +++ b/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 => ../. diff --git a/example/tictactoe/cmd/realtime_events/main.go b/example/tictactoe/cmd/realtime_events/main.go new file mode 100644 index 00000000..ee347b18 --- /dev/null +++ b/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 +} diff --git a/projections_test.go b/projections_test.go index 4b1980f3..a1454c00 100644 --- a/projections_test.go +++ b/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)