SCFF-6 added the modifications regarding the test and Event handling

Author: nmillalaf

eventQueue/eventQueue.go

@@ -0,0 +1,64 @@
+package eventQueue
+
+import (
+	. "bitbucket.org/mcplusa-ondemand/firehose-to-sumologic/events"
+)
+
+//Node to put in queue
+type Node struct {
+	event Event
+}
+
+// Queue is a basic FIFO queue based on a circular list that resizes as needed.
+type Queue struct {
+	nodes []*Node
+	head  int
+	tail  int
+	count int
+}
+
+func NewNode(event Event) *Node {
+	return &Node{
+		event: event,
+	}
+}
+
+func NewQueue(n []*Node) *Queue {
+	return &Queue{
+		nodes: n,
+	}
+}
+
+func (q *Queue) GetNode() []*Node {
+	return q.nodes
+}
+
+func (n *Node) GetNodeEvent() 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])
+		q.head = 0
+		q.tail = len(q.nodes)
+		q.nodes = nodes
+	}
+	q.nodes[q.tail] = n
+	q.tail = (q.tail + 1) % len(q.nodes)
+	q.count++
+}
+
+// Pop removes and returns a node from the queue in first to last order.
+func (q *Queue) Pop() *Node {
+	if q.count == 0 {
+		return nil
+	}
+	node := q.nodes[q.head]
+	q.head = (q.head + 1) % len(q.nodes)
+	q.count--
+	return node
+}

eventQueue/eventQueue_test.go

@@ -0,0 +1,51 @@
+package eventQueue
+
+import (
+	"testing"
+
+	. "bitbucket.org/mcplusa-ondemand/firehose-to-sumologic/events"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestQueueFIFO(t *testing.T) {
+
+	node1 := Node{
+		event: Event{
+			Fields: map[string]interface{}{
+				"message_type": "OUT",
+				"cf_app_id":    "011",
+			},
+			Msg: "index [01]",
+		},
+	}
+	node2 := Node{
+		event: Event{
+			Fields: map[string]interface{}{
+				"message_type": "OUT",
+				"cf_app_id":    "022",
+			},
+			Msg: "index [02]",
+		},
+	}
+	node3 := Node{
+		event: Event{
+			Fields: map[string]interface{}{
+				"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)
+
+	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]", "")
+}

eventRouting/eventRouting_test.go

@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"bitbucket.org/mcplusa-ondemand/firehose-to-sumologic/caching"
+	"bitbucket.org/mcplusa-ondemand/firehose-to-sumologic/eventQueue"
 	fevents "bitbucket.org/mcplusa-ondemand/firehose-to-sumologic/events"
 	"bitbucket.org/mcplusa-ondemand/firehose-to-sumologic/sumoCFFirehose"
 	"github.com/Sirupsen/logrus"
@@ -19,18 +20,18 @@ type EventRouting struct {
 	selectedEvents      map[string]bool
 	selectedEventsCount map[string]uint64
 	mutex               *sync.Mutex
-	sLAppender          sumoCFFirehose.SumoCFFirehose //**
-	//*log                 logging.Logging
+	sLAppender          sumoCFFirehose.SumoLogicAppender //**
+	queue               eventQueue.Queue
 }
 
-func NewEventRouting(caching caching.Caching, sLAppender sumoCFFirehose.SumoCFFirehose) *EventRouting {
+func NewEventRouting(caching caching.Caching, sLAppender sumoCFFirehose.SumoLogicAppender, queue eventQueue.Queue) *EventRouting {
 	return &EventRouting{
 		CachingClient:       caching,
 		selectedEvents:      make(map[string]bool),
 		selectedEventsCount: make(map[string]uint64),
 		sLAppender:          sLAppender, //**
-		//*		log:                 logging,
-		mutex: &sync.Mutex{},
+		queue:               queue,
+		mutex:               &sync.Mutex{},
 	}
 }
 
@@ -74,9 +75,14 @@ func (e *EventRouting) RouteEvent(msg *events.Envelope) {
 		if ignored, hasIgnoredField := event.Fields["cf_ignored_app"]; ignored == true && hasIgnoredField {
 			e.selectedEventsCount["ignored_app_message"]++
 		} else {
-			/*fmt.Println("This is the message field")
-			fmt.Println(event.Msg)*/
-			e.sLAppender.AppendLogs(event.Fields, event.Msg) //**/here we have to change the method for the one on sumoLogicAppender
+			if e.queue.Pop() != nil { //if the queue is not empty
+				fmt.Println("sendig event from queue to appender")
+				e.sLAppender.AppendLogs(e.queue.Pop().GetNodeEvent()) // send to appender event from queue
+			} else { //if the queue is empty, send the event to queue
+				fmt.Println(event.Msg)
+				fmt.Println("sendig event TO queue")
+				e.queue.Push(eventQueue.NewNode(*event))
+			}
 			e.selectedEventsCount[eventType.String()]++
 
 		}
@@ -145,7 +151,7 @@ func (e *EventRouting) LogEventTotals(logTotalsTime time.Duration) {
 			event, lastCount := e.getEventTotals(totalElapsedTime, elapsedTime, count)
 			count = lastCount
 			//*e.log.ShipEvents(event.Fields, event.Msg)
-			e.sLAppender.AppendLogs(event.Fields, event.Msg)
+			e.sLAppender.AppendLogs(*event)
 		}
 	}()
 }

eventRouting/eventrouting.go

@@ -1,10 +1,11 @@
 package events_test
 
 import (
+	"testing"
+
 	. "github.com/cloudfoundry/sonde-go/events"
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
-	"testing"
 )
 
 func TestEvents(t *testing.T) {

events/events_suite_test.go

@@ -7,6 +7,7 @@ import (
 	"time"
 
 	"bitbucket.org/mcplusa-ondemand/firehose-to-sumologic/caching"
+	"bitbucket.org/mcplusa-ondemand/firehose-to-sumologic/eventQueue"
 	"bitbucket.org/mcplusa-ondemand/firehose-to-sumologic/eventRouting"
 	"bitbucket.org/mcplusa-ondemand/firehose-to-sumologic/firehoseclient"
 	"bitbucket.org/mcplusa-ondemand/firehose-to-sumologic/sumoCFFirehose"
@@ -62,8 +63,12 @@ func main() {
 	} else {
 		cachingClient = caching.NewCachingEmpty()
 	}
+
+	//Creating queue
+	queue := eventQueue.NewQueue(make([]*eventQueue.Node, 100))
+
 	//Creating Events
-	events := eventRouting.NewEventRouting(cachingClient, loggingClientSumo)
+	events := eventRouting.NewEventRouting(cachingClient, *loggingClientSumo, *queue)
 	err := events.SetupEventRouting(*wantedEvents)
 	if err != nil {
 		log.Fatal("Error setting up event routing: ", err)
@@ -91,6 +96,7 @@ func main() {
 	if /*loggingClientSumo.Connect() ||*/ debug {
 
 		fmt.Printf("Connected to Server! Connecting to Firehose... \n")
+
 		firehoseClient := firehoseclient.NewFirehoseNozzle(cfClient, events, firehoseConfig)
 		err = firehoseClient.Start()
 		fmt.Printf("I created the Firehose... \n")