add protocol support for float64s

Author: petedannemann

protocol/alterclientquotas/alterclientquotas.go

@@ -12,7 +12,7 @@ type Request struct {
 	_ struct{} `kafka:"min=v2,max=v2,tag"`
 
 	Entries      []Entry `kafka:"min=v0,max=v1"`
-	ValidateOnly boolean `kafka:"min=v0,max=v1"`
+	ValidateOnly bool    `kafka:"min=v0,max=v1"`
 }
 
 func (r *Request) ApiKey() protocol.ApiKey { return protocol.AlterClientQuotas }
@@ -26,15 +26,22 @@ type Entry struct {
 	Ops      []Ops    `kafka:"min=v0,max=v1"`
 }
 
+type EntityType string
+
+const (
+	ClientID EntityType = "client-id"
+	User     EntityType = "user"
+)
+
 type Entity struct {
-	EntityType string `kafka:"min=v0,max=v1"`
-	EntityName string `kafka:"min=v0,max=v1,nullable"`
+	EntityType EntityType `kafka:"min=v0,max=v1"`
+	EntityName string     `kafka:"min=v0,max=v1,nullable"`
 }
 
 type Ops struct {
 	Key    string  `kafka:"min=v0,max=v1"`
 	Value  float64 `kafka:"min=v0,max=v1"`
-	Remove boolean `kafka:"min=v0,max=v1"`
+	Remove bool    `kafka:"min=v0,max=v1"`
 }
 
 type Response struct {

protocol/decode.go

@@ -7,6 +7,7 @@ import (
 	"hash/crc32"
 	"io"
 	"io/ioutil"
+	"math"
 	"reflect"
 	"sync"
 	"sync/atomic"
@@ -85,6 +86,10 @@ func (d *decoder) decodeInt64(v value) {
 	v.setInt64(d.readInt64())
 }
 
+func (d *decoder) decodeFloat64(v value) {
+	v.setFloat64(d.readFloat64())
+}
+
 func (d *decoder) decodeString(v value) {
 	v.setString(d.readString())
 }
@@ -216,6 +221,13 @@ func (d *decoder) readInt64() int64 {
 	return 0
 }
 
+func (d *decoder) readFloat64() float64 {
+	if d.readFull(d.buffer[:8]) {
+		return readFloat64(d.buffer[:8])
+	}
+	return 0
+}
+
 func (d *decoder) readString() string {
 	if n := d.readInt16(); n < 0 {
 		return ""
@@ -342,6 +354,8 @@ func decodeFuncOf(typ reflect.Type, version int16, flexible bool, tag structTag)
 		return (*decoder).decodeInt32
 	case reflect.Int64:
 		return (*decoder).decodeInt64
+	case reflect.Float64:
+		return (*decoder).decodeFloat64
 	case reflect.String:
 		return stringDecodeFuncOf(flexible, tag)
 	case reflect.Struct:
@@ -469,6 +483,10 @@ func readInt64(b []byte) int64 {
 	return int64(binary.BigEndian.Uint64(b))
 }
 
+func readFloat64(b []byte) float64 {
+	return math.Float64frombits(binary.BigEndian.Uint64(b))
+}
+
 func Unmarshal(data []byte, version int16, value interface{}) error {
 	typ := elemTypeOf(value)
 	cache, _ := unmarshalers.Load().(map[versionedType]decodeFunc)

protocol/encode.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"hash/crc32"
 	"io"
+	"math"
 	"reflect"
 	"sync"
 	"sync/atomic"
@@ -129,6 +130,10 @@ func (e *encoder) encodeInt64(v value) {
 	e.writeInt64(v.int64())
 }
 
+func (e *encoder) encodeFloat64(v value) {
+	e.writeFloat64(v.float64())
+}
+
 func (e *encoder) encodeString(v value) {
 	e.writeString(v.string())
 }
@@ -230,6 +235,11 @@ func (e *encoder) writeInt64(i int64) {
 	e.Write(e.buffer[:8])
 }
 
+func (e *encoder) writeFloat64(f float64) {
+	writeFloat64(e.buffer[:8], f)
+	e.Write(e.buffer[:8])
+}
+
 func (e *encoder) writeString(s string) {
 	e.writeInt16(int16(len(s)))
 	e.WriteString(s)
@@ -378,6 +388,8 @@ func encodeFuncOf(typ reflect.Type, version int16, flexible bool, tag structTag)
 		return (*encoder).encodeInt32
 	case reflect.Int64:
 		return (*encoder).encodeInt64
+	case reflect.Float64:
+		return (*encoder).encodeFloat64
 	case reflect.String:
 		return stringEncodeFuncOf(flexible, tag)
 	case reflect.Struct:
@@ -530,6 +542,10 @@ func writeInt64(b []byte, i int64) {
 	binary.BigEndian.PutUint64(b, uint64(i))
 }
 
+func writeFloat64(b []byte, f float64) {
+	binary.BigEndian.PutUint64(b, math.Float64bits(f))
+}
+
 func Marshal(version int16, value interface{}) ([]byte, error) {
 	typ := typeOf(value)
 	cache, _ := marshalers.Load().(map[versionedType]encodeFunc)

protocol/protocol_test.go

@@ -2,6 +2,7 @@ package protocol
 
 import (
 	"bytes"
+	"math"
 	"reflect"
 	"testing"
 )
@@ -279,3 +280,63 @@ func TestVarInts(t *testing.T) {
 
 	}
 }
+
+func TestFloat64(t *testing.T) {
+	type tc struct {
+		input    float64
+		expected []byte
+	}
+
+	tcs := []tc{
+		{
+			input:    0.0,
+			expected: []byte{0, 0, 0, 0, 0, 0, 0, 0},
+		},
+		{
+			input:    math.MaxFloat64,
+			expected: []byte{127, 239, 255, 255, 255, 255, 255, 255},
+		},
+		{
+			input:    -math.MaxFloat64,
+			expected: []byte{255, 239, 255, 255, 255, 255, 255, 255},
+		},
+		{
+			input:    math.SmallestNonzeroFloat64,
+			expected: []byte{0, 0, 0, 0, 0, 0, 0, 1},
+		},
+		{
+			input:    -math.SmallestNonzeroFloat64,
+			expected: []byte{128, 0, 0, 0, 0, 0, 0, 1},
+		},
+	}
+
+	for _, tc := range tcs {
+		b := &bytes.Buffer{}
+		e := &encoder{writer: b}
+		e.writeFloat64(tc.input)
+		if e.err != nil {
+			t.Errorf(
+				"Unexpected error encoding %f as float64: %+v",
+				tc.input,
+				e.err,
+			)
+		}
+		if !reflect.DeepEqual(b.Bytes(), tc.expected) {
+			t.Error(
+				"Wrong output encoding value", tc.input, "as float64",
+				"expected", tc.expected,
+				"got", b.Bytes(),
+			)
+		}
+
+		d := &decoder{reader: b, remain: len(b.Bytes())}
+		v := d.readFloat64()
+		if v != tc.input {
+			t.Error(
+				"Decoded float64 value does not equal encoded one",
+				"expected", tc.input,
+				"got", v,
+			)
+		}
+	}
+}

protocol/reflect.go

@@ -45,6 +45,8 @@ func (v value) int32() int32 { return int32(v.int64()) }
 
 func (v value) int64() int64 { return v.val.Int() }
 
+func (v value) float64() float64 { return v.val.Float() }
+
 func (v value) string() string { return v.val.String() }
 
 func (v value) bytes() []byte { return v.val.Bytes() }
@@ -63,6 +65,8 @@ func (v value) setInt32(i int32) { v.setInt64(int64(i)) }
 
 func (v value) setInt64(i int64) { v.val.SetInt(i) }
 
+func (v value) setFloat64(f float64) { v.val.SetFloat(f) }
+
 func (v value) setString(s string) { v.val.SetString(s) }
 
 func (v value) setBytes(b []byte) { v.val.SetBytes(b) }