mirror of
https://github.com/hanzoai/traces.git
synced 2026-10-30 18:20:41 +00:00
+6



![dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>](/assets/img/avatar_default.png)





Mario
GitHub
Javier Molina Reyes
Zach Leslie
Joe Elliott
dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
Ryan Perry
Kim Nylander
Suraj Nath
Alex Bikfalvi
Andrey Karpov
Jennifer Villa
Martin Disibio
Markus Toivonen
Daniel Strobusch
Carles Garcia
835e1fd250
* Block-builder PoC * Add unit test for block-builder (#4289) * Add unit test for block-builder * fmt * Update tests * cmon * Deterministically build blocks for partition sections (#4327) * Pull main (#4342) * chore: remove gofakeit dependency (#4274) * Further reduce Labes() calls in the metrics registry (#4283) * Respect passed headers in read path requests (#4287) * Ingester: Validate completed blocks (#4256) * Add validate method to block Signed-off-by: Joe Elliott <[email protected]> * Add Validate usage in the ingester Signed-off-by: Joe Elliott <[email protected]> * changelog Signed-off-by: Joe Elliott <[email protected]> * add test and fix replay Signed-off-by: Joe Elliott <[email protected]> * increment metric Signed-off-by: Joe Elliott <[email protected]> --------- Signed-off-by: Joe Elliott <[email protected]> * Add `invalid_utf8` to reasons spans could be rejected (#4293) * Add `invalid_utf8` to reasons spans could be rejected * Update changelog * Update docs * Ensure test covers invalid UTF-8 and not slack time * add signals for duplicate rf1 data (#4296) Signed-off-by: Joe Elliott <[email protected]> * Bump anchore/sbom-action from 0.17.5 to 0.17.7 (#4307) Bumps [anchore/sbom-action](https://github.com/anchore/sbom-action) from 0.17.5 to 0.17.7. - [Release notes](https://github.com/anchore/sbom-action/releases) - [Changelog](https://github.com/anchore/sbom-action/blob/main/RELEASE.md) - [Commits](https://github.com/anchore/sbom-action/compare/v0.17.5...v0.17.7) --- updated-dependencies: - dependency-name: anchore/sbom-action dependency-type: direct:production update-type: version-update:semver-patch ... Signed-off-by: dependabot[bot] <[email protected]> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> * docs: Update readme with explore traces info (#4263) * docs: Update readme with explore traces info Co-authored-by: Kim Nylander <[email protected]> * chore: remove spanlogger (#4312) * chore: remove spanlogger * Query-Frontend: Add middleware to drop headers (#4298) * header strip ware Signed-off-by: Joe Elliott <[email protected]> * comment Signed-off-by: Joe Elliott <[email protected]> * changelog Signed-off-by: Joe Elliott <[email protected]> * remove header strip wear from metrics summary Signed-off-by: Joe Elliott <[email protected]> --------- Signed-off-by: Joe Elliott <[email protected]> * Increase length of time compactions have to fail (#4315) * increase length of time compactions have to fail Signed-off-by: Joe Elliott <[email protected]> * gen Signed-off-by: Joe Elliott <[email protected]> --------- Signed-off-by: Joe Elliott <[email protected]> * docs: mark serverless as deprecated (#4017) * docs: mark serverless as deprecated * Changelog + readme * docs: Remove duplicated examples (#4295) This removes duplicates examples from the Configure TraceQL metrics page. Signed-off-by: Alex Bikfalvi <[email protected]> * tempo-cli: support dropping multiple traces in a single operation (#4266) * tempo-cli: support dropping multiple traces in a single operation * update final log message --------- Co-authored-by: Suraj Nath <[email protected]> * [DOC] Add clarification for metrics summary and traceQL metrics (#4316) * Add clarification for metrics summary and traceQL metrics * Apply suggestions from code review Co-authored-by: Jennifer Villa <[email protected]> * Update docs/sources/tempo/api_docs/metrics-summary.md --------- Co-authored-by: Jennifer Villa <[email protected]> * TraceQL metrics time range fixes (#4325) * Disconnect job time range filtering from step, so that results in split backend/recent range is accurate * changelog * Fix to assert metrics query range before alignment because alignment may increase it, which is not the responsibility of the caller to account for (#4331) * Add doc about configuring TLS with Helm (#4328) * Add doc about configuring TLS with Helm * Add memberlist and readinessProbe to example * Include server config for listening on TLS * Add note about scraping * Update docs/sources/tempo/configuration/network/tls.md Co-authored-by: Markus Toivonen <[email protected]> * Update docs/sources/tempo/configuration/network/tls.md Co-authored-by: Kim Nylander <[email protected]> * Update docs/sources/tempo/configuration/network/tls.md Co-authored-by: Kim Nylander <[email protected]> * Add memcached config for TLS --------- Co-authored-by: Markus Toivonen <[email protected]> Co-authored-by: Kim Nylander <[email protected]> * [DOC] Add TLS info to Helm chart doc (#4334) --------- Signed-off-by: Joe Elliott <[email protected]> Signed-off-by: dependabot[bot] <[email protected]> Signed-off-by: Alex Bikfalvi <[email protected]> Co-authored-by: Javier Molina Reyes <[email protected]> Co-authored-by: Zach Leslie <[email protected]> Co-authored-by: Joe Elliott <[email protected]> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> Co-authored-by: Ryan Perry <[email protected]> Co-authored-by: Kim Nylander <[email protected]> Co-authored-by: Suraj Nath <[email protected]> Co-authored-by: Alex Bikfalvi <[email protected]> Co-authored-by: Andrey Karpov <[email protected]> Co-authored-by: Jennifer Villa <[email protected]> Co-authored-by: Martin Disibio <[email protected]> Co-authored-by: Markus Toivonen <[email protected]> * WIP: Rhythm ingest path (#4314) * Validate distributor config. Finish encoder/decoder tests * Repair tests * Make SingleBinary work out of the box by defaulting to partition 0 * Fix first time startup where blockbuilder fails before ingester can create topic * Fix initial startup cycle time and delay * Add more failure modes to the block-builder (#4345) * Add more tests to the block-builder * stuff * Add comments * [Rhythm] Metrics generator read from kafka first pass (#4359) * Metrics generator read from kafka first pass * review feedback * Multiple fixes in block-builder (#4364) * [rhythm] git merge origin/main (#4376) * chore: remove gofakeit dependency (#4274) * Further reduce Labes() calls in the metrics registry (#4283) * Respect passed headers in read path requests (#4287) * Ingester: Validate completed blocks (#4256) * Add validate method to block Signed-off-by: Joe Elliott <[email protected]> * Add Validate usage in the ingester Signed-off-by: Joe Elliott <[email protected]> * changelog Signed-off-by: Joe Elliott <[email protected]> * add test and fix replay Signed-off-by: Joe Elliott <[email protected]> * increment metric Signed-off-by: Joe Elliott <[email protected]> --------- Signed-off-by: Joe Elliott <[email protected]> * Add `invalid_utf8` to reasons spans could be rejected (#4293) * Add `invalid_utf8` to reasons spans could be rejected * Update changelog * Update docs * Ensure test covers invalid UTF-8 and not slack time * add signals for duplicate rf1 data (#4296) Signed-off-by: Joe Elliott <[email protected]> * Bump anchore/sbom-action from 0.17.5 to 0.17.7 (#4307) Bumps [anchore/sbom-action](https://github.com/anchore/sbom-action) from 0.17.5 to 0.17.7. - [Release notes](https://github.com/anchore/sbom-action/releases) - [Changelog](https://github.com/anchore/sbom-action/blob/main/RELEASE.md) - [Commits](https://github.com/anchore/sbom-action/compare/v0.17.5...v0.17.7) --- updated-dependencies: - dependency-name: anchore/sbom-action dependency-type: direct:production update-type: version-update:semver-patch ... Signed-off-by: dependabot[bot] <[email protected]> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> * docs: Update readme with explore traces info (#4263) * docs: Update readme with explore traces info Co-authored-by: Kim Nylander <[email protected]> * chore: remove spanlogger (#4312) * chore: remove spanlogger * Query-Frontend: Add middleware to drop headers (#4298) * header strip ware Signed-off-by: Joe Elliott <[email protected]> * comment Signed-off-by: Joe Elliott <[email protected]> * changelog Signed-off-by: Joe Elliott <[email protected]> * remove header strip wear from metrics summary Signed-off-by: Joe Elliott <[email protected]> --------- Signed-off-by: Joe Elliott <[email protected]> * Increase length of time compactions have to fail (#4315) * increase length of time compactions have to fail Signed-off-by: Joe Elliott <[email protected]> * gen Signed-off-by: Joe Elliott <[email protected]> --------- Signed-off-by: Joe Elliott <[email protected]> * docs: mark serverless as deprecated (#4017) * docs: mark serverless as deprecated * Changelog + readme * docs: Remove duplicated examples (#4295) This removes duplicates examples from the Configure TraceQL metrics page. Signed-off-by: Alex Bikfalvi <[email protected]> * tempo-cli: support dropping multiple traces in a single operation (#4266) * tempo-cli: support dropping multiple traces in a single operation * update final log message --------- Co-authored-by: Suraj Nath <[email protected]> * [DOC] Add clarification for metrics summary and traceQL metrics (#4316) * Add clarification for metrics summary and traceQL metrics * Apply suggestions from code review Co-authored-by: Jennifer Villa <[email protected]> * Update docs/sources/tempo/api_docs/metrics-summary.md --------- Co-authored-by: Jennifer Villa <[email protected]> * TraceQL metrics time range fixes (#4325) * Disconnect job time range filtering from step, so that results in split backend/recent range is accurate * changelog * Fix to assert metrics query range before alignment because alignment may increase it, which is not the responsibility of the caller to account for (#4331) * Add doc about configuring TLS with Helm (#4328) * Add doc about configuring TLS with Helm * Add memberlist and readinessProbe to example * Include server config for listening on TLS * Add note about scraping * Update docs/sources/tempo/configuration/network/tls.md Co-authored-by: Markus Toivonen <[email protected]> * Update docs/sources/tempo/configuration/network/tls.md Co-authored-by: Kim Nylander <[email protected]> * Update docs/sources/tempo/configuration/network/tls.md Co-authored-by: Kim Nylander <[email protected]> * Add memcached config for TLS --------- Co-authored-by: Markus Toivonen <[email protected]> Co-authored-by: Kim Nylander <[email protected]> * [DOC] Add TLS info to Helm chart doc (#4334) * fix deprecation warning by switching to DoBatchWithOptions (#4343) Signed-off-by: Daniel Strobusch <[email protected]> * bump dskit to v0.0.0-20241115082728-f2a7eb3aa0e9 to leverage benefits for context causes for DoBatch calls. (#4341) See https://github.com/grafana/dskit/issues/576 Signed-off-by: Daniel Strobusch <[email protected]> * Bump github.com/minio/minio-go/v7 from 7.0.70 to 7.0.80 (#4282) * Bump github.com/minio/minio-go/v7 from 7.0.70 to 7.0.80 Bumps [github.com/minio/minio-go/v7](https://github.com/minio/minio-go) from 7.0.70 to 7.0.80. - [Release notes](https://github.com/minio/minio-go/releases) - [Commits](https://github.com/minio/minio-go/compare/v7.0.70...v7.0.80) --- updated-dependencies: - dependency-name: github.com/minio/minio-go/v7 dependency-type: direct:production update-type: version-update:semver-patch ... Signed-off-by: dependabot[bot] <[email protected]> * Update serverless vendor --------- Signed-off-by: dependabot[bot] <[email protected]> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> Co-authored-by: Zach Leslie <[email protected]> * update default config values to better align with production workloads (#4340) * update default config values to better align with production workloads * Update CHANGELOG.md and config docs * Ingester memory improvements by adjusting prealloc (#4344) * remove trace ids Signed-off-by: Joe Elliott <[email protected]> * linear buckets Signed-off-by: Joe Elliott <[email protected]> * changelog Signed-off-by: Joe Elliott <[email protected]> * tuney tune Signed-off-by: Joe Elliott <[email protected]> * metric misses and increase pool size Signed-off-by: Joe Elliott <[email protected]> * lint Signed-off-by: Joe Elliott <[email protected]> --------- Signed-off-by: Joe Elliott <[email protected]> * Bump github.com/Azure/azure-sdk-for-go/sdk/azcore from 1.13.0 to 1.16.0 (#4302) * Bump github.com/Azure/azure-sdk-for-go/sdk/azcore from 1.13.0 to 1.16.0 Bumps [github.com/Azure/azure-sdk-for-go/sdk/azcore](https://github.com/Azure/azure-sdk-for-go) from 1.13.0 to 1.16.0. - [Release notes](https://github.com/Azure/azure-sdk-for-go/releases) - [Changelog](https://github.com/Azure/azure-sdk-for-go/blob/main/documentation/release.md) - [Commits](https://github.com/Azure/azure-sdk-for-go/compare/sdk/azcore/v1.13.0...sdk/azcore/v1.16.0) --- updated-dependencies: - dependency-name: github.com/Azure/azure-sdk-for-go/sdk/azcore dependency-type: direct:production update-type: version-update:semver-minor ... Signed-off-by: dependabot[bot] <[email protected]> * Update serverless vendor --------- Signed-off-by: dependabot[bot] <[email protected]> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> Co-authored-by: Zach Leslie <[email protected]> * Use Prometheus fast regexp (#4329) * basic integration Signed-off-by: Joe Elliott <[email protected]> * patch tests for new meaning Signed-off-by: Joe Elliott <[email protected]> * patch up more tests Signed-off-by: Joe Elliott <[email protected]> * add basic tests Signed-off-by: Joe Elliott <[email protected]> * changelog + docs Signed-off-by: Joe Elliott <[email protected]> * remove benches Signed-off-by: Joe Elliott <[email protected]> * Cleaned up + tests Signed-off-by: Joe Elliott <[email protected]> * comment Signed-off-by: Joe Elliott <[email protected]> * lint Signed-off-by: Joe Elliott <[email protected]> * Update docs/sources/tempo/traceql/_index.md Co-authored-by: Kim Nylander <[email protected]> * comment Signed-off-by: Joe Elliott <[email protected]> --------- Signed-off-by: Joe Elliott <[email protected]> Co-authored-by: Kim Nylander <[email protected]> * Fix broken link in service-graphs docs (#4351) * Fix minor typo in TraceQL docs (#4356) * Bump default memcached version (#4363) * Exemplar fixes (#4366) * Fix exemplars based on duration to convert to seconds, fix various other issues * changelog * fix: initialize histogram buckets to 0 to avoid them being downsampled (#4368) * initialized histogram buckets to 0 to avoid them being downsampled * Ingester/Generator Live trace cleanup (#4365) * moved trace sizes somewhere shareable Signed-off-by: Joe Elliott <[email protected]> * use tracesizes in ingester Signed-off-by: Joe Elliott <[email protected]> * make tests work Signed-off-by: Joe Elliott <[email protected]> * trace bytes in generator Signed-off-by: Joe Elliott <[email protected]> * remove traceCount Signed-off-by: Joe Elliott <[email protected]> * live trace shenanigans Signed-off-by: Joe Elliott <[email protected]> * changelog Signed-off-by: Joe Elliott <[email protected]> * Update modules/generator/processor/localblocks/livetraces.go Co-authored-by: Mario <[email protected]> * Update modules/ingester/instance.go Co-authored-by: Mario <[email protected]> * Test cleanup. Add sz test, restore commented out and fix e2e Signed-off-by: Joe Elliott <[email protected]> * remove todo comment Signed-off-by: Joe Elliott <[email protected]> --------- Signed-off-by: Joe Elliott <[email protected]> Co-authored-by: Mario <[email protected]> * Bump anchore/sbom-action from 0.17.7 to 0.17.8 (#4371) Bumps [anchore/sbom-action](https://github.com/anchore/sbom-action) from 0.17.7 to 0.17.8. - [Release notes](https://github.com/anchore/sbom-action/releases) - [Changelog](https://github.com/anchore/sbom-action/blob/main/RELEASE.md) - [Commits](https://github.com/anchore/sbom-action/compare/v0.17.7...v0.17.8) --- updated-dependencies: - dependency-name: anchore/sbom-action dependency-type: direct:production update-type: version-update:semver-patch ... Signed-off-by: dependabot[bot] <[email protected]> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> * Update for IDs change * Only run blockbuilder if ingest enabled --------- Signed-off-by: Joe Elliott <[email protected]> Signed-off-by: dependabot[bot] <[email protected]> Signed-off-by: Alex Bikfalvi <[email protected]> Signed-off-by: Daniel Strobusch <[email protected]> Co-authored-by: Javier Molina Reyes <[email protected]> Co-authored-by: Zach Leslie <[email protected]> Co-authored-by: Joe Elliott <[email protected]> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> Co-authored-by: Ryan Perry <[email protected]> Co-authored-by: Kim Nylander <[email protected]> Co-authored-by: Suraj Nath <[email protected]> Co-authored-by: Alex Bikfalvi <[email protected]> Co-authored-by: Andrey Karpov <[email protected]> Co-authored-by: Jennifer Villa <[email protected]> Co-authored-by: Martin Disibio <[email protected]> Co-authored-by: Markus Toivonen <[email protected]> Co-authored-by: Daniel Strobusch <[email protected]> Co-authored-by: Carles Garcia <[email protected]> * [rhythm] Changes to simplify operations (#4389) * Use mapping for assigning partitions * Use mapping for assigning partitions in the generator too * Add support for SASL auth to kafka clients * Add metrics to ingest (#4395) * [rhythm] Extract block-builder into its own module (#4396) * Extract block-builder into its own module * Update /operations and examples * No ephemeral storage * No rolling strategy either * fmt and compile * Address review comment * [rhythm] Correctly pass start/end time when appending a trace (#4410) * Correctly pass start/end times * Different code, same result * [rhythm] Multiple fixes to block-builder consumption (#4413) * Multiple fixes to cycle consumption * fmt * happy now? * ups * Rhythm: Separate non-flushing local blocks processor to store new queue data for reads (#4411) * wip: separate non-flushing local blocks processor to store new queue data for reads * Make real config for non-flushing local blocks processor, optional, validate wal config and use defaults if needed * Fix defaulting of second WAL config * [rhythm] Make ID generator more robust (#4416) * Make ID generator more robust * Simplify * Update toc3ff5ba61* Fix registering of kafka read client metrics (#4502) * [rhythm] Make ID generator more robust (#4416) (#4507) * Make ID generator more robust * Simplify * Removed references to Loki and Mimir (#4509) Signed-off-by: Joe Elliott <[email protected]> * [Rhythm] Block builder test updates (#4510) * Make blockbuilder tests closer to real kafka and less implementation specific by always enabling support for consumer groups, call commit control func in order * Verify last committed offset in each test * hide test function * lint * lint * [Rhythm] Block-builder consumption loop (#4480) * Alternate block-builder consume * Set timeout on PollFetches, reduce initial poll delay, update 1 test to work using real consumergroup functionality * restore metrics * Re-add original partition lag metric, polled in separate goroutine. Fix consume loop to only consume full-duration cycles for more determinism * merge conflict * Review feedback * Review feedback * Comment * code cleanup, lint * logs * code cleanup * lint * Review feedback * Remove missed lookback_on_no_commit config in e2e tests and regen manifest * Review feedback * Fix rewind to latest commit to init correctly, it didn't work in some clusters (#4532) * [rhythm] merge main at2c44177(#4531) * Fixes * More fixes --------- Signed-off-by: Joe Elliott <[email protected]> Signed-off-by: dependabot[bot] <[email protected]> Signed-off-by: Alex Bikfalvi <[email protected]> Signed-off-by: Daniel Strobusch <[email protected]> Co-authored-by: Javier Molina Reyes <[email protected]> Co-authored-by: Zach Leslie <[email protected]> Co-authored-by: Joe Elliott <[email protected]> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> Co-authored-by: Ryan Perry <[email protected]> Co-authored-by: Kim Nylander <[email protected]> Co-authored-by: Suraj Nath <[email protected]> Co-authored-by: Alex Bikfalvi <[email protected]> Co-authored-by: Andrey Karpov <[email protected]> Co-authored-by: Jennifer Villa <[email protected]> Co-authored-by: Martin Disibio <[email protected]> Co-authored-by: Markus Toivonen <[email protected]> Co-authored-by: Daniel Strobusch <[email protected]> Co-authored-by: Carles Garcia <[email protected]>
857 lines
21 KiB
Go
857 lines
21 KiB
Go
// Package kbin contains Kafka primitive reading and writing functions.
|
|
package kbin
|
|
|
|
import (
|
|
"encoding/binary"
|
|
"errors"
|
|
"math"
|
|
"math/bits"
|
|
"reflect"
|
|
"unsafe"
|
|
)
|
|
|
|
// This file contains primitive type encoding and decoding.
|
|
//
|
|
// The Reader helper can be used even when content runs out
|
|
// or an error is hit; all other number requests will return
|
|
// zero so a decode will basically no-op.
|
|
|
|
// ErrNotEnoughData is returned when a type could not fully decode
|
|
// from a slice because the slice did not have enough data.
|
|
var ErrNotEnoughData = errors.New("response did not contain enough data to be valid")
|
|
|
|
// AppendBool appends 1 for true or 0 for false to dst.
|
|
func AppendBool(dst []byte, v bool) []byte {
|
|
if v {
|
|
return append(dst, 1)
|
|
}
|
|
return append(dst, 0)
|
|
}
|
|
|
|
// AppendInt8 appends an int8 to dst.
|
|
func AppendInt8(dst []byte, i int8) []byte {
|
|
return append(dst, byte(i))
|
|
}
|
|
|
|
// AppendInt16 appends a big endian int16 to dst.
|
|
func AppendInt16(dst []byte, i int16) []byte {
|
|
return AppendUint16(dst, uint16(i))
|
|
}
|
|
|
|
// AppendUint16 appends a big endian uint16 to dst.
|
|
func AppendUint16(dst []byte, u uint16) []byte {
|
|
return append(dst, byte(u>>8), byte(u))
|
|
}
|
|
|
|
// AppendInt32 appends a big endian int32 to dst.
|
|
func AppendInt32(dst []byte, i int32) []byte {
|
|
return AppendUint32(dst, uint32(i))
|
|
}
|
|
|
|
// AppendInt64 appends a big endian int64 to dst.
|
|
func AppendInt64(dst []byte, i int64) []byte {
|
|
return appendUint64(dst, uint64(i))
|
|
}
|
|
|
|
// AppendFloat64 appends a big endian float64 to dst.
|
|
func AppendFloat64(dst []byte, f float64) []byte {
|
|
return appendUint64(dst, math.Float64bits(f))
|
|
}
|
|
|
|
// AppendUuid appends the 16 uuid bytes to dst.
|
|
func AppendUuid(dst []byte, uuid [16]byte) []byte {
|
|
return append(dst, uuid[:]...)
|
|
}
|
|
|
|
func appendUint64(dst []byte, u uint64) []byte {
|
|
return append(dst, byte(u>>56), byte(u>>48), byte(u>>40), byte(u>>32),
|
|
byte(u>>24), byte(u>>16), byte(u>>8), byte(u))
|
|
}
|
|
|
|
// AppendUint32 appends a big endian uint32 to dst.
|
|
func AppendUint32(dst []byte, u uint32) []byte {
|
|
return append(dst, byte(u>>24), byte(u>>16), byte(u>>8), byte(u))
|
|
}
|
|
|
|
// uvarintLens could only be length 65, but using 256 allows bounds check
|
|
// elimination on lookup.
|
|
const uvarintLens = "\x01\x01\x01\x01\x01\x01\x01\x01\x02\x02\x02\x02\x02\x02\x02\x03\x03\x03\x03\x03\x03\x03\x04\x04\x04\x04\x04\x04\x04\x05\x05\x05\x05\x05\x05\x05\x06\x06\x06\x06\x06\x06\x06\x07\x07\x07\x07\x07\x07\x07\x08\x08\x08\x08\x08\x08\x08\x09\x09\x09\x09\x09\x09\x09\x0a\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
|
|
|
|
// VarintLen returns how long i would be if it were varint encoded.
|
|
func VarintLen(i int32) int {
|
|
u := uint32(i)<<1 ^ uint32(i>>31)
|
|
return UvarintLen(u)
|
|
}
|
|
|
|
// UvarintLen returns how long u would be if it were uvarint encoded.
|
|
func UvarintLen(u uint32) int {
|
|
return int(uvarintLens[byte(bits.Len32(u))])
|
|
}
|
|
|
|
// VarlongLen returns how long i would be if it were varlong encoded.
|
|
func VarlongLen(i int64) int {
|
|
u := uint64(i)<<1 ^ uint64(i>>63)
|
|
return uvarlongLen(u)
|
|
}
|
|
|
|
func uvarlongLen(u uint64) int {
|
|
return int(uvarintLens[byte(bits.Len64(u))])
|
|
}
|
|
|
|
// Varint is a loop unrolled 32 bit varint decoder. The return semantics
|
|
// are the same as binary.Varint, with the added benefit that overflows
|
|
// in 5 byte encodings are handled rather than left to the user.
|
|
func Varint(in []byte) (int32, int) {
|
|
x, n := Uvarint(in)
|
|
return int32((x >> 1) ^ -(x & 1)), n
|
|
}
|
|
|
|
// Uvarint is a loop unrolled 32 bit uvarint decoder. The return semantics
|
|
// are the same as binary.Uvarint, with the added benefit that overflows
|
|
// in 5 byte encodings are handled rather than left to the user.
|
|
func Uvarint(in []byte) (uint32, int) {
|
|
var x uint32
|
|
var overflow int
|
|
|
|
if len(in) < 1 {
|
|
goto fail
|
|
}
|
|
|
|
x = uint32(in[0] & 0x7f)
|
|
if in[0]&0x80 == 0 {
|
|
return x, 1
|
|
} else if len(in) < 2 {
|
|
goto fail
|
|
}
|
|
|
|
x |= uint32(in[1]&0x7f) << 7
|
|
if in[1]&0x80 == 0 {
|
|
return x, 2
|
|
} else if len(in) < 3 {
|
|
goto fail
|
|
}
|
|
|
|
x |= uint32(in[2]&0x7f) << 14
|
|
if in[2]&0x80 == 0 {
|
|
return x, 3
|
|
} else if len(in) < 4 {
|
|
goto fail
|
|
}
|
|
|
|
x |= uint32(in[3]&0x7f) << 21
|
|
if in[3]&0x80 == 0 {
|
|
return x, 4
|
|
} else if len(in) < 5 {
|
|
goto fail
|
|
}
|
|
|
|
x |= uint32(in[4]) << 28
|
|
if in[4] <= 0x0f {
|
|
return x, 5
|
|
}
|
|
|
|
overflow = -5
|
|
|
|
fail:
|
|
return 0, overflow
|
|
}
|
|
|
|
// Varlong is a loop unrolled 64 bit varint decoder. The return semantics
|
|
// are the same as binary.Varint, with the added benefit that overflows
|
|
// in 10 byte encodings are handled rather than left to the user.
|
|
func Varlong(in []byte) (int64, int) {
|
|
x, n := uvarlong(in)
|
|
return int64((x >> 1) ^ -(x & 1)), n
|
|
}
|
|
|
|
func uvarlong(in []byte) (uint64, int) {
|
|
var x uint64
|
|
var overflow int
|
|
|
|
if len(in) < 1 {
|
|
goto fail
|
|
}
|
|
|
|
x = uint64(in[0] & 0x7f)
|
|
if in[0]&0x80 == 0 {
|
|
return x, 1
|
|
} else if len(in) < 2 {
|
|
goto fail
|
|
}
|
|
|
|
x |= uint64(in[1]&0x7f) << 7
|
|
if in[1]&0x80 == 0 {
|
|
return x, 2
|
|
} else if len(in) < 3 {
|
|
goto fail
|
|
}
|
|
|
|
x |= uint64(in[2]&0x7f) << 14
|
|
if in[2]&0x80 == 0 {
|
|
return x, 3
|
|
} else if len(in) < 4 {
|
|
goto fail
|
|
}
|
|
|
|
x |= uint64(in[3]&0x7f) << 21
|
|
if in[3]&0x80 == 0 {
|
|
return x, 4
|
|
} else if len(in) < 5 {
|
|
goto fail
|
|
}
|
|
|
|
x |= uint64(in[4]&0x7f) << 28
|
|
if in[4]&0x80 == 0 {
|
|
return x, 5
|
|
} else if len(in) < 6 {
|
|
goto fail
|
|
}
|
|
|
|
x |= uint64(in[5]&0x7f) << 35
|
|
if in[5]&0x80 == 0 {
|
|
return x, 6
|
|
} else if len(in) < 7 {
|
|
goto fail
|
|
}
|
|
|
|
x |= uint64(in[6]&0x7f) << 42
|
|
if in[6]&0x80 == 0 {
|
|
return x, 7
|
|
} else if len(in) < 8 {
|
|
goto fail
|
|
}
|
|
|
|
x |= uint64(in[7]&0x7f) << 49
|
|
if in[7]&0x80 == 0 {
|
|
return x, 8
|
|
} else if len(in) < 9 {
|
|
goto fail
|
|
}
|
|
|
|
x |= uint64(in[8]&0x7f) << 56
|
|
if in[8]&0x80 == 0 {
|
|
return x, 9
|
|
} else if len(in) < 10 {
|
|
goto fail
|
|
}
|
|
|
|
x |= uint64(in[9]) << 63
|
|
if in[9] <= 0x01 {
|
|
return x, 10
|
|
}
|
|
|
|
overflow = -10
|
|
|
|
fail:
|
|
return 0, overflow
|
|
}
|
|
|
|
// AppendVarint appends a varint encoded i to dst.
|
|
func AppendVarint(dst []byte, i int32) []byte {
|
|
return AppendUvarint(dst, uint32(i)<<1^uint32(i>>31))
|
|
}
|
|
|
|
// AppendUvarint appends a uvarint encoded u to dst.
|
|
func AppendUvarint(dst []byte, u uint32) []byte {
|
|
switch UvarintLen(u) {
|
|
case 5:
|
|
return append(dst,
|
|
byte(u&0x7f|0x80),
|
|
byte((u>>7)&0x7f|0x80),
|
|
byte((u>>14)&0x7f|0x80),
|
|
byte((u>>21)&0x7f|0x80),
|
|
byte(u>>28))
|
|
case 4:
|
|
return append(dst,
|
|
byte(u&0x7f|0x80),
|
|
byte((u>>7)&0x7f|0x80),
|
|
byte((u>>14)&0x7f|0x80),
|
|
byte(u>>21))
|
|
case 3:
|
|
return append(dst,
|
|
byte(u&0x7f|0x80),
|
|
byte((u>>7)&0x7f|0x80),
|
|
byte(u>>14))
|
|
case 2:
|
|
return append(dst,
|
|
byte(u&0x7f|0x80),
|
|
byte(u>>7))
|
|
case 1:
|
|
return append(dst, byte(u))
|
|
}
|
|
return dst
|
|
}
|
|
|
|
// AppendVarlong appends a varint encoded i to dst.
|
|
func AppendVarlong(dst []byte, i int64) []byte {
|
|
return appendUvarlong(dst, uint64(i)<<1^uint64(i>>63))
|
|
}
|
|
|
|
func appendUvarlong(dst []byte, u uint64) []byte {
|
|
switch uvarlongLen(u) {
|
|
case 10:
|
|
return append(dst,
|
|
byte(u&0x7f|0x80),
|
|
byte((u>>7)&0x7f|0x80),
|
|
byte((u>>14)&0x7f|0x80),
|
|
byte((u>>21)&0x7f|0x80),
|
|
byte((u>>28)&0x7f|0x80),
|
|
byte((u>>35)&0x7f|0x80),
|
|
byte((u>>42)&0x7f|0x80),
|
|
byte((u>>49)&0x7f|0x80),
|
|
byte((u>>56)&0x7f|0x80),
|
|
byte(u>>63))
|
|
case 9:
|
|
return append(dst,
|
|
byte(u&0x7f|0x80),
|
|
byte((u>>7)&0x7f|0x80),
|
|
byte((u>>14)&0x7f|0x80),
|
|
byte((u>>21)&0x7f|0x80),
|
|
byte((u>>28)&0x7f|0x80),
|
|
byte((u>>35)&0x7f|0x80),
|
|
byte((u>>42)&0x7f|0x80),
|
|
byte((u>>49)&0x7f|0x80),
|
|
byte(u>>56))
|
|
case 8:
|
|
return append(dst,
|
|
byte(u&0x7f|0x80),
|
|
byte((u>>7)&0x7f|0x80),
|
|
byte((u>>14)&0x7f|0x80),
|
|
byte((u>>21)&0x7f|0x80),
|
|
byte((u>>28)&0x7f|0x80),
|
|
byte((u>>35)&0x7f|0x80),
|
|
byte((u>>42)&0x7f|0x80),
|
|
byte(u>>49))
|
|
case 7:
|
|
return append(dst,
|
|
byte(u&0x7f|0x80),
|
|
byte((u>>7)&0x7f|0x80),
|
|
byte((u>>14)&0x7f|0x80),
|
|
byte((u>>21)&0x7f|0x80),
|
|
byte((u>>28)&0x7f|0x80),
|
|
byte((u>>35)&0x7f|0x80),
|
|
byte(u>>42))
|
|
case 6:
|
|
return append(dst,
|
|
byte(u&0x7f|0x80),
|
|
byte((u>>7)&0x7f|0x80),
|
|
byte((u>>14)&0x7f|0x80),
|
|
byte((u>>21)&0x7f|0x80),
|
|
byte((u>>28)&0x7f|0x80),
|
|
byte(u>>35))
|
|
case 5:
|
|
return append(dst,
|
|
byte(u&0x7f|0x80),
|
|
byte((u>>7)&0x7f|0x80),
|
|
byte((u>>14)&0x7f|0x80),
|
|
byte((u>>21)&0x7f|0x80),
|
|
byte(u>>28))
|
|
case 4:
|
|
return append(dst,
|
|
byte(u&0x7f|0x80),
|
|
byte((u>>7)&0x7f|0x80),
|
|
byte((u>>14)&0x7f|0x80),
|
|
byte(u>>21))
|
|
case 3:
|
|
return append(dst,
|
|
byte(u&0x7f|0x80),
|
|
byte((u>>7)&0x7f|0x80),
|
|
byte(u>>14))
|
|
case 2:
|
|
return append(dst,
|
|
byte(u&0x7f|0x80),
|
|
byte(u>>7))
|
|
case 1:
|
|
return append(dst, byte(u))
|
|
}
|
|
return dst
|
|
}
|
|
|
|
// AppendString appends a string to dst prefixed with its int16 length.
|
|
func AppendString(dst []byte, s string) []byte {
|
|
dst = AppendInt16(dst, int16(len(s)))
|
|
return append(dst, s...)
|
|
}
|
|
|
|
// AppendCompactString appends a string to dst prefixed with its uvarint length
|
|
// starting at 1; 0 is reserved for null, which compact strings are not
|
|
// (nullable compact ones are!). Thus, the length is the decoded uvarint - 1.
|
|
//
|
|
// For KIP-482.
|
|
func AppendCompactString(dst []byte, s string) []byte {
|
|
dst = AppendUvarint(dst, 1+uint32(len(s)))
|
|
return append(dst, s...)
|
|
}
|
|
|
|
// AppendNullableString appends potentially nil string to dst prefixed with its
|
|
// int16 length or int16(-1) if nil.
|
|
func AppendNullableString(dst []byte, s *string) []byte {
|
|
if s == nil {
|
|
return AppendInt16(dst, -1)
|
|
}
|
|
return AppendString(dst, *s)
|
|
}
|
|
|
|
// AppendCompactNullableString appends a potentially nil string to dst with its
|
|
// uvarint length starting at 1, with 0 indicating null. Thus, the length is
|
|
// the decoded uvarint - 1.
|
|
//
|
|
// For KIP-482.
|
|
func AppendCompactNullableString(dst []byte, s *string) []byte {
|
|
if s == nil {
|
|
return AppendUvarint(dst, 0)
|
|
}
|
|
return AppendCompactString(dst, *s)
|
|
}
|
|
|
|
// AppendBytes appends bytes to dst prefixed with its int32 length.
|
|
func AppendBytes(dst, b []byte) []byte {
|
|
dst = AppendInt32(dst, int32(len(b)))
|
|
return append(dst, b...)
|
|
}
|
|
|
|
// AppendCompactBytes appends bytes to dst prefixed with a its uvarint length
|
|
// starting at 1; 0 is reserved for null, which compact bytes are not (nullable
|
|
// compact ones are!). Thus, the length is the decoded uvarint - 1.
|
|
//
|
|
// For KIP-482.
|
|
func AppendCompactBytes(dst, b []byte) []byte {
|
|
dst = AppendUvarint(dst, 1+uint32(len(b)))
|
|
return append(dst, b...)
|
|
}
|
|
|
|
// AppendNullableBytes appends a potentially nil slice to dst prefixed with its
|
|
// int32 length or int32(-1) if nil.
|
|
func AppendNullableBytes(dst, b []byte) []byte {
|
|
if b == nil {
|
|
return AppendInt32(dst, -1)
|
|
}
|
|
return AppendBytes(dst, b)
|
|
}
|
|
|
|
// AppendCompactNullableBytes appends a potentially nil slice to dst with its
|
|
// uvarint length starting at 1, with 0 indicating null. Thus, the length is
|
|
// the decoded uvarint - 1.
|
|
//
|
|
// For KIP-482.
|
|
func AppendCompactNullableBytes(dst, b []byte) []byte {
|
|
if b == nil {
|
|
return AppendUvarint(dst, 0)
|
|
}
|
|
return AppendCompactBytes(dst, b)
|
|
}
|
|
|
|
// AppendVarintString appends a string to dst prefixed with its length encoded
|
|
// as a varint.
|
|
func AppendVarintString(dst []byte, s string) []byte {
|
|
dst = AppendVarint(dst, int32(len(s)))
|
|
return append(dst, s...)
|
|
}
|
|
|
|
// AppendVarintBytes appends a slice to dst prefixed with its length encoded as
|
|
// a varint.
|
|
func AppendVarintBytes(dst, b []byte) []byte {
|
|
if b == nil {
|
|
return AppendVarint(dst, -1)
|
|
}
|
|
dst = AppendVarint(dst, int32(len(b)))
|
|
return append(dst, b...)
|
|
}
|
|
|
|
// AppendArrayLen appends the length of an array as an int32 to dst.
|
|
func AppendArrayLen(dst []byte, l int) []byte {
|
|
return AppendInt32(dst, int32(l))
|
|
}
|
|
|
|
// AppendCompactArrayLen appends the length of an array as a uvarint to dst
|
|
// as the length + 1.
|
|
//
|
|
// For KIP-482.
|
|
func AppendCompactArrayLen(dst []byte, l int) []byte {
|
|
return AppendUvarint(dst, 1+uint32(l))
|
|
}
|
|
|
|
// AppendNullableArrayLen appends the length of an array as an int32 to dst,
|
|
// or -1 if isNil is true.
|
|
func AppendNullableArrayLen(dst []byte, l int, isNil bool) []byte {
|
|
if isNil {
|
|
return AppendInt32(dst, -1)
|
|
}
|
|
return AppendInt32(dst, int32(l))
|
|
}
|
|
|
|
// AppendCompactNullableArrayLen appends the length of an array as a uvarint to
|
|
// dst as the length + 1; if isNil is true, this appends 0 as a uvarint.
|
|
//
|
|
// For KIP-482.
|
|
func AppendCompactNullableArrayLen(dst []byte, l int, isNil bool) []byte {
|
|
if isNil {
|
|
return AppendUvarint(dst, 0)
|
|
}
|
|
return AppendUvarint(dst, 1+uint32(l))
|
|
}
|
|
|
|
// Reader is used to decode Kafka messages.
|
|
//
|
|
// For all functions on Reader, if the reader has been invalidated, functions
|
|
// return defaults (false, 0, nil, ""). Use Complete to detect if the reader
|
|
// was invalidated or if the reader has remaining data.
|
|
type Reader struct {
|
|
Src []byte
|
|
bad bool
|
|
}
|
|
|
|
// Bool returns a bool from the reader.
|
|
func (b *Reader) Bool() bool {
|
|
if len(b.Src) < 1 {
|
|
b.bad = true
|
|
b.Src = nil
|
|
return false
|
|
}
|
|
t := b.Src[0] != 0 // if '0', false
|
|
b.Src = b.Src[1:]
|
|
return t
|
|
}
|
|
|
|
// Int8 returns an int8 from the reader.
|
|
func (b *Reader) Int8() int8 {
|
|
if len(b.Src) < 1 {
|
|
b.bad = true
|
|
b.Src = nil
|
|
return 0
|
|
}
|
|
r := b.Src[0]
|
|
b.Src = b.Src[1:]
|
|
return int8(r)
|
|
}
|
|
|
|
// Int16 returns an int16 from the reader.
|
|
func (b *Reader) Int16() int16 {
|
|
if len(b.Src) < 2 {
|
|
b.bad = true
|
|
b.Src = nil
|
|
return 0
|
|
}
|
|
r := int16(binary.BigEndian.Uint16(b.Src))
|
|
b.Src = b.Src[2:]
|
|
return r
|
|
}
|
|
|
|
// Uint16 returns an uint16 from the reader.
|
|
func (b *Reader) Uint16() uint16 {
|
|
if len(b.Src) < 2 {
|
|
b.bad = true
|
|
b.Src = nil
|
|
return 0
|
|
}
|
|
r := binary.BigEndian.Uint16(b.Src)
|
|
b.Src = b.Src[2:]
|
|
return r
|
|
}
|
|
|
|
// Int32 returns an int32 from the reader.
|
|
func (b *Reader) Int32() int32 {
|
|
if len(b.Src) < 4 {
|
|
b.bad = true
|
|
b.Src = nil
|
|
return 0
|
|
}
|
|
r := int32(binary.BigEndian.Uint32(b.Src))
|
|
b.Src = b.Src[4:]
|
|
return r
|
|
}
|
|
|
|
// Int64 returns an int64 from the reader.
|
|
func (b *Reader) Int64() int64 {
|
|
return int64(b.readUint64())
|
|
}
|
|
|
|
// Uuid returns a uuid from the reader.
|
|
func (b *Reader) Uuid() [16]byte {
|
|
var r [16]byte
|
|
copy(r[:], b.Span(16))
|
|
return r
|
|
}
|
|
|
|
// Float64 returns a float64 from the reader.
|
|
func (b *Reader) Float64() float64 {
|
|
return math.Float64frombits(b.readUint64())
|
|
}
|
|
|
|
func (b *Reader) readUint64() uint64 {
|
|
if len(b.Src) < 8 {
|
|
b.bad = true
|
|
b.Src = nil
|
|
return 0
|
|
}
|
|
r := binary.BigEndian.Uint64(b.Src)
|
|
b.Src = b.Src[8:]
|
|
return r
|
|
}
|
|
|
|
// Uint32 returns a uint32 from the reader.
|
|
func (b *Reader) Uint32() uint32 {
|
|
if len(b.Src) < 4 {
|
|
b.bad = true
|
|
b.Src = nil
|
|
return 0
|
|
}
|
|
r := binary.BigEndian.Uint32(b.Src)
|
|
b.Src = b.Src[4:]
|
|
return r
|
|
}
|
|
|
|
// Varint returns a varint int32 from the reader.
|
|
func (b *Reader) Varint() int32 {
|
|
val, n := Varint(b.Src)
|
|
if n <= 0 {
|
|
b.bad = true
|
|
b.Src = nil
|
|
return 0
|
|
}
|
|
b.Src = b.Src[n:]
|
|
return val
|
|
}
|
|
|
|
// Varlong returns a varlong int64 from the reader.
|
|
func (b *Reader) Varlong() int64 {
|
|
val, n := Varlong(b.Src)
|
|
if n <= 0 {
|
|
b.bad = true
|
|
b.Src = nil
|
|
return 0
|
|
}
|
|
b.Src = b.Src[n:]
|
|
return val
|
|
}
|
|
|
|
// Uvarint returns a uvarint encoded uint32 from the reader.
|
|
func (b *Reader) Uvarint() uint32 {
|
|
val, n := Uvarint(b.Src)
|
|
if n <= 0 {
|
|
b.bad = true
|
|
b.Src = nil
|
|
return 0
|
|
}
|
|
b.Src = b.Src[n:]
|
|
return val
|
|
}
|
|
|
|
// Span returns l bytes from the reader.
|
|
func (b *Reader) Span(l int) []byte {
|
|
if len(b.Src) < l || l < 0 {
|
|
b.bad = true
|
|
b.Src = nil
|
|
return nil
|
|
}
|
|
r := b.Src[:l:l]
|
|
b.Src = b.Src[l:]
|
|
return r
|
|
}
|
|
|
|
// UnsafeString returns a Kafka string from the reader without allocating using
|
|
// the unsafe package. This must be used with care; note the string holds a
|
|
// reference to the original slice.
|
|
func (b *Reader) UnsafeString() string {
|
|
l := b.Int16()
|
|
return UnsafeString(b.Span(int(l)))
|
|
}
|
|
|
|
// String returns a Kafka string from the reader.
|
|
func (b *Reader) String() string {
|
|
l := b.Int16()
|
|
return string(b.Span(int(l)))
|
|
}
|
|
|
|
// UnsafeCompactString returns a Kafka compact string from the reader without
|
|
// allocating using the unsafe package. This must be used with care; note the
|
|
// string holds a reference to the original slice.
|
|
func (b *Reader) UnsafeCompactString() string {
|
|
l := int(b.Uvarint()) - 1
|
|
return UnsafeString(b.Span(l))
|
|
}
|
|
|
|
// CompactString returns a Kafka compact string from the reader.
|
|
func (b *Reader) CompactString() string {
|
|
l := int(b.Uvarint()) - 1
|
|
return string(b.Span(l))
|
|
}
|
|
|
|
// UnsafeNullableString returns a Kafka nullable string from the reader without
|
|
// allocating using the unsafe package. This must be used with care; note the
|
|
// string holds a reference to the original slice.
|
|
func (b *Reader) UnsafeNullableString() *string {
|
|
l := b.Int16()
|
|
if l < 0 {
|
|
return nil
|
|
}
|
|
s := UnsafeString(b.Span(int(l)))
|
|
return &s
|
|
}
|
|
|
|
// NullableString returns a Kafka nullable string from the reader.
|
|
func (b *Reader) NullableString() *string {
|
|
l := b.Int16()
|
|
if l < 0 {
|
|
return nil
|
|
}
|
|
s := string(b.Span(int(l)))
|
|
return &s
|
|
}
|
|
|
|
// UnsafeCompactNullableString returns a Kafka compact nullable string from the
|
|
// reader without allocating using the unsafe package. This must be used with
|
|
// care; note the string holds a reference to the original slice.
|
|
func (b *Reader) UnsafeCompactNullableString() *string {
|
|
l := int(b.Uvarint()) - 1
|
|
if l < 0 {
|
|
return nil
|
|
}
|
|
s := UnsafeString(b.Span(l))
|
|
return &s
|
|
}
|
|
|
|
// CompactNullableString returns a Kafka compact nullable string from the
|
|
// reader.
|
|
func (b *Reader) CompactNullableString() *string {
|
|
l := int(b.Uvarint()) - 1
|
|
if l < 0 {
|
|
return nil
|
|
}
|
|
s := string(b.Span(l))
|
|
return &s
|
|
}
|
|
|
|
// Bytes returns a Kafka byte array from the reader.
|
|
//
|
|
// This never returns nil.
|
|
func (b *Reader) Bytes() []byte {
|
|
l := b.Int32()
|
|
// This is not to spec, but it is not clearly documented and Microsoft
|
|
// EventHubs fails here. -1 means null, which should throw an
|
|
// exception. EventHubs uses -1 to mean "does not exist" on some
|
|
// non-nullable fields.
|
|
//
|
|
// Until EventHubs is fixed, we return an empty byte slice for null.
|
|
if l == -1 {
|
|
return []byte{}
|
|
}
|
|
return b.Span(int(l))
|
|
}
|
|
|
|
// CompactBytes returns a Kafka compact byte array from the reader.
|
|
//
|
|
// This never returns nil.
|
|
func (b *Reader) CompactBytes() []byte {
|
|
l := int(b.Uvarint()) - 1
|
|
if l == -1 { // same as above: -1 should not be allowed here
|
|
return []byte{}
|
|
}
|
|
return b.Span(l)
|
|
}
|
|
|
|
// NullableBytes returns a Kafka nullable byte array from the reader, returning
|
|
// nil as appropriate.
|
|
func (b *Reader) NullableBytes() []byte {
|
|
l := b.Int32()
|
|
if l < 0 {
|
|
return nil
|
|
}
|
|
r := b.Span(int(l))
|
|
return r
|
|
}
|
|
|
|
// CompactNullableBytes returns a Kafka compact nullable byte array from the
|
|
// reader, returning nil as appropriate.
|
|
func (b *Reader) CompactNullableBytes() []byte {
|
|
l := int(b.Uvarint()) - 1
|
|
if l < 0 {
|
|
return nil
|
|
}
|
|
r := b.Span(l)
|
|
return r
|
|
}
|
|
|
|
// ArrayLen returns a Kafka array length from the reader.
|
|
func (b *Reader) ArrayLen() int32 {
|
|
r := b.Int32()
|
|
// The min size of a Kafka type is a byte, so if we do not have
|
|
// at least the array length of bytes left, it is bad.
|
|
if len(b.Src) < int(r) {
|
|
b.bad = true
|
|
b.Src = nil
|
|
return 0
|
|
}
|
|
return r
|
|
}
|
|
|
|
// VarintArrayLen returns a Kafka array length from the reader.
|
|
func (b *Reader) VarintArrayLen() int32 {
|
|
r := b.Varint()
|
|
// The min size of a Kafka type is a byte, so if we do not have
|
|
// at least the array length of bytes left, it is bad.
|
|
if len(b.Src) < int(r) {
|
|
b.bad = true
|
|
b.Src = nil
|
|
return 0
|
|
}
|
|
return r
|
|
}
|
|
|
|
// CompactArrayLen returns a Kafka compact array length from the reader.
|
|
func (b *Reader) CompactArrayLen() int32 {
|
|
r := int32(b.Uvarint()) - 1
|
|
// The min size of a Kafka type is a byte, so if we do not have
|
|
// at least the array length of bytes left, it is bad.
|
|
if len(b.Src) < int(r) {
|
|
b.bad = true
|
|
b.Src = nil
|
|
return 0
|
|
}
|
|
return r
|
|
}
|
|
|
|
// VarintBytes returns a Kafka encoded varint array from the reader, returning
|
|
// nil as appropriate.
|
|
func (b *Reader) VarintBytes() []byte {
|
|
l := b.Varint()
|
|
if l < 0 {
|
|
return nil
|
|
}
|
|
return b.Span(int(l))
|
|
}
|
|
|
|
// UnsafeVarintString returns a Kafka encoded varint string from the reader
|
|
// without allocating using the unsafe package. This must be used with care;
|
|
// note the string holds a reference to the original slice.
|
|
func (b *Reader) UnsafeVarintString() string {
|
|
return UnsafeString(b.VarintBytes())
|
|
}
|
|
|
|
// VarintString returns a Kafka encoded varint string from the reader.
|
|
func (b *Reader) VarintString() string {
|
|
return string(b.VarintBytes())
|
|
}
|
|
|
|
// Complete returns ErrNotEnoughData if the source ran out while decoding.
|
|
func (b *Reader) Complete() error {
|
|
if b.bad {
|
|
return ErrNotEnoughData
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// Ok returns true if the reader is still ok.
|
|
func (b *Reader) Ok() bool {
|
|
return !b.bad
|
|
}
|
|
|
|
// UnsafeString returns the slice as a string using unsafe rule (6).
|
|
func UnsafeString(slice []byte) string {
|
|
var str string
|
|
strhdr := (*reflect.StringHeader)(unsafe.Pointer(&str)) //nolint:gosec // known way to convert slice to string
|
|
strhdr.Data = ((*reflect.SliceHeader)(unsafe.Pointer(&slice))).Data //nolint:gosec // known way to convert slice to string
|
|
strhdr.Len = len(slice)
|
|
return str
|
|
}
|