buffer

Documentation

Index

• Constants
• type Chunk
  • func NewChunk(content []byte, startIdx, payloadSize int32, isEOF byte, seqID int32) *Chunk
  • func NewChunkFromRawBytes(bytes []byte) (*Chunk, error)
  • func NewEmptyChunk(seqID int32) *Chunk
  • func NewUninitializedChunk(payloadSize int32) *Chunk
  • func (chunk *Chunk) Bytes() []byte
  • func (chunk *Chunk) IsEOF() bool
  • func (chunk *Chunk) IsEmpty() bool
  • func (chunk *Chunk) IsValid() bool
  • func (chunk *Chunk) Payload() []byte
  • func (chunk *Chunk) SeqID() int32
  • func (chunk Chunk) String() string
• type RecvBuffer
  • func NewRecvBuffer(config RecvBufferConfig, rawStream cmn.ReadWriteCloser, ...) RecvBuffer
  • func (rb *RecvBuffer) GetSize() int
  • func (rb *RecvBuffer) Read() ([]byte, error)
  • func (rb *RecvBuffer) Start(ctx context.Context) bool
  • func (rb *RecvBuffer) Stop()
  • func (rb *RecvBuffer) Wait()
• type RecvBufferConfig
  • func GetDefaultRecvBufferConfig() RecvBufferConfig
• type SendBuffer
  • func NewSendBuffer(config SendBufferConfig, rawStream cmn.ReadWriteCloser, ...) SendBuffer
  • func (sb *SendBuffer) CanInsert() bool
  • func (sb *SendBuffer) GetSize() int
  • func (sb *SendBuffer) IsEmpty() bool
  • func (sb *SendBuffer) Start(ctx context.Context) bool
  • func (sb *SendBuffer) Stop()
  • func (sb *SendBuffer) Wait()
  • func (sb *SendBuffer) Write(bytes []byte) bool
• type SendBufferConfig
  • func GetDefaultSendBufferConfig() SendBufferConfig

Constants

const MaxErrorInARow = 3

Types

type Chunk

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

Chunk Format:

bytes[0..3]  : seqID, 32 bits (int32)
bytes[4..7]  : payloadSize, 32 bits (int32)
bytes[8]     : isEOF, 8 bits
bytes[9..15] : reserved
bytes[16..payloadSize+15]: the actual payload

                    Chunk Bit Map

0 1 2 3 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ | seqID (32 bits) | +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ | payloadSize (32 bits) | +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ | isEOF | reserved | +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ | reserved | +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ | | | payload | | | +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+

func NewChunk

func NewChunk(content []byte, startIdx, payloadSize int32, isEOF byte, seqID int32) *Chunk

NewChunk will create a chunk from the given "content" byte slice. It extracts the subslice content[startIdx:startIdx+payloadSize] as the payload of the chunk, and adds the meta data (isEOF, seqID) after the payload

func NewChunkFromRawBytes

func NewChunkFromRawBytes(bytes []byte) (*Chunk, error)

func NewEmptyChunk

func NewEmptyChunk(seqID int32) *Chunk

func NewUninitializedChunk

func NewUninitializedChunk(payloadSize int32) *Chunk

func (*Chunk) Bytes

func (chunk *Chunk) Bytes() []byte

func (*Chunk) IsEOF

func (chunk *Chunk) IsEOF() bool

func (*Chunk) IsEmpty

func (chunk *Chunk) IsEmpty() bool

func (*Chunk) IsValid

func (chunk *Chunk) IsValid() bool

func (*Chunk) Payload

func (chunk *Chunk) Payload() []byte

func (*Chunk) SeqID

func (chunk *Chunk) SeqID() int32

func (Chunk) String

func (chunk Chunk) String() string

type RecvBuffer

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

func NewRecvBuffer

func NewRecvBuffer(config RecvBufferConfig, rawStream cmn.ReadWriteCloser, onError cmn.ErrorHandler) RecvBuffer

NewRecvBuffer creates a RecvBuffer instance for the given config

func (*RecvBuffer) GetSize

func (rb *RecvBuffer) GetSize() int

GetSize returns the size of the SendBuffer. It is goroutine safe

func (*RecvBuffer) Read

func (rb *RecvBuffer) Read() ([]byte, error)

Read blocks until a message can be retrived from the queue

func (*RecvBuffer) Start

func (rb *RecvBuffer) Start(ctx context.Context) bool

func (*RecvBuffer) Stop

func (rb *RecvBuffer) Stop()

Stop is called when the RecvBuffer stops

func (*RecvBuffer) Wait

func (rb *RecvBuffer) Wait()

Wait suspends the caller goroutine

type RecvBufferConfig

type RecvBufferConfig struct {
	RecvRate int64
	// contains filtered or unexported fields
}

func GetDefaultRecvBufferConfig

func GetDefaultRecvBufferConfig() RecvBufferConfig

GetDefaultRecvBufferConfig returns the default config for the RecvBuffer

type SendBuffer

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

func NewSendBuffer

func NewSendBuffer(config SendBufferConfig, rawStream cmn.ReadWriteCloser, onError cmn.ErrorHandler) SendBuffer

NewSendBuffer creates a SendBuffer instance for the given config

func (*SendBuffer) CanInsert

func (sb *SendBuffer) CanInsert() bool

CanInsert return whether more bytes can be inserted into the send buffer. It is goroutine safe

func (*SendBuffer) GetSize

func (sb *SendBuffer) GetSize() int

GetSize returns the size of the SendBuffer. It is goroutine safe

func (*SendBuffer) IsEmpty

func (sb *SendBuffer) IsEmpty() bool

IsEmpty indicates whether the SendBuffer is empty

func (*SendBuffer) Start

func (sb *SendBuffer) Start(ctx context.Context) bool

func (*SendBuffer) Stop

func (sb *SendBuffer) Stop()

Stop is called when the SendBuffer stops

func (*SendBuffer) Wait

func (sb *SendBuffer) Wait()

Wait suspends the caller goroutine

func (*SendBuffer) Write

func (sb *SendBuffer) Write(bytes []byte) bool

Write insert the bytes to queue, and times out after the configured timeout. It is goroutine safe

type SendBufferConfig

type SendBufferConfig struct {
	SendRate int64

	ChunkBatchSize int64
	// contains filtered or unexported fields
}

func GetDefaultSendBufferConfig

func GetDefaultSendBufferConfig() SendBufferConfig

GetDefaultSendBufferConfig returns the default config for the SendBuffer