123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468 |
- package sarama
- import (
- "encoding/binary"
- "math"
- "github.com/rcrowley/go-metrics"
- )
- var (
- errInvalidArrayLength = PacketDecodingError{"invalid array length"}
- errInvalidByteSliceLength = PacketDecodingError{"invalid byteslice length"}
- errInvalidStringLength = PacketDecodingError{"invalid string length"}
- errVarintOverflow = PacketDecodingError{"varint overflow"}
- errUVarintOverflow = PacketDecodingError{"uvarint overflow"}
- errInvalidBool = PacketDecodingError{"invalid bool"}
- )
- type realDecoder struct {
- raw []byte
- off int
- stack []pushDecoder
- registry metrics.Registry
- }
- // primitives
- func (rd *realDecoder) getInt8() (int8, error) {
- if rd.remaining() < 1 {
- rd.off = len(rd.raw)
- return -1, ErrInsufficientData
- }
- tmp := int8(rd.raw[rd.off])
- rd.off++
- return tmp, nil
- }
- func (rd *realDecoder) getInt16() (int16, error) {
- if rd.remaining() < 2 {
- rd.off = len(rd.raw)
- return -1, ErrInsufficientData
- }
- tmp := int16(binary.BigEndian.Uint16(rd.raw[rd.off:]))
- rd.off += 2
- return tmp, nil
- }
- func (rd *realDecoder) getInt32() (int32, error) {
- if rd.remaining() < 4 {
- rd.off = len(rd.raw)
- return -1, ErrInsufficientData
- }
- tmp := int32(binary.BigEndian.Uint32(rd.raw[rd.off:]))
- rd.off += 4
- return tmp, nil
- }
- func (rd *realDecoder) getInt64() (int64, error) {
- if rd.remaining() < 8 {
- rd.off = len(rd.raw)
- return -1, ErrInsufficientData
- }
- tmp := int64(binary.BigEndian.Uint64(rd.raw[rd.off:]))
- rd.off += 8
- return tmp, nil
- }
- func (rd *realDecoder) getVarint() (int64, error) {
- tmp, n := binary.Varint(rd.raw[rd.off:])
- if n == 0 {
- rd.off = len(rd.raw)
- return -1, ErrInsufficientData
- }
- if n < 0 {
- rd.off -= n
- return -1, errVarintOverflow
- }
- rd.off += n
- return tmp, nil
- }
- func (rd *realDecoder) getUVarint() (uint64, error) {
- tmp, n := binary.Uvarint(rd.raw[rd.off:])
- if n == 0 {
- rd.off = len(rd.raw)
- return 0, ErrInsufficientData
- }
- if n < 0 {
- rd.off -= n
- return 0, errUVarintOverflow
- }
- rd.off += n
- return tmp, nil
- }
- func (rd *realDecoder) getFloat64() (float64, error) {
- if rd.remaining() < 8 {
- rd.off = len(rd.raw)
- return -1, ErrInsufficientData
- }
- tmp := math.Float64frombits(binary.BigEndian.Uint64(rd.raw[rd.off:]))
- rd.off += 8
- return tmp, nil
- }
- func (rd *realDecoder) getArrayLength() (int, error) {
- if rd.remaining() < 4 {
- rd.off = len(rd.raw)
- return -1, ErrInsufficientData
- }
- tmp := int(int32(binary.BigEndian.Uint32(rd.raw[rd.off:])))
- rd.off += 4
- if tmp > rd.remaining() {
- rd.off = len(rd.raw)
- return -1, ErrInsufficientData
- } else if tmp > 2*math.MaxUint16 {
- return -1, errInvalidArrayLength
- }
- return tmp, nil
- }
- func (rd *realDecoder) getCompactArrayLength() (int, error) {
- n, err := rd.getUVarint()
- if err != nil {
- return 0, err
- }
- if n == 0 {
- return 0, nil
- }
- return int(n) - 1, nil
- }
- func (rd *realDecoder) getBool() (bool, error) {
- b, err := rd.getInt8()
- if err != nil || b == 0 {
- return false, err
- }
- if b != 1 {
- return false, errInvalidBool
- }
- return true, nil
- }
- func (rd *realDecoder) getEmptyTaggedFieldArray() (int, error) {
- tagCount, err := rd.getUVarint()
- if err != nil {
- return 0, err
- }
- // skip over any tagged fields without deserializing them
- // as we don't currently support doing anything with them
- for i := uint64(0); i < tagCount; i++ {
- // fetch and ignore tag identifier
- _, err := rd.getUVarint()
- if err != nil {
- return 0, err
- }
- length, err := rd.getUVarint()
- if err != nil {
- return 0, err
- }
- if _, err := rd.getRawBytes(int(length)); err != nil {
- return 0, err
- }
- }
- return 0, nil
- }
- // collections
- func (rd *realDecoder) getBytes() ([]byte, error) {
- tmp, err := rd.getInt32()
- if err != nil {
- return nil, err
- }
- if tmp == -1 {
- return nil, nil
- }
- return rd.getRawBytes(int(tmp))
- }
- func (rd *realDecoder) getVarintBytes() ([]byte, error) {
- tmp, err := rd.getVarint()
- if err != nil {
- return nil, err
- }
- if tmp == -1 {
- return nil, nil
- }
- return rd.getRawBytes(int(tmp))
- }
- func (rd *realDecoder) getCompactBytes() ([]byte, error) {
- n, err := rd.getUVarint()
- if err != nil {
- return nil, err
- }
- length := int(n - 1)
- return rd.getRawBytes(length)
- }
- func (rd *realDecoder) getStringLength() (int, error) {
- length, err := rd.getInt16()
- if err != nil {
- return 0, err
- }
- n := int(length)
- switch {
- case n < -1:
- return 0, errInvalidStringLength
- case n > rd.remaining():
- rd.off = len(rd.raw)
- return 0, ErrInsufficientData
- }
- return n, nil
- }
- func (rd *realDecoder) getString() (string, error) {
- n, err := rd.getStringLength()
- if err != nil || n == -1 {
- return "", err
- }
- tmpStr := string(rd.raw[rd.off : rd.off+n])
- rd.off += n
- return tmpStr, nil
- }
- func (rd *realDecoder) getNullableString() (*string, error) {
- n, err := rd.getStringLength()
- if err != nil || n == -1 {
- return nil, err
- }
- tmpStr := string(rd.raw[rd.off : rd.off+n])
- rd.off += n
- return &tmpStr, err
- }
- func (rd *realDecoder) getCompactString() (string, error) {
- n, err := rd.getUVarint()
- if err != nil {
- return "", err
- }
- length := int(n - 1)
- if length < 0 {
- return "", errInvalidByteSliceLength
- }
- tmpStr := string(rd.raw[rd.off : rd.off+length])
- rd.off += length
- return tmpStr, nil
- }
- func (rd *realDecoder) getCompactNullableString() (*string, error) {
- n, err := rd.getUVarint()
- if err != nil {
- return nil, err
- }
- length := int(n - 1)
- if length < 0 {
- return nil, err
- }
- tmpStr := string(rd.raw[rd.off : rd.off+length])
- rd.off += length
- return &tmpStr, err
- }
- func (rd *realDecoder) getCompactInt32Array() ([]int32, error) {
- n, err := rd.getUVarint()
- if err != nil {
- return nil, err
- }
- if n == 0 {
- return nil, nil
- }
- arrayLength := int(n) - 1
- ret := make([]int32, arrayLength)
- for i := range ret {
- ret[i] = int32(binary.BigEndian.Uint32(rd.raw[rd.off:]))
- rd.off += 4
- }
- return ret, nil
- }
- func (rd *realDecoder) getInt32Array() ([]int32, error) {
- if rd.remaining() < 4 {
- rd.off = len(rd.raw)
- return nil, ErrInsufficientData
- }
- n := int(binary.BigEndian.Uint32(rd.raw[rd.off:]))
- rd.off += 4
- if rd.remaining() < 4*n {
- rd.off = len(rd.raw)
- return nil, ErrInsufficientData
- }
- if n == 0 {
- return nil, nil
- }
- if n < 0 {
- return nil, errInvalidArrayLength
- }
- ret := make([]int32, n)
- for i := range ret {
- ret[i] = int32(binary.BigEndian.Uint32(rd.raw[rd.off:]))
- rd.off += 4
- }
- return ret, nil
- }
- func (rd *realDecoder) getInt64Array() ([]int64, error) {
- if rd.remaining() < 4 {
- rd.off = len(rd.raw)
- return nil, ErrInsufficientData
- }
- n := int(binary.BigEndian.Uint32(rd.raw[rd.off:]))
- rd.off += 4
- if rd.remaining() < 8*n {
- rd.off = len(rd.raw)
- return nil, ErrInsufficientData
- }
- if n == 0 {
- return nil, nil
- }
- if n < 0 {
- return nil, errInvalidArrayLength
- }
- ret := make([]int64, n)
- for i := range ret {
- ret[i] = int64(binary.BigEndian.Uint64(rd.raw[rd.off:]))
- rd.off += 8
- }
- return ret, nil
- }
- func (rd *realDecoder) getStringArray() ([]string, error) {
- if rd.remaining() < 4 {
- rd.off = len(rd.raw)
- return nil, ErrInsufficientData
- }
- n := int(binary.BigEndian.Uint32(rd.raw[rd.off:]))
- rd.off += 4
- if n == 0 {
- return nil, nil
- }
- if n < 0 {
- return nil, errInvalidArrayLength
- }
- ret := make([]string, n)
- for i := range ret {
- str, err := rd.getString()
- if err != nil {
- return nil, err
- }
- ret[i] = str
- }
- return ret, nil
- }
- // subsets
- func (rd *realDecoder) remaining() int {
- return len(rd.raw) - rd.off
- }
- func (rd *realDecoder) getSubset(length int) (packetDecoder, error) {
- buf, err := rd.getRawBytes(length)
- if err != nil {
- return nil, err
- }
- return &realDecoder{raw: buf}, nil
- }
- func (rd *realDecoder) getRawBytes(length int) ([]byte, error) {
- if length < 0 {
- return nil, errInvalidByteSliceLength
- } else if length > rd.remaining() {
- rd.off = len(rd.raw)
- return nil, ErrInsufficientData
- }
- start := rd.off
- rd.off += length
- return rd.raw[start:rd.off], nil
- }
- func (rd *realDecoder) peek(offset, length int) (packetDecoder, error) {
- if rd.remaining() < offset+length {
- return nil, ErrInsufficientData
- }
- off := rd.off + offset
- return &realDecoder{raw: rd.raw[off : off+length]}, nil
- }
- func (rd *realDecoder) peekInt8(offset int) (int8, error) {
- const byteLen = 1
- if rd.remaining() < offset+byteLen {
- return -1, ErrInsufficientData
- }
- return int8(rd.raw[rd.off+offset]), nil
- }
- // stacks
- func (rd *realDecoder) push(in pushDecoder) error {
- in.saveOffset(rd.off)
- var reserve int
- if dpd, ok := in.(dynamicPushDecoder); ok {
- if err := dpd.decode(rd); err != nil {
- return err
- }
- } else {
- reserve = in.reserveLength()
- if rd.remaining() < reserve {
- rd.off = len(rd.raw)
- return ErrInsufficientData
- }
- }
- rd.stack = append(rd.stack, in)
- rd.off += reserve
- return nil
- }
- func (rd *realDecoder) pop() error {
- // this is go's ugly pop pattern (the inverse of append)
- in := rd.stack[len(rd.stack)-1]
- rd.stack = rd.stack[:len(rd.stack)-1]
- return in.check(rd.off, rd.raw)
- }
- func (rd *realDecoder) metricRegistry() metrics.Registry {
- return rd.registry
- }
|