grafana/vendor/github.com/linkedin/goavro/v2/binaryReader.go

161 lines
5.6 KiB
Go

// Copyright [2019] LinkedIn Corp. Licensed under the Apache License, Version
// 2.0 (the "License"); you may not use this file except in compliance with the
// License. You may obtain a copy of the License at
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
// WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
package goavro
import (
"fmt"
"io"
"math"
)
// bytesBinaryReader reads bytes from io.Reader and returns byte slice of
// specified size or the error encountered while trying to read those bytes.
func bytesBinaryReader(ior io.Reader) ([]byte, error) {
size, err := longBinaryReader(ior)
if err != nil {
return nil, fmt.Errorf("cannot read bytes: cannot read size: %s", err)
}
if size < 0 {
return nil, fmt.Errorf("cannot read bytes: size is negative: %d", size)
}
if size > MaxBlockSize {
return nil, fmt.Errorf("cannot read bytes: size exceeds MaxBlockSize: %d > %d", size, MaxBlockSize)
}
buf := make([]byte, size)
_, err = io.ReadAtLeast(ior, buf, int(size))
if err != nil {
return nil, fmt.Errorf("cannot read bytes: %s", err)
}
return buf, nil
}
// longBinaryReader reads bytes from io.Reader until has complete long value, or
// read error.
func longBinaryReader(ior io.Reader) (int64, error) {
var value uint64
var shift uint
var err error
var b byte
// NOTE: While benchmarks show it's more performant to invoke ReadByte when
// available, testing whether a variable's data type implements a particular
// method is quite slow too. So perform the test once, and branch to the
// appropriate loop based on the results.
if byteReader, ok := ior.(io.ByteReader); ok {
for {
if b, err = byteReader.ReadByte(); err != nil {
return 0, err // NOTE: must send back unaltered error to detect io.EOF
}
value |= uint64(b&intMask) << shift
if b&intFlag == 0 {
return (int64(value>>1) ^ -int64(value&1)), nil
}
shift += 7
}
}
// NOTE: ior does not also implement io.ByteReader, so we must allocate a
// byte slice with a single byte, and read each byte into the slice.
buf := make([]byte, 1)
for {
if _, err = ior.Read(buf); err != nil {
return 0, err // NOTE: must send back unaltered error to detect io.EOF
}
b = buf[0]
value |= uint64(b&intMask) << shift
if b&intFlag == 0 {
return (int64(value>>1) ^ -int64(value&1)), nil
}
shift += 7
}
}
// metadataBinaryReader reads bytes from io.Reader until has entire map value,
// or read error.
func metadataBinaryReader(ior io.Reader) (map[string][]byte, error) {
var err error
var value interface{}
// block count and block size
if value, err = longBinaryReader(ior); err != nil {
return nil, fmt.Errorf("cannot read map block count: %s", err)
}
blockCount := value.(int64)
if blockCount < 0 {
if blockCount == math.MinInt64 {
// The minimum number for any signed numerical type can never be
// made positive
return nil, fmt.Errorf("cannot read map with block count: %d", blockCount)
}
// NOTE: A negative block count implies there is a long encoded block
// size following the negative block count. We have no use for the block
// size in this decoder, so we read and discard the value.
blockCount = -blockCount // convert to its positive equivalent
if _, err = longBinaryReader(ior); err != nil {
return nil, fmt.Errorf("cannot read map block size: %s", err)
}
}
// Ensure block count does not exceed some sane value.
if blockCount > MaxBlockCount {
return nil, fmt.Errorf("cannot read map when block count exceeds MaxBlockCount: %d > %d", blockCount, MaxBlockCount)
}
// NOTE: While the attempt of a RAM optimization shown below is not
// necessary, many encoders will encode all items in a single block. We can
// optimize amount of RAM allocated by runtime for the array by initializing
// the array for that number of items.
mapValues := make(map[string][]byte, blockCount)
for blockCount != 0 {
// Decode `blockCount` datum values from buffer
for i := int64(0); i < blockCount; i++ {
// first decode the key string
keyBytes, err := bytesBinaryReader(ior)
if err != nil {
return nil, fmt.Errorf("cannot read map key: %s", err)
}
key := string(keyBytes)
if _, ok := mapValues[key]; ok {
return nil, fmt.Errorf("cannot read map: duplicate key: %q", key)
}
// metadata values are always bytes
buf, err := bytesBinaryReader(ior)
if err != nil {
return nil, fmt.Errorf("cannot read map value for key %q: %s", key, err)
}
mapValues[key] = buf
}
// Decode next blockCount from buffer, because there may be more blocks
if value, err = longBinaryReader(ior); err != nil {
return nil, fmt.Errorf("cannot read map block count: %s", err)
}
blockCount = value.(int64)
if blockCount < 0 {
if blockCount == math.MinInt64 {
// The minimum number for any signed numerical type can never be
// made positive
return nil, fmt.Errorf("cannot read map with block count: %d", blockCount)
}
// NOTE: A negative block count implies there is a long encoded
// block size following the negative block count. We have no use for
// the block size in this decoder, so we read and discard the value.
blockCount = -blockCount // convert to its positive equivalent
if _, err = longBinaryReader(ior); err != nil {
return nil, fmt.Errorf("cannot read map block size: %s", err)
}
}
// Ensure block count does not exceed some sane value.
if blockCount > MaxBlockCount {
return nil, fmt.Errorf("cannot read map when block count exceeds MaxBlockCount: %d > %d", blockCount, MaxBlockCount)
}
}
return mapValues, nil
}