List streams

actions/http_server.go

@@ -113,6 +113,19 @@ func (hs HttpServer) Run(c Context) error {
 	handler.Get("/_check", func(ctx http.GetContext) {
 		ctx.SendJson("OK")
 	})
+	handler.Get("/streams", func(ctx http.GetContext) {
+		session, err := c.Persistence().Session()
+		err = errors.Wrap(err, "Obtaining session failed")
+		if err != nil {
+			c.HandleErr(err)
+			ctx.ServerErr(err)
+			return
+		}
+		streams, err := session.ListStreams()
+		view := make(simplejson.Object)
+		view["streams"] = streams
+		ctx.SendJson(view)
+	})
 	handler.Get("/stream/:id", func(ctx http.GetContext) {
 		stream := ctx.GetSegment("id")
 		action := ReadEvents{Stream: stream}

actions/list_streams.go

@@ -0,0 +1,26 @@
+package actions
+
+import (
+	"fmt"
+
+	"github.com/pkg/errors"
+)
+
+type ListStreams struct {
+}
+
+func (re ListStreams) Run(c Context) error {
+	session, err := c.Persistence().Session()
+	if err != nil {
+		return errors.Wrap(err, "Obtaining session failed")
+	}
+	streams, err := session.ListStreams()
+	for _, s := range streams {
+		println(s)
+	}
+	return errors.Wrap(err, fmt.Sprintf("Error listing streams"))
+}
+
+func (re ListStreams) MarshalJSON() ([]byte, error) {
+	return []byte(`"List streams"`), nil
+}

main.go

@@ -52,6 +52,10 @@ func main() {
 			Value: "",
 			Usage: "Cassandra keyspace replication options",
 		},
+		cli.BoolFlag{
+			Name:  "list-streams",
+			Usage: "List streams in Cassandra",
+		},
 	}
 	app.Action = func(c *cli.Context) error {
 		config := Config{}
@@ -64,6 +68,9 @@ func main() {
 		if c.Bool("migrate") {
 			config.Append(actions.Migrate{})
 		}
+		if c.Bool("list-streams") {
+			config.Append(actions.ListStreams{})
+		}
 		if c.Bool("dev-logger") {
 			config.Features["dev_logger"] = true
 		}

persistence/session.go

@@ -11,6 +11,7 @@ import (
 type Session interface {
 	WriteEvent(string, []byte, map[string]string) error
 	ReadEvents(string, string, int) ([]types.Event, error)
+	ListStreams() ([]string, error)
 	Close()
 }
 
@@ -59,3 +60,16 @@ func (sess *CassandraSession) ReadEvents(stream string, cursor string, amount in
 	err = iter.Close()
 	return events, errors.Wrap(err, "Failed readEvent")
 }
+
+const listStreams = `SELECT DISTINCT stream FROM events`
+
+func (sess *CassandraSession) ListStreams() ([]string, error) {
+	iter := sess.Query(listStreams).Iter()
+	rows, err := iter.SliceMap()
+	streams := make([]string, len(rows))
+	for i, r := range rows {
+		streams[i] = r["stream"].(string)
+	}
+	err = iter.Close()
+	return streams, errors.Wrap(err, "Failed listStreams")
+}