metron_throughput.go

gistfile1.txt

package main_test

import (
	"time"

	. "github.com/onsi/ginkgo"
	. "github.com/onsi/gomega"

	. "github.com/jriguera/metron-throughput/receiver"

	"github.com/cloudfoundry/sonde-go/events"
	"github.com/gogo/protobuf/proto"
)

var _ = Describe("Dispatcher", func() {

	It("Can process a single log message", func() {

		jobQueue := make(chan Job, 10)
		go func() {
			for i := 0; i < 100000; i++ {
				envelope, _ := buildLogMessage()

				job := Job{
					PayloadV1: envelope,
					PayloadV2: nil,
					Version:   1,
				}

				jobQueue <- job

			}
		}()
		dispatcher := NewDispatcher(jobQueue, 10)
		dispatcher.Run("metron-throughput/zz", 10)

		dispatcher.WaitStop()
		dispatcher.Print(1)
	})
})

func buildLogMessage() (*events.Envelope, []byte) {
	envelope := &events.Envelope{
		Origin:    proto.String("metron-throughput/zz"),
		EventType: events.Envelope_LogMessage.Enum(),
		Timestamp: proto.Int64(time.Now().UnixNano()),
		LogMessage: &events.LogMessage{
			Message:     []byte("some-log-message"),
			MessageType: events.LogMessage_OUT.Enum(),
			Timestamp:   proto.Int64(time.Now().UnixNano()),
		},
	}
	data, err := proto.Marshal(envelope)
	Expect(err).ToNot(HaveOccurred())
	return envelope, data
}