Simplify the headers type

Author: mrahs

context.go

@@ -11,7 +11,7 @@ import (
 	"github.com/lovoo/goka/multierr"
 )
 
-type emitter func(topic string, key string, value []byte, headers *Headers) *Promise
+type emitter func(topic string, key string, value []byte, headers Headers) *Promise
 
 // Context provides access to the processor's table and emit capabilities to
 // arbitrary topics in kafka.
@@ -62,7 +62,7 @@ type Context interface {
 	Value() interface{}
 
 	// Headers returns the headers of the input message
-	Headers() *Headers
+	Headers() Headers
 
 	// SetValue updates the value of the key in the group table.
 	// It stores the value in the local cache and sends the
@@ -148,7 +148,7 @@ type cbContext struct {
 
 	// Headers as passed from sarama. Note that this field will be filled
 	// lazily after the first call to Headers
-	headers *Headers
+	headers Headers
 
 	table *PartitionTable
 	// joins
@@ -223,7 +223,7 @@ func (ctx *cbContext) Loopback(key string, value interface{}, options ...Context
 	ctx.emit(l.Topic(), key, data, opts.emitHeaders)
 }
 
-func (ctx *cbContext) emit(topic string, key string, value []byte, headers *Headers) {
+func (ctx *cbContext) emit(topic string, key string, value []byte, headers Headers) {
 	ctx.counters.emits++
 	ctx.emitter(topic, key, value, ctx.emitterDefaultHeaders.Merged(headers)).Then(func(err error) {
 		if err != nil {
@@ -285,7 +285,7 @@ func (ctx *cbContext) Partition() int32 {
 	return ctx.msg.Partition
 }
 
-func (ctx *cbContext) Headers() *Headers {
+func (ctx *cbContext) Headers() Headers {
 	if ctx.headers == nil {
 		ctx.headers = HeadersFromSarama(ctx.msg.Headers)
 	}
@@ -349,7 +349,7 @@ func (ctx *cbContext) valueForKey(key string) (interface{}, error) {
 	return value, nil
 }
 
-func (ctx *cbContext) deleteKey(key string, headers *Headers) error {
+func (ctx *cbContext) deleteKey(key string, headers Headers) error {
 	if ctx.graph.GroupTable() == nil {
 		return fmt.Errorf("Cannot access state in stateless processor")
 	}
@@ -368,7 +368,7 @@ func (ctx *cbContext) deleteKey(key string, headers *Headers) error {
 }
 
 // setValueForKey sets a value for a key in the processor state.
-func (ctx *cbContext) setValueForKey(key string, value interface{}, headers *Headers) error {
+func (ctx *cbContext) setValueForKey(key string, value interface{}, headers Headers) error {
 	if ctx.graph.GroupTable() == nil {
 		return fmt.Errorf("Cannot access state in stateless processor")
 	}

context_test.go

@@ -17,7 +17,7 @@ import (
 )
 
 func newEmitter(err error, done func(err error)) emitter {
-	return func(topic string, key string, value []byte, headers *Headers) *Promise {
+	return func(topic string, key string, value []byte, headers Headers) *Promise {
 		p := NewPromise()
 		if done != nil {
 			p.Then(done)
@@ -27,7 +27,7 @@ func newEmitter(err error, done func(err error)) emitter {
 }
 
 func newEmitterW(wg *sync.WaitGroup, err error, done func(err error)) emitter {
-	return func(topic string, key string, value []byte, headers *Headers) *Promise {
+	return func(topic string, key string, value []byte, headers Headers) *Promise {
 		wg.Add(1)
 		p := NewPromise()
 		if done != nil {
@@ -369,14 +369,14 @@ func TestContext_GetSetStateful(t *testing.T) {
 	defer ctrl.Finish()
 
 	var (
-		group      Group = "some-group"
-		key              = "key"
-		value            = "value"
-		headers, _       = NewHeaders("key", "headerValue")
-		offset           = int64(123)
-		wg               = new(sync.WaitGroup)
-		st               = NewMockStorage(ctrl)
-		pt               = &PartitionTable{
+		group   Group = "some-group"
+		key           = "key"
+		value         = "value"
+		headers       = Headers{"key": []byte("headerValue")}
+		offset        = int64(123)
+		wg            = new(sync.WaitGroup)
+		st            = NewMockStorage(ctrl)
+		pt            = &PartitionTable{
 			st: &storageProxy{
 				Storage: st,
 			},
@@ -397,7 +397,7 @@ func TestContext_GetSetStateful(t *testing.T) {
 		graph:            graph,
 		trackOutputStats: func(ctx context.Context, topic string, size int) {},
 		msg:              &sarama.ConsumerMessage{Key: []byte(key), Offset: offset},
-		emitter: func(tp string, k string, v []byte, h *Headers) *Promise {
+		emitter: func(tp string, k string, v []byte, h Headers) *Promise {
 			wg.Add(1)
 			test.AssertEqual(t, tp, graph.GroupTable().Topic())
 			test.AssertEqual(t, string(k), key)
@@ -493,18 +493,18 @@ func TestContext_Loopback(t *testing.T) {
 	defer ctrl.Finish()
 
 	var (
-		key        = "key"
-		value      = "value"
-		headers, _ = NewHeaders("key", "headerValue")
-		cnt        = 0
+		key     = "key"
+		value   = "value"
+		headers = Headers{"key": []byte("headerValue")}
+		cnt     = 0
 	)
 
 	graph := DefineGroup("group", Persist(c), Loop(c, cb))
 	ctx := &cbContext{
 		graph:            graph,
 		msg:              &sarama.ConsumerMessage{},
 		trackOutputStats: func(ctx context.Context, topic string, size int) {},
-		emitter: func(tp string, k string, v []byte, h *Headers) *Promise {
+		emitter: func(tp string, k string, v []byte, h Headers) *Promise {
 			cnt++
 			test.AssertEqual(t, tp, graph.LoopStream().Topic())
 			test.AssertEqual(t, string(k), key)
@@ -632,7 +632,7 @@ func TestContext_Headers(t *testing.T) {
 	}
 	headers := ctx.Headers()
 	test.AssertNotNil(t, headers)
-	test.AssertEqual(t, headers.Len(), 0)
+	test.AssertEqual(t, len(headers), 0)
 
 	ctx = &cbContext{
 		msg: &sarama.ConsumerMessage{Key: []byte("key"), Headers: []*sarama.RecordHeader{
@@ -643,7 +643,7 @@ func TestContext_Headers(t *testing.T) {
 		}},
 	}
 	headers = ctx.Headers()
-	test.AssertEqual(t, headers.Val("key"), []byte("value"))
+	test.AssertEqual(t, headers["key"], []byte("value"))
 }
 
 func TestContext_Fail(t *testing.T) {

emitter.go

@@ -17,7 +17,7 @@ type Emitter struct {
 	producer Producer
 
 	topic          string
-	defaultHeaders *Headers
+	defaultHeaders Headers
 
 	wg   sync.WaitGroup
 	mu   sync.RWMutex
@@ -57,7 +57,7 @@ func NewEmitter(brokers []string, topic Stream, codec Codec, options ...EmitterO
 func (e *Emitter) emitDone(err error) { e.wg.Done() }
 
 // EmitWithHeaders sends a message with the given headers for the passed key using the emitter's codec.
-func (e *Emitter) EmitWithHeaders(key string, msg interface{}, headers *Headers) (*Promise, error) {
+func (e *Emitter) EmitWithHeaders(key string, msg interface{}, headers Headers) (*Promise, error) {
 	var (
 		err  error
 		data []byte
@@ -95,7 +95,7 @@ func (e *Emitter) Emit(key string, msg interface{}) (*Promise, error) {
 }
 
 // EmitSyncWithHeaders sends a message with the given headers to passed topic and key.
-func (e *Emitter) EmitSyncWithHeaders(key string, msg interface{}, headers *Headers) error {
+func (e *Emitter) EmitSyncWithHeaders(key string, msg interface{}, headers Headers) error {
 	var (
 		err     error
 		promise *Promise

emitter_test.go

@@ -118,9 +118,7 @@ func TestEmitter_Emit(t *testing.T) {
 	})
 	t.Run("default_headers", func(t *testing.T) {
 		emitter, bm, ctrl := createEmitter(t)
-		defaultHeaders, err := NewHeaders("header-key", "header-val")
-		test.AssertNil(t, err)
-		emitter.defaultHeaders = defaultHeaders
+		emitter.defaultHeaders = Headers{"header-key": []byte("header-val")}
 		defer ctrl.Finish()
 
 		var (

header_test.go

@@ -2,81 +2,37 @@ package goka
 
 import (
 	"testing"
-
-	"github.com/Shopify/sarama"
 )
 
 func TestHeaders_Merged(t *testing.T) {
-	h1, _ := NewHeaders()
-	h1.Set("key1", []byte("val1"))
-	h2, _ := NewHeaders()
-	h2.Set("key1", []byte("val1b"))
-	h2.Set("key2", []byte("val2"))
+	h1 := Headers{
+		"key1": []byte("val1"),
+	}
+	h2 := Headers{
+		"key1": []byte("val1b"),
+		"key2": []byte("val2"),
+	}
 	merged := h1.Merged(h2)
 
-	if h1.Len() != 1 || h1.StrVal("key1") != "val1" {
-		t.Errorf("Merged failed: reciver was modified")
+	if len(h1) != 1 || string(h1["key1"]) != "val1" {
+		t.Errorf("Merged failed: receiver was modified")
 	}
 
-	if h2.Len() != 2 || h2.StrVal("key1") != "val1b" || h2.StrVal("key2") != "val2" {
+	if len(h2) != 2 || string(h2["key1"]) != "val1b" || string(h2["key2"]) != "val2" {
 		t.Errorf("Merged failed: argument was modified")
 	}
 
-	if merged.Len() != 2 {
-		t.Errorf("Merged failed: expected %d keys, but found %d", 2, merged.Len())
+	if len(merged) != 2 {
+		t.Errorf("Merged failed: expected %d keys, but found %d", 2, len(merged))
 	}
 
-	if merged.StrVal("key1") != "val1b" {
+	if string(merged["key1"]) != "val1b" {
 		t.Errorf("Merged failed: expected %q for key %q, but found %q",
-			"val1b", "key1", merged.StrVal("key1"))
+			"val1b", "key1", string(merged["key1"]))
 	}
 
-	if merged.StrVal("key2") != "val2" {
+	if string(merged["key2"]) != "val2" {
 		t.Errorf("Merged failed: expected %q for key %q, but found %q",
-			"val2", "key2", merged.StrVal("key2"))
-	}
-}
-
-func TestHeaders_Lazy(t *testing.T) {
-	h := HeadersFromSarama([]*sarama.RecordHeader{{Key: []byte("key"), Value: []byte("value")}})
-	if len(h.sarama) == 0 {
-		t.Fatalf("Sarama headers should be set")
-	}
-	if len(h.goka) > 0 {
-		t.Fatalf("Goka headers should not be set")
-	}
-
-	if h.Len() != 1 {
-		t.Fatalf("Unexpected header count: expected 1, but found %d", h.Len())
-	}
-
-	if len(h.sarama) > 0 {
-		t.Fatalf("Sarama headers should not be set")
-	}
-	if len(h.goka) == 0 {
-		t.Fatalf("Goka headers should be set")
-	}
-}
-
-func TestHeaders_New(t *testing.T) {
-	h, err := NewHeaders("k1", "v1", "k2", "v2")
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
-
-	if h.Len() != 2 {
-		t.Fatalf("Unexpected header count: expected 2, but found %d", h.Len())
-	}
-
-	if h.StrVal("k1") != "v1" {
-		t.Fatalf("Unexpected value for k1: %q", h.StrVal("k1"))
-	}
-	if h.StrVal("k2") != "v2" {
-		t.Fatalf("Unexpected value for k2: %q", h.StrVal("k2"))
-	}
-
-	h, err = NewHeaders("k")
-	if err == nil {
-		t.Fatalf("Expected an error for a key without value")
+			"val2", "key2", string(merged["key2"]))
 	}
 }