aggregate data before write

Author: hosseinfakhari

vmetrics/message.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"reflect"
+	"strconv"
 	"time"
 )
 
@@ -20,6 +21,10 @@ func NewMessage() *Message {
 
 func toMap(intf interface{}) (map[string]interface{}, error) {
 	var m map[string]interface{}
+	kind := reflect.ValueOf(intf).Kind()
+	if kind != reflect.Struct && kind != reflect.Ptr {
+		return m, errors.New("message should be type of struct: " + reflect.ValueOf(intf).Kind().String())
+	}
 	b, err := json.Marshal(&intf)
 	if err != nil {
 		return m, err
@@ -29,10 +34,6 @@ func toMap(intf interface{}) (map[string]interface{}, error) {
 }
 
 func (m *Message) jsonString(inf interface{}) (string, error) {
-	kind := reflect.ValueOf(inf).Kind()
-	if kind != reflect.Struct && kind != reflect.Ptr {
-		return "", errors.New("message should be type of struct: " + reflect.ValueOf(inf).Kind().String())
-	}
 	infMap, err := toMap(inf)
 	if err != nil {
 		return "", errors.New("message should be type of struct: " + reflect.ValueOf(inf).Kind().String())
@@ -62,6 +63,32 @@ func (m *Message) Consume() []string {
 	return messages
 }
 
+func (m *Message) Aggregated() []string {
+	var messages []string
+	var uniqueMap = make(map[string]int64)
+	for _, msg := range m.messages {
+		infMap, err := toMap(msg)
+		if err != nil {
+			fmt.Println(err)
+		} else {
+			var kstr = ""
+			for k := range infMap {
+				kstr += k + "_"
+			}
+			if val, ok := uniqueMap[kstr]; ok {
+				uniqueMap[kstr] = val + 1
+			} else {
+				uniqueMap[kstr] = 1
+			}
+		}
+	}
+	for msg := range uniqueMap {
+		count := strconv.Itoa(int(uniqueMap[msg]))
+		messages = append(messages, "{ \"key\": \""+msg+"\" \"count\": "+count+"}")
+	}
+	return messages
+}
+
 func (m *Message) Clear() {
 	m.messages = nil
 	m.messages = make([]interface{}, 0)

vmetrics/metrics.go

@@ -3,5 +3,6 @@ package vmetrics
 type Metric interface {
 	Record(inf interface{})
 	Consume() []string
+	Aggregated() []string
 	Clear()
 }

vmetrics/registry.go

@@ -81,8 +81,9 @@ func (r *Registry) Start() {
 	go func() {
 		for {
 			for _, metric := range r.Metrics {
-				messages := metric.Consume()
-				go r.writeToKafka(messages)
+				//messages := metric.Consume()
+				agMessages := metric.Aggregated()
+				go r.writeToKafka(agMessages)
 				metric.Clear()
 			}
 			time.Sleep(r.Config.Cycle)