message

Documentation

Index

• Constants
• Variables
• type BasicHeader
  • func ParseBasicHeader(prs binary.Parser, b []byte) (BasicHeader, error)
  • func PeakBasicHeader(rb *binary.ReadBuffer) (BasicHeader, error)
  • func (b *BasicHeader) ReadHeader(rb *binary.ReadBuffer, payloadSize uint32) error
• type Bundle
• type Chunk
• type ConsumeRequest
  • func (f *ConsumeRequest) WriteToBuffer(w *binary.WriteBuffer) error
• type ConsumeResponse
  • func (fr *ConsumeResponse) Consume(rb *binary.ReadBuffer, payloadSize uint32, msgLog chan<- Log) error
  • func (fr *ConsumeResponse) GetTopicsLatestSequenceNumber(rb *binary.ReadBuffer) (map[string]uint64, error)
• type DiscoverPartitionsRequest
• type DiscoverPartitionsResponse
  • func (bh *DiscoverPartitionsResponse) ReadFromBuffer(rb *binary.ReadBuffer, payloadSize uint32) error
• type FetchRequestTopic
• type FetchRequestTopicPartition
• type Log
• type Message
• type MessageError
  • func NewMessageError(inner error, msg string) *MessageError
  • func (r *MessageError) Error() string
  • func (r *MessageError) Unwrap() error
• type Partition
• type Ping
• type Topic
  • func (t *Topic) NextBundle(rb *binary.ReadBuffer) bool
• type TopicHeader
• type Type

Constants

const SizeOfBasicHeader = binary.SizeOfUint8Bytes + binary.SizeOfUint32Bytes

SizeOfBasicHeader = msgId:u8 + payload size:u32 .

Variables

var (
	ErrHighWaterMarkExceeded = errors.New("high water mark exceeded")
	ErrEmptyResponse         = errors.New("empty tank response")
)

Types

type BasicHeader

type BasicHeader struct {
	MessageType Type
	PayloadSize uint32
}

BasicHeader struct.

func ParseBasicHeader

func ParseBasicHeader(prs binary.Parser, b []byte) (BasicHeader, error)

ParseBasicHeader parses a basic header from a slice.

func PeakBasicHeader

func PeakBasicHeader(rb *binary.ReadBuffer) (BasicHeader, error)

PeakBasicHeader parses the first 5 bytes into basicHeader without discarding the bytes from read buffer.

func (*BasicHeader) ReadHeader

func (b *BasicHeader) ReadHeader(rb *binary.ReadBuffer, payloadSize uint32) error

ReadHeader parses the first 5 bytes into basicHeader.

type Bundle

type Bundle struct {
	BundleFlags uint8
	Messages    []Message
}

type Chunk

type Chunk struct {
	Bundles            []Bundle
	ConsumedFromBundle uint64
}

type ConsumeRequest

type ConsumeRequest struct {
	ClientVersion uint16
	RequestID     uint32
	Client        string
	MaxWaitMS     uint64
	MinBytes      uint32
	Topics        []FetchRequestTopic
}

ConsumeRequest message.

func (*ConsumeRequest) WriteToBuffer

func (f *ConsumeRequest) WriteToBuffer(w *binary.WriteBuffer) error

WriteToBuffer implements theWritable interface.

type ConsumeResponse

type ConsumeResponse struct {
	TopicHeader           TopicHeader
	TopicPartitionBaseSeq map[string]uint64
	// contains filtered or unexported fields
}

ConsumeResponse struct.

func (*ConsumeResponse) Consume

func (fr *ConsumeResponse) Consume(rb *binary.ReadBuffer, payloadSize uint32, msgLog chan<- Log) error

func (*ConsumeResponse) GetTopicsLatestSequenceNumber

func (fr *ConsumeResponse) GetTopicsLatestSequenceNumber(rb *binary.ReadBuffer) (map[string]uint64, error)

type DiscoverPartitionsRequest

type DiscoverPartitionsRequest struct{}

DiscoverPartitionsRequest struct.

type DiscoverPartitionsResponse

type DiscoverPartitionsResponse struct{}

DiscoverPartitionsResponse struct.

func (*DiscoverPartitionsResponse) ReadFromBuffer

func (bh *DiscoverPartitionsResponse) ReadFromBuffer(rb *binary.ReadBuffer, payloadSize uint32) error

ReadFromBuffer parses the first 5 bytes into basicHeader.

type FetchRequestTopic

type FetchRequestTopic struct {
	Name       string
	Partitions []FetchRequestTopicPartition
}

FetchRequestTopic struct.

type FetchRequestTopicPartition

type FetchRequestTopicPartition struct {
	PartitionID       int16
	ABSSequenceNumber uint64
	FetchSize         int32
}

FetchRequestTopicPartition struct.

type Log

type Log struct {
	CreatedAt uint64
	SeqNumber uint64
	Key       []byte
	Payload   []byte
}

type Message

type Message struct {
	Type Type
}

type MessageError

type MessageError struct {
	// contains filtered or unexported fields
}

func NewMessageError

func NewMessageError(inner error, msg string) *MessageError

NewMessage constructs a message error.

func (*MessageError) Error

func (r *MessageError) Error() string

Error function.

func (*MessageError) Unwrap

func (r *MessageError) Unwrap() error

Unwrap function.

type Partition

type Partition struct {
	PartitionID       uint16
	ErrorOrFlags      uint8
	HighWaterMark     uint64
	ChuckLength       uint32
	FirstAvailSeqNum  uint64
	Chunk             Chunk
	ConsumedFromChunk uint64
}

type Ping

type Ping struct{}

Ping struct.

type Topic

type Topic struct {
	Name            string
	TotalPartitions uint8
	Partition       Partition
	LogBaseSeqNum   uint64
	MessageEnd      uint64
}

func (*Topic) NextBundle

func (t *Topic) NextBundle(rb *binary.ReadBuffer) bool

type TopicHeader

type TopicHeader struct {
	RequestID   uint32
	TopicsCount uint8
	Topics      []Topic
}

type Type

type Type uint8

Type alias.

const (
	TypeUnknown            Type = 0
	TypeConsume            Type = 2
	TypeProduce            Type = 1
	TypeProduce5           Type = 5
	TypePing               Type = 3
	TypeDiscoverPartitions Type = 6
	TypeCreateTopic        Type = 7
	TypeReloadConf         Type = 8
	TypeConsumePeer        Type = 9
	TypeStatus             Type = 10
)

Message types id.