Tests and benchmarks

Quite bad performance:
Transactions:		         754 hits
Availability:		      100.00 %
Elapsed time:		       59.74 secs
Data transferred:	        0.00 MB
Response time:		        1.96 secs
Transaction rate:	       12.62 trans/sec
Throughput:		        0.00 MB/sec
Concurrency:		       24.71
Successful transactions:         754
Failed transactions:	           0
Longest transaction:	        2.31
Shortest transaction:	        1.86

actions/http_server.go

@@ -25,6 +25,7 @@ func (hs HttpServer) Run(c Context) error {
 			c.HandleErr(err)
 			ctx.ServerErr(err)
 		} else {
+			root := make(map[string]interface{})
 			events := make([]map[string]interface{}, len(action.Events))
 			for i, ev := range action.Events {
 				events[i] = make(map[string]interface{})
@@ -37,7 +38,8 @@ func (hs HttpServer) Run(c Context) error {
 				}
 				events[i]["blob"] = payload
 			}
-			ctx.SendJson(events)
+			root["results"] = events
+			ctx.SendJson(root)
 		}
 	})
 	handler.Post("/stream/:id", func(ctx http.PostContext) {
@@ -65,6 +67,10 @@ func (hs HttpServer) Run(c Context) error {
 			ctx.SendJson("OK")
 		}
 	})
+
 	connString := fmt.Sprintf("%s:%d", hs.Addr, hs.Port)
 	return errors.Wrap(handler.Run(connString), "HttpServer starting..")
 }
+
+func (hs HttpServer) Stop() {
+}

actions_test.go

@@ -0,0 +1,99 @@
+package main_test
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"io"
+	"io/ioutil"
+	"net/http"
+
+	. "github.com/scoiatael/archai"
+	"github.com/scoiatael/archai/actions"
+	"github.com/scoiatael/archai/util"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+)
+
+const testingKeyspace = "archai_test"
+
+var (
+	config Config
+)
+
+var _ = BeforeSuite(func() {
+	config = Config{}
+	config.Hosts = []string{"127.0.0.1"}
+	config.Keyspace = testingKeyspace
+})
+
+var _ = AfterSuite(func() {
+
+})
+
+var _ = Describe("Actions", func() {
+	Describe("HttpServer", func() {
+		var (
+			action  actions.HttpServer
+			port    int
+			address string
+			stream  string
+			buf     io.Reader
+		)
+		port = 9080
+		BeforeEach(func() {
+			action.Addr = "127.0.0.1"
+			port = port + 1 + util.RandomInt(1000)
+			action.Port = port
+		})
+		JustBeforeEach(func() {
+			go action.Run(config)
+			address = fmt.Sprintf("http://127.0.0.1:%d", action.Port)
+		})
+
+		AfterEach(func() {
+			action.Stop()
+		})
+
+		Describe("/stream/:id", func() {
+			JustBeforeEach(func() {
+				stream = util.RandomString(10)
+				address = fmt.Sprintf("%s/stream/%s", address, stream)
+				buf = bytes.NewBufferString(`{ "foo": "bar" }`)
+			})
+
+			It("allows writing events", func() {
+				resp, err := http.Post(address, "application/json", buf)
+
+				Expect(err).NotTo(HaveOccurred())
+				body, err := ioutil.ReadAll(resp.Body)
+				Expect(err).NotTo(HaveOccurred())
+				Expect(string(body)).To(Equal(`"OK"`))
+			})
+
+			Context("After some event was written", func() {
+				JustBeforeEach(func() {
+					event := actions.WriteEvent{}
+					event.Stream = stream
+					event.Payload = []byte(`{ "foo": "bar"}`)
+					event.Meta = make(map[string]string)
+
+					event.Run(config)
+
+				})
+				It("allows reading events", func() {
+					resp, err := http.Get(address)
+
+					Expect(err).NotTo(HaveOccurred())
+					body, err := ioutil.ReadAll(resp.Body)
+					Expect(err).NotTo(HaveOccurred())
+					js := make(map[string]interface{})
+					err = json.Unmarshal(body, &js)
+					Expect(err).NotTo(HaveOccurred())
+					Expect(js["results"]).NotTo(BeEmpty())
+				})
+			})
+		})
+	})
+})

http/iris.go

@@ -48,7 +48,11 @@ func (hc IrisPostContext) JsonBodyParams() (map[string]interface{}, error) {
 	err := hc.ReadJSON(&sess)
 
 	if err != nil {
-		hc.JSON(iris.StatusBadRequest, iris.Map{"error": "expected JSON body"})
+		hc.JSON(iris.StatusBadRequest,
+			iris.Map{"error": "expected JSON body",
+				"details":  err,
+				"received": hc.Request.Body,
+			})
 	}
 	return sess, err
 }

persistence/persistence_suite_test.go

@@ -0,0 +1,13 @@
+package persistence_test
+
+import (
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+
+	"testing"
+)
+
+func TestPersistence(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "Persistence Suite")
+}

persistence_test.go → persistence/persistence_test.go

@@ -1,13 +1,11 @@
-package main_test
+package persistence_test
 
 import (
 	"fmt"
-	"math/rand"
-	"time"
 
-	. "github.com/scoiatael/archai"
 	. "github.com/scoiatael/archai/persistence"
 	"github.com/scoiatael/archai/types"
+	"github.com/scoiatael/archai/util"
 
 	"github.com/gocql/gocql"
 	. "github.com/onsi/ginkgo"
@@ -19,7 +17,6 @@ const testingKeyspace = "archai_test"
 const findKeyspace = `select keyspace_name from system_schema.keyspaces where keyspace_name = ?`
 
 var (
-	config    Config
 	provider  CassandraProvider
 	root_sess *gocql.Session
 )
@@ -50,21 +47,13 @@ func migrationTableExists() (bool, error) {
 	return exists, err
 }
 
-const letters = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ"
-
 func randomString() string {
-	b := make([]byte, 10)
-	for i := range b {
-		b[i] = letters[rand.Intn(len(letters))]
-	}
-	return string(b)
+	return util.RandomString(10)
 }
 
 var _ = BeforeSuite(func() {
 	var err error
-	rand.Seed(time.Now().UnixNano())
-	config = Config{Keyspace: testingKeyspace, Hosts: []string{"127.0.0.1"}}
-	provider = CassandraProvider{Hosts: config.Hosts, Keyspace: config.Keyspace}
+	provider = CassandraProvider{Hosts: []string{"127.0.0.1"}, Keyspace: testingKeyspace}
 	cluster := provider.NewCluster()
 	cluster.Consistency = gocql.All
 	root_sess, err = cluster.CreateSession()
@@ -156,7 +145,7 @@ var _ = Describe("Persistence", func() {
 		})
 	})
 
-	Describe("ReadEvents", func() {
+	Describe("ReadEvents & WriteEvent", func() {
 		var (
 			sess Session
 		)
@@ -218,5 +207,32 @@ var _ = Describe("Persistence", func() {
 				})
 			})
 		})
+		Describe("WriteEvent", func() {
+			Measure("small events", func(b Benchmarker) {
+				var (
+					err    error
+					stream string
+				)
+				b.Time("", func() {
+					stream = randomString()
+					err = sess.WriteEvent(stream, []byte(`{ "a": 1 }`), make(map[string]string))
+					Expect(err).NotTo(HaveOccurred())
+				})
+			}, 20)
+
+			Measure("big events", func(b Benchmarker) {
+				var (
+					err    error
+					stream string
+					blob   []byte
+				)
+				blob = make([]byte, 1024*1024)
+				b.Time("", func() {
+					stream = randomString()
+					err = sess.WriteEvent(stream, blob, make(map[string]string))
+					Expect(err).NotTo(HaveOccurred())
+				})
+			}, 1)
+		})
 	})
 })

scripts/siege.sh

@@ -0,0 +1,3 @@
+#!/usr/bin/env bash
+
+siege -t60S -b -H 'Content-Type: application/json' "http://localhost:8080/stream/siege-test POST < $(dirname $0)/siege_data.json"

scripts/siege_data.json

@@ -0,0 +1,16 @@
+{
+  "payload": {
+    "post_id": "491452930867938_1095485790464646",
+    "date": "2015-09-03T02:30:01+0000",
+    "field1": "19565815",
+    "field2": "186716",
+    "field3": "38213",
+    "field4": "302554",
+    "field5": "15.033",
+    "field7": "Egg-In-The-Hole Grilled Cheese",
+    "post_url": "https://www.facebook.com/BuzzFeedFood/videos/1095485790464646/",
+    "thumbnail_url": "https://fbcdn-vthumb-a.akamaihd.net/hvthumb-ak-xpf1/v/t15.0-10/p130x130/11838221_1095489770464248_1585269284_n.jpg?oh=f132b8b6b3d3d613e274f94c570fe98c&oe=56893684&__gda__=1456401596_c9408cc1e7995d6d454bcdc14e7a1729",
+    "video_url": "https://video.xx.fbcdn.net/hvideo-xfp1/v/t43.1792-2/11961942_1095489660464259_581814512_n.mp4?efg=eyJybHIiOjE1MDAsInJsYSI6MTAyNCwidmVuY29kZV90YWciOiJoZCJ9&rl=1500&vabr=505&oh=fb028b89baa62dd9b55e6c2a2842c250&oe=5617530A"
+  },
+  "stream": "BuzzThis"
+}

util/random_int.go

@@ -0,0 +1,11 @@
+package util
+
+import (
+	"math/rand"
+	"time"
+)
+
+func RandomInt(max int) int {
+	rand.Seed(time.Now().UnixNano())
+	return rand.Intn(max)
+}

util/random_string.go

@@ -0,0 +1,17 @@
+package util
+
+import (
+	"math/rand"
+	"time"
+)
+
+const letters = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ"
+
+func RandomString(length int) string {
+	rand.Seed(time.Now().UnixNano())
+	b := make([]byte, length)
+	for i := range b {
+		b[i] = letters[rand.Intn(len(letters))]
+	}
+	return string(b)
+}