77
88 "github.com/get-eventually/go-eventually"
99 "github.com/get-eventually/go-eventually/eventstore"
10+ "github.com/get-eventually/go-eventually/eventstore/stream"
1011)
1112
1213var _ eventstore.Store = & EventStore {}
@@ -28,21 +29,7 @@ func NewEventStore() *EventStore {
2829 }
2930}
3031
31- // StreamAll streams all Event Store committed events onto the provided EventStream,
32- // from the specified Global Sequence Number in `from`.
33- //
34- // Note: this call is synchronous, and will return when all the Events
35- // have been successfully written to the provided EventStream, or when
36- // the context has been canceled.
37- //
38- // An error is returned if one Event in the Event Store does not have a
39- // Global Sequence Number (which should never happen), or when the context
40- // is done.
41- func (s * EventStore ) StreamAll (ctx context.Context , es eventstore.EventStream , selectt eventstore.Select ) error {
42- s .mx .RLock ()
43- defer s .mx .RUnlock ()
44- defer close (es )
45-
32+ func (s * EventStore ) streamAll (ctx context.Context , es eventstore.EventStream , selectt eventstore.Select ) error {
4633 for _ , event := range s .events {
4734 if event .SequenceNumber < selectt .From {
4835 continue
@@ -58,26 +45,12 @@ func (s *EventStore) StreamAll(ctx context.Context, es eventstore.EventStream, s
5845 return nil
5946}
6047
61- // StreamByType streams all Event Store committed events of a specific Type (or Category)
62- // onto the provided EventStream, from the specified Global Sequence Number in `from`.
63- //
64- // Note: this call is synchronous, and will return when all the Events
65- // have been successfully written to the provided EventStream, or when
66- // the context has been canceled.
67- //
68- // An error is returned if one Event in the Event Store does not have a
69- // Global Sequence Number (which should never happen), or when the context
70- // is done.
71- func (s * EventStore ) StreamByType (
48+ func (s * EventStore ) streamByType (
7249 ctx context.Context ,
7350 es eventstore.EventStream ,
7451 typ string ,
7552 selectt eventstore.Select ,
7653) error {
77- s .mx .RLock ()
78- defer s .mx .RUnlock ()
79- defer close (es )
80-
8154 for _ , eventIdx := range s .byType [typ ] {
8255 event := s .events [eventIdx ]
8356
@@ -95,24 +68,42 @@ func (s *EventStore) StreamByType(
9568 return nil
9669}
9770
98- // Stream streams all Domain Events committed in a specific event stream
99- // onto the provided EventStream channel, from the specified version in `from`.
100- //
101- // Note: this call is synchronous, and will return when all the Events
102- // have been successfully written to the provided EventStream, or when
103- // the context has been canceled.
104- //
105- // An error is returned when the context is done.
106- func (s * EventStore ) Stream (
71+ func (s * EventStore ) streamByTypes (
10772 ctx context.Context ,
10873 es eventstore.EventStream ,
109- id eventstore. StreamID ,
74+ types [] string ,
11075 selectt eventstore.Select ,
11176) error {
112- s .mx .RLock ()
113- defer s .mx .RUnlock ()
114- defer close (es )
77+ indexedTypes := make (map [string ]struct {})
78+ for _ , typ := range types {
79+ indexedTypes [typ ] = struct {}{}
80+ }
81+
82+ for _ , event := range s .events {
83+ if event .SequenceNumber < selectt .From {
84+ continue
85+ }
11586
87+ if _ , ok := indexedTypes [event .Stream .Type ]; ! ok {
88+ continue
89+ }
90+
91+ select {
92+ case es <- event :
93+ case <- ctx .Done ():
94+ return contextErr (ctx )
95+ }
96+ }
97+
98+ return nil
99+ }
100+
101+ func (s * EventStore ) streamByID (
102+ ctx context.Context ,
103+ es eventstore.EventStream ,
104+ id stream.ID ,
105+ selectt eventstore.Select ,
106+ ) error {
116107 if m , ok := s .byTypeAndInstance [id .Type ]; ! ok || m == nil {
117108 return nil
118109 }
@@ -139,6 +130,38 @@ func (s *EventStore) Stream(
139130 return nil
140131}
141132
133+ // Stream streams committed events in the Event Store onto the provided EventStream,
134+ // from the specified Global Sequence Number in `from`, based on the provided stream.Target.
135+ //
136+ // Note: this call is synchronous, and will return when all the Events
137+ // have been successfully written to the provided EventStream, or when
138+ // the context has been canceled.
139+ //
140+ // This method fails only when the context is canceled.
141+ func (s * EventStore ) Stream (
142+ ctx context.Context ,
143+ es eventstore.EventStream ,
144+ target stream.Target ,
145+ selectt eventstore.Select ,
146+ ) error {
147+ s .mx .RLock ()
148+ defer s .mx .RUnlock ()
149+ defer close (es )
150+
151+ switch t := target .(type ) {
152+ case stream.All :
153+ return s .streamAll (ctx , es , selectt )
154+ case stream.ByType :
155+ return s .streamByType (ctx , es , string (t ), selectt )
156+ case stream.ByTypes :
157+ return s .streamByTypes (ctx , es , []string (t ), selectt )
158+ case stream.ByID :
159+ return s .streamByID (ctx , es , stream .ID (t ), selectt )
160+ default :
161+ return fmt .Errorf ("inmemory.EventStore: unsupported stream target type provided: %T" , t )
162+ }
163+ }
164+
142165// Append inserts the specified Domain Events into the Event Stream specified
143166// by the current instance, returning the new version of the Event Stream.
144167//
@@ -153,7 +176,7 @@ func (s *EventStore) Stream(
153176// version check fails against the current version of the Event Stream.
154177func (s * EventStore ) Append (
155178 ctx context.Context ,
156- id eventstore. StreamID ,
179+ id stream. ID ,
157180 expected eventstore.VersionCheck ,
158181 events ... eventually.Event ,
159182) (int64 , error ) {
0 commit comments