12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394 |
- package sarama
- import (
- "fmt"
- "github.com/rcrowley/go-metrics"
- )
- // Encoder is the interface that wraps the basic Encode method.
- // Anything implementing Encoder can be turned into bytes using Kafka's encoding rules.
- type encoder interface {
- encode(pe packetEncoder) error
- }
- type encoderWithHeader interface {
- encoder
- headerVersion() int16
- }
- // Encode takes an Encoder and turns it into bytes while potentially recording metrics.
- func encode(e encoder, metricRegistry metrics.Registry) ([]byte, error) {
- if e == nil {
- return nil, nil
- }
- var prepEnc prepEncoder
- var realEnc realEncoder
- err := e.encode(&prepEnc)
- if err != nil {
- return nil, err
- }
- if prepEnc.length < 0 || prepEnc.length > int(MaxRequestSize) {
- return nil, PacketEncodingError{fmt.Sprintf("invalid request size (%d)", prepEnc.length)}
- }
- realEnc.raw = make([]byte, prepEnc.length)
- realEnc.registry = metricRegistry
- err = e.encode(&realEnc)
- if err != nil {
- return nil, err
- }
- return realEnc.raw, nil
- }
- // decoder is the interface that wraps the basic Decode method.
- // Anything implementing Decoder can be extracted from bytes using Kafka's encoding rules.
- type decoder interface {
- decode(pd packetDecoder) error
- }
- type versionedDecoder interface {
- decode(pd packetDecoder, version int16) error
- }
- // decode takes bytes and a decoder and fills the fields of the decoder from the bytes,
- // interpreted using Kafka's encoding rules.
- func decode(buf []byte, in decoder) error {
- if buf == nil {
- return nil
- }
- helper := realDecoder{raw: buf}
- err := in.decode(&helper)
- if err != nil {
- return err
- }
- if helper.off != len(buf) {
- return PacketDecodingError{"invalid length"}
- }
- return nil
- }
- func versionedDecode(buf []byte, in versionedDecoder, version int16) error {
- if buf == nil {
- return nil
- }
- helper := realDecoder{raw: buf}
- err := in.decode(&helper, version)
- if err != nil {
- return err
- }
- if helper.off != len(buf) {
- return PacketDecodingError{"invalid length"}
- }
- return nil
- }
|