Better reader structure

actions/read_event.go → actions/read_events.go

@@ -4,17 +4,20 @@ import (
 	"fmt"
 
 	"github.com/pkg/errors"
+	"github.com/scoiatael/archai/persistence"
 )
 
-type ReadEvent struct {
+type ReadEvents struct {
 	Stream string
 	Cursor string
 	Amount int
+
+	Output chan (persistence.Event)
 }
 
 const minTimeuuid = "00000000-0000-1000-8080-808080808080"
 
-func (re ReadEvent) Run(c Context) error {
+func (re ReadEvents) Run(c Context) error {
 	persistenceProvider := c.Persistence()
 	session, err := persistenceProvider.Session()
 	if err != nil {
@@ -25,8 +28,7 @@ func (re ReadEvent) Run(c Context) error {
 	}
 	events, err := session.ReadEvents(re.Stream, re.Cursor, re.Amount)
 	for _, ev := range events {
-		js := string(ev.Blob)
-		fmt.Printf("%s - %s: {%v} %s\n", ev.Stream, ev.ID, ev.Meta, js)
+		re.Output <- ev
 	}
 	return errors.Wrap(err, fmt.Sprintf("Error reading event from stream %s", re.Stream))
 }

actions/read_events_to_stream.go

@@ -0,0 +1,38 @@
+package actions
+
+import (
+	"fmt"
+	"io"
+
+	"github.com/scoiatael/archai/persistence"
+)
+
+type ReadEventsToStream struct {
+	Stream string
+	Cursor string
+	Amount int
+
+	Output io.Writer
+}
+
+func printEventToStream(out io.Writer, ev persistence.Event) error {
+	js := string(ev.Blob)
+	str := fmt.Sprintf("%s - %s: {%v} %s\n", ev.Stream, ev.ID, ev.Meta, js)
+	_, err := out.Write([]byte(str))
+	return err
+}
+
+func (res ReadEventsToStream) Run(c Context) error {
+	ch := make(chan (persistence.Event), 100)
+	err := ReadEvents{Stream: res.Stream, Cursor: res.Cursor, Amount: 10, Output: ch}.Run(c)
+	if err != nil {
+		return err
+	}
+	for ev := range ch {
+		err := printEventToStream(res.Output, ev)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}

main.go

@@ -14,7 +14,7 @@ func main() {
 	app.Version = Version
 	app.Action = func(c *cli.Context) error {
 		config := Config{keyspace: "archai_test"}
-		action := actions.ReadEvent{Stream: "testing-stream", Amount: 10}
+		action := actions.ReadEventsToStream{Stream: "testing-stream", Output: os.Stdout}
 		err := action.Run(config)
 		return err
 	}