main.go

/***** BEGIN LICENSE BLOCK *****
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this file,
# You can obtain one at http://mozilla.org/MPL/2.0/.
#
# The Initial Developer of the Original Code is the Mozilla Foundation.
# Portions created by the Initial Developer are Copyright (C) 2012-2014
# the Initial Developer. All Rights Reserved.
#
# Contributor(s):
#   Rob Miller (rmiller@mozilla.com)
#
# ***** END LICENSE BLOCK *****/

package main

import (
	"code.google.com/p/go-uuid/uuid"
	"flag"
	"fmt"
	"github.com/glycerine/go-capnproto"
	"github.com/rafrombrc/capnheka/message"
	"log"
	"math/rand"
	"os"
	"time"
)

func makeMessage() message.Message {

	hostname, _ := os.Hostname()
	pid := int32(os.Getpid())

	seg := capn.NewBuffer(nil)
	msg := message.NewRootMessage(seg)
	msg.SetType("hekabench")
	msg.SetTimestamp(time.Now().UnixNano())
	msg.SetUuid(uuid.NewRandom())
	msg.SetSeverity(int32(6))
	msg.SetEnvVersion("0.8")
	msg.SetLogger("flood")
	msg.SetPid(pid)
	msg.SetHostname(hostname)
	msg.SetPayload("This is the payload")

	rand.Seed(time.Now().UnixNano())
	cntStr := rand.Int()%5 + 1
	cntByte := rand.Int()%5 + 1
	cntInt := rand.Int()%5 + 1
	cntFloat := rand.Int()%5 + 1
	// cntBool := rand.Int()%5 + 1

	// fields := message.NewFieldList(seg, cntStr+cntByte+cntInt+cntFloat+cntBool)
	fields := message.NewFieldList(seg, cntStr+cntByte+cntInt+cntFloat)
	_fields := capn.PointerList(fields)

	cnt := 0
	for c := 1; c <= cntStr; c++ {
		name := fmt.Sprintf("string%d", c)
		value := []string{fmt.Sprintf("value%d", c)}
		field, err := message.MakeField(seg, name, value, "")
		if err != nil {
			log.Println(err)
		}
		_fields.Set(cnt, capn.Object(field))
		cnt++
	}

	for c := 1; c <= cntByte; c++ {
		b := byte(c)
		field, err := message.MakeField(seg, fmt.Sprintf("bytes%d", c),
			[][]byte{[]byte{b, b, b, b, b, b, b, b}}, "")
		if err != nil {
			log.Println(err)
		}
		_fields.Set(cnt, capn.Object(field))
		cnt++
	}

	for c := 1; c <= cntInt; c++ {
		field, err := message.MakeField(seg, fmt.Sprintf("int%d", c),
			[]int32{int32(c)}, "")
		if err != nil {
			log.Println(err)
		}
		_fields.Set(cnt, capn.Object(field))
		cnt++
	}

	for c := 1; c <= cntFloat; c++ {
		field, err := message.MakeField(seg, fmt.Sprintf("double%d", c),
			[]float32{float32(c)}, "")
		if err != nil {
			log.Println(err)
		}
		_fields.Set(cnt, capn.Object(field))
		cnt++
	}

	// for c := 1; c <= cntBool; c++ {
	// 	field, err := message.MakeField(seg, fmt.Sprintf("bool%d", c),
	// 		[]bool{true}, "")
	// 	if err != nil {
	// 		log.Println(err)
	// 	}
	// 	_fields.Set(cnt, capn.Object(field))
	// 	cnt++
	// }

	msg.SetFields(fields)

	return msg
}

func readMessage(fileName *string) {
	file, err := os.Open(*fileName)
	if err != nil {
		panic(err)
	}
	seg, err := capn.ReadFromStream(file, nil)
	if err != nil {
		panic(err)
	}
	msg := message.ReadRootMessage(seg)

	fmt.Printf("message: %+v\n", msg)
	fmt.Println("payload: ", msg.Payload())
	for _, field := range msg.Fields().ToArray() {
		fmt.Println("field: ", field.Name())
		fmt.Println("type: ", field.ValueType())
		fmt.Println("value: ", field.GetValue())
		fmt.Println("-----")
	}
}

func main() {
	fileName := flag.String("file", "message.out", "Name of file.")
	read := flag.Bool("read", false, "Read from the file.")
	flag.Parse()

	if *read {
		readMessage(fileName)
		os.Exit(0)
	}

	msg := makeMessage()
	file, err := os.Create(*fileName)
	defer file.Close()
	if err != nil {
		panic(err)
	}
	msg.Segment.WriteTo(file)
}

message.capnp

@0xb3d0696dbec6dbaa;

using Go = import "go.capnp";
$Go.package("message");
$Go.import("github.com/mozilla-services/heka/message");

struct Header {
	enum HmacHashFunction {
		md5 @0;
		sha1 @1;
	}
	messageLength     @0 :UInt32;
	hmacHashFunction  @1 :HmacHashFunction = md5;
	hmacSigner        @2 :Text;
	hmacKeyVersion    @3 :UInt32;
	hmac              @4 :Data;
}

struct Field {
	enum ValueType {
		string  @0;
		bytes   @1;
		integer @2;
		double  @3;
		bool    @4;
  }
  name            @0 :Text;
  valueType       @1 :ValueType = string;
  representation  @2 :Text;
  valueString     @3 :List(Text);
  valueBytes      @4 :List(Data);
  valueInteger    @5 :List(Int64);
  valueDouble     @6 :List(Float64);
  valueBool       @7 :List(Bool);
}

struct Message {
  uuid        @0 :Data;
  timestamp   @1 :Int64; # nanoseconds since UNIX epoch
  type        @2 :Text;
  logger      @3 :Text;
  severity    @4 :Int32 = 7;
  payload     @5 :Text;
  envVersion  @6 :Text;
  pid         @7 :Int32;
  hostname    @8 :Text;
  fields      @9 :List(Field);
}

message.go

/***** BEGIN LICENSE BLOCK *****
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this file,
# You can obtain one at http://mozilla.org/MPL/2.0/.
#
# The Initial Developer of the Original Code is the Mozilla Foundation.
# Portions created by the Initial Developer are Copyright (C) 2012-2014
# the Initial Developer. All Rights Reserved.
#
# Contributor(s):
#   Mike Trinkala (trink@mozilla.com)
#   Rob Miller (rmiller@mozilla.com)
#
# ***** END LICENSE BLOCK *****/

package message

import (
	"fmt"
	"github.com/glycerine/go-capnproto"
	"reflect"
)

const (
	HEADER_DELIMITER_SIZE = 2                         // record separator + len
	HEADER_FRAMING_SIZE   = HEADER_DELIMITER_SIZE + 1 // unit separator
	MAX_HEADER_SIZE       = 255
	MAX_MESSAGE_SIZE      = 128 * 1024
	MAX_RECORD_SIZE       = HEADER_FRAMING_SIZE + MAX_HEADER_SIZE + MAX_MESSAGE_SIZE
	RECORD_SEPARATOR      = uint8(0x1e)
	UNIT_SEPARATOR        = uint8(0x1f)
	UUID_SIZE             = 16
)

type MessageSigningConfig struct {
	Name    string `toml:"name"`
	Hash    string `toml:"hmac_hash"`
	Key     string `toml:"hmac_key"`
	Version uint32 `toml:"version"`
}

func getValueType(v reflect.Type) (t FieldValueType, err error) {
	switch v.Kind() {
	case reflect.String:
		t = FIELDVALUETYPE_STRING
	case reflect.Array, reflect.Slice:
		if v.Elem().Kind() == reflect.Uint8 {
			t = FIELDVALUETYPE_BYTES
		}
	case reflect.Int, reflect.Int8, reflect.Int16, reflect.Int32, reflect.Int64:
		t = FIELDVALUETYPE_INTEGER
	case reflect.Float32, reflect.Float64:
		t = FIELDVALUETYPE_DOUBLE
	case reflect.Bool:
		t = FIELDVALUETYPE_BOOL
	default:
		err = fmt.Errorf("unsupported value kind: %v type: %v", v.Kind(), v)
	}
	return
}

// MakeField creates a Field w/ provided attributes, inferring type from
// the given value.
func MakeField(seg *capn.Segment, name string, value interface{},
	representation string) (Field, error) {

	var f Field
	v := reflect.ValueOf(value)
	if v.Kind() != reflect.Slice {
		return f, fmt.Errorf("%s must be slice type, not %s", name, v.Kind())
	}
	t, err := getValueType(v.Type().Elem())
	if err != nil {
		return f, fmt.Errorf("%s field - %s", name, err)
	}
	fmt.Println("Type: ", t)
	f = NewFieldInit(seg, t, name, value, representation)
	return f, nil
}

// NewFieldInit creates and initializes a field.
func NewFieldInit(seg *capn.Segment, valueType FieldValueType, name string,
	value interface{}, representation string) Field {

	f := NewField(seg)
	f.SetName(name)
	f.SetValueType(valueType)
	f.SetRepresentation(representation)

	switch valueType {
	case FIELDVALUETYPE_STRING:
		values := value.([]string)
		vlist := seg.NewTextList(len(values))
		for i, v := range values {
			vlist.Set(i, v)
		}
		f.SetValueString(vlist)
	case FIELDVALUETYPE_BYTES:
		values := value.([][]byte)
		vlist := seg.NewDataList(len(values))
		for i, v := range values {
			vlist.Set(i, v)
		}
		f.SetValueBytes(vlist)
	case FIELDVALUETYPE_INTEGER:
		var vlist capn.Int64List
		switch value.(type) {
		case []int:
			values := value.([]int)
			vlist = seg.NewInt64List(len(values))
			for i, v := range values {
				vlist.Set(i, int64(v))
			}
		case []int8:
			values := value.([]int8)
			vlist = seg.NewInt64List(len(values))
			for i, v := range values {
				vlist.Set(i, int64(v))
			}
		case []int16:
			values := value.([]int16)
			vlist = seg.NewInt64List(len(values))
			for i, v := range values {
				vlist.Set(i, int64(v))
			}
		case []int32:
			values := value.([]int32)
			vlist = seg.NewInt64List(len(values))
			for i, v := range values {
				vlist.Set(i, int64(v))
			}
		case []int64:
			values := value.([]int64)
			vlist = seg.NewInt64List(len(values))
			for i, v := range values {
				vlist.Set(i, v)
			}
		}
		f.SetValueInteger(vlist)
	case FIELDVALUETYPE_DOUBLE:
		var vlist capn.Float64List
		switch value.(type) {
		case []float32:
			values := value.([]float32)
			vlist = seg.NewFloat64List(len(values))
			for i, v := range values {
				vlist.Set(i, float64(v))
			}
		case []float64:
			values := value.([]float64)
			vlist = seg.NewFloat64List(len(values))
			for i, v := range values {
				vlist.Set(i, v)
			}
		}
		f.SetValueDouble(vlist)
	case FIELDVALUETYPE_BOOL:
		values := value.([]bool)
		vlist := seg.NewBitList(len(values))
		for i, v := range values {
			vlist.Set(i, v)
		}
		f.SetValueBool(vlist)
	}
	return f
}

// Helper function that returns the first value object for the given field.
func (f *Field) GetValue() (value interface{}) {
	switch f.ValueType() {
	case FIELDVALUETYPE_STRING:
		v := f.ValueString()
		if v.Len() > 0 {
			value = v.At(0)
		}
	case FIELDVALUETYPE_BYTES:
		v := f.ValueBytes()
		if v.Len() > 0 {
			value = v.At(0)
		}
	case FIELDVALUETYPE_INTEGER:
		v := f.ValueInteger()
		if v.Len() > 0 {
			value = v.At(0)
		}
	case FIELDVALUETYPE_DOUBLE:
		v := f.ValueDouble()
		if v.Len() > 0 {
			value = v.At(0)
		}
	case FIELDVALUETYPE_BOOL:
		v := f.ValueBool()
		if v.Len() > 0 {
			value = v.At(0)
		}
	}
	return
}