SCFF-30 unit test for StringBuilder method in Appender

Author: nmillalaf

eventQueue/eventQueue.go

@@ -6,53 +6,53 @@ import (
 
 //Node to put in queue
 type Node struct {
-	event Event
+	Event Event
 }
 
 // Queue is a basic FIFO queue based on a circular list that resizes as needed.
 type Queue struct {
-	nodes []*Node
+	Nodes []*Node
 	head  int
 	tail  int
 	count int
 }
 
 func NewNode(event Event) *Node {
 	return &Node{
-		event: event,
+		Event: event,
 	}
 }
 
 func NewQueue(n []*Node) *Queue {
 	return &Queue{
-		nodes: n,
+		Nodes: n,
 	}
 }
 
 func (q *Queue) GetNode() []*Node {
-	return q.nodes
+	return q.Nodes
 }
 
 func (n *Queue) GetCount() int {
 	return n.count
 }
 
 func (n *Node) GetNodeEvent() Event {
-	return n.event
+	return n.Event
 }
 
 // Push adds a node to the queue.
 func (q *Queue) Push(n *Node) {
 	if q.head == q.tail && q.count > 0 {
-		nodes := make([]*Node, len(q.nodes)*2)
-		copy(nodes, q.nodes[q.head:])
-		copy(nodes[len(q.nodes)-q.head:], q.nodes[:q.head])
+		nodes := make([]*Node, len(q.Nodes)*2)
+		copy(nodes, q.Nodes[q.head:])
+		copy(nodes[len(q.Nodes)-q.head:], q.Nodes[:q.head])
 		q.head = 0
-		q.tail = len(q.nodes)
-		q.nodes = nodes
+		q.tail = len(q.Nodes)
+		q.Nodes = nodes
 	}
-	q.nodes[q.tail] = n
-	q.tail = (q.tail + 1) % len(q.nodes)
+	q.Nodes[q.tail] = n
+	q.tail = (q.tail + 1) % len(q.Nodes)
 	q.count++
 }
 
@@ -61,8 +61,8 @@ func (q *Queue) Pop() *Node {
 	if q.count == 0 {
 		return nil
 	}
-	node := q.nodes[q.head]
-	q.head = (q.head + 1) % len(q.nodes)
+	node := q.Nodes[q.head]
+	q.head = (q.head + 1) % len(q.Nodes)
 	q.count--
 	return node
 }

eventQueue/eventQueue_test.go

@@ -10,7 +10,7 @@ import (
 func TestQueueFIFO(t *testing.T) {
 
 	node1 := Node{
-		event: Event{
+		Event: Event{
 			Fields: map[string]interface{}{
 				"message_type": "OUT",
 				"cf_app_id":    "011",
@@ -19,7 +19,7 @@ func TestQueueFIFO(t *testing.T) {
 		},
 	}
 	node2 := Node{
-		event: Event{
+		Event: Event{
 			Fields: map[string]interface{}{
 				"message_type": "OUT",
 				"cf_app_id":    "022",
@@ -28,7 +28,7 @@ func TestQueueFIFO(t *testing.T) {
 		},
 	}
 	node3 := Node{
-		event: Event{
+		Event: Event{
 			Fields: map[string]interface{}{
 				"message_type": "OUT",
 				"cf_app_id":    "033",
@@ -38,14 +38,14 @@ func TestQueueFIFO(t *testing.T) {
 	}
 
 	queue := Queue{
-		nodes: make([]*Node, 3),
+		Nodes: make([]*Node, 3),
 	}
 
 	queue.Push(&node1)
 	queue.Push(&node2)
 	queue.Push(&node3)
 
-	assert.Equal(t, queue.Pop().event.Msg, "index [01]", "")
-	assert.Equal(t, queue.Pop().event.Msg, "index [02]", "")
-	assert.Equal(t, queue.Pop().event.Msg, "index [03]", "")
+	assert.Equal(t, queue.Pop().Event.Msg, "index [01]", "")
+	assert.Equal(t, queue.Pop().Event.Msg, "index [02]", "")
+	assert.Equal(t, queue.Pop().Event.Msg, "index [03]", "")
 }

main.go

@@ -28,7 +28,7 @@ var (
 	wantedEvents      = kingpin.Flag("events", fmt.Sprintf("Comma separated list of events you would like. Valid options are %s", eventRouting.GetListAuthorizedEventEvents())).Default("LogMessage").OverrideDefaultFromEnvar("EVENTS").String()
 	boltDatabasePath  = "my.db"                   //default
 	tickerTime, errT  = time.ParseDuration("60s") //Default
-	eventsAmount      = kingpin.Flag("event-amount", "Events amount").Int()
+	eventsBatch       = kingpin.Flag("event-amount", "Events amount").Int()
 )
 
 var (
@@ -42,10 +42,6 @@ func main() {
 	fmt.Println("this is the sumo endpoint")
 	fmt.Println(sumoEndpoint)
 
-	//Creating queue
-	queue := eventQueue.NewQueue(make([]*eventQueue.Node, 100))
-	loggingClientSumo := sumoCFFirehose.NewSumoLogicAppender(*sumoEndpoint, 1000, *queue, *eventsAmount)
-
 	fmt.Printf("Starting firehose-to-sumo %s \n", version)
 
 	c := cfclient.Config{
@@ -68,6 +64,10 @@ func main() {
 		cachingClient = caching.NewCachingEmpty()
 	}
 
+	//Creating queue
+	queue := eventQueue.NewQueue(make([]*eventQueue.Node, 100))
+	loggingClientSumo := sumoCFFirehose.NewSumoLogicAppender(*sumoEndpoint, 1000, *queue, *eventsBatch)
+
 	//Creating Events
 	events := eventRouting.NewEventRouting(cachingClient, *loggingClientSumo, *queue)
 	err := events.SetupEventRouting(*wantedEvents)

sumoCFFirehose/sumoLogicAppender.go

@@ -15,16 +15,16 @@ type SumoLogicAppender struct {
 	connectionTimeout int //10000
 	httpClient        http.Client
 	nozzleQueue       eventQueue.Queue
-	eventsAmount      int
+	eventsBatch       int
 }
 
-func NewSumoLogicAppender(urlValue string, connectionTimeoutValue int, nozzleQueue eventQueue.Queue, eventsAmount int) *SumoLogicAppender {
+func NewSumoLogicAppender(urlValue string, connectionTimeoutValue int, nozzleQueue eventQueue.Queue, eventsBatch int) *SumoLogicAppender {
 	return &SumoLogicAppender{
 		url:               urlValue,
 		connectionTimeout: connectionTimeoutValue,
 		httpClient:        http.Client{Timeout: time.Duration(connectionTimeoutValue * int(time.Millisecond))},
 		nozzleQueue:       nozzleQueue,
-		eventsAmount:      eventsAmount,
+		eventsBatch:       eventsBatch,
 	}
 }
 
@@ -45,22 +45,30 @@ func (s *SumoLogicAppender) Connect() bool {
 	return success
 }
 
-func (s *SumoLogicAppender) AppendLogs() {
-	fmt.Println("i'm in appendLogs")
-	// the appender calls for the next message in the queue and parse it to a string
+func StringBuilder(queue eventQueue.Queue) string {
 	buf := new(bytes.Buffer)
-	for i := 0; i <= s.eventsAmount; i++ { //Pop eventsAmount from queue
-		event := s.nozzleQueue.Pop().GetNodeEvent()
+	for queue.GetCount() > 0 { //Pop eventsBatch from queue
+		event := queue.Pop().GetNodeEvent()
 		if event.Fields["message_type"] == nil {
-			return
+			return ""
 		}
 		if event.Fields["message_type"] == "" {
-			return
+			return ""
 		}
 		message := time.Unix(0, event.Fields["timestamp"].(int64)*int64(time.Nanosecond)).String() + "\t" + event.Fields["message_type"].(string) + "\t" + event.Msg + "\n"
 		buf.WriteString(message)
 	}
-	s.SendToSumo(buf.String())
+
+	return buf.String()
+}
+
+func (s *SumoLogicAppender) AppendLogs(queue eventQueue.Queue) {
+	// the appender calls for the next message in the queue and parse it to a string
+	if queue.GetCount() == s.eventsBatch { //whent the batch limit is met, call stringBuilder
+		logMessage := StringBuilder(queue)
+		s.SendToSumo(logMessage)
+	}
+
 }
 
 func (s *SumoLogicAppender) SendToSumo(log string) {

sumoCFFirehose/sumoLogicAppender_test.go

@@ -0,0 +1,56 @@
+package sumoCFFirehose
+
+import (
+	"testing"
+
+	. "bitbucket.org/mcplusa-ondemand/firehose-to-sumologic/eventQueue"
+	. "bitbucket.org/mcplusa-ondemand/firehose-to-sumologic/events"
+	"github.com/stretchr/testify/assert"
+)
+
+func testAppenderStringBuilder(t *testing.T) {
+
+	node1 := Node{
+		Event: Event{
+			Fields: map[string]interface{}{
+				"timestamp":    "1481569361828366387",
+				"message_type": "OUT",
+				"cf_app_id":    "011",
+			},
+			Msg: "index [01]",
+		},
+	}
+	node2 := Node{
+		Event: Event{
+			Fields: map[string]interface{}{
+				"timestamp":    "1481569362844737993",
+				"message_type": "OUT",
+				"cf_app_id":    "022",
+			},
+			Msg: "index [02]",
+		},
+	}
+	node3 := Node{
+		Event: Event{
+			Fields: map[string]interface{}{
+				"timestamp":    "1481569363862436654",
+				"message_type": "OUT",
+				"cf_app_id":    "033",
+			},
+			Msg: "index [03]",
+		},
+	}
+
+	queue := Queue{
+		Nodes: make([]*Node, 3),
+	}
+	queue.Push(&node1)
+	queue.Push(&node2)
+	queue.Push(&node3)
+
+	finalString := StringBuilder(queue)
+	assert.Equal(t, finalString, "2016-12-12 16:02:41.828366387 -0300 CLST"+"\t"+"OUT"+"\t"+"index [01]"+"\n"+
+		"2016-12-12 16:02:42.844737993 -0300 CLST"+"\t"+"OUT"+"\t"+"index [02]"+"\n"+
+		"2016-12-12 16:02:43.862436654 -0300 CLST"+"\t"+"OUT"+"\t"+"index [03]"+"\n", "")
+
+}