core

Documentation

Index

• Constants
• type BufferItem
• type DataCallBack
• type DataChunk
  • func (d *DataChunk) Bytes() []byte
  • func (d *DataChunk) Release()
• type DoubleBuffer
  • func NewDoubleBuffer(size int32, sc *config.SwitchCondition, opts ...Options) (*DoubleBuffer, error)
  • func (d *DoubleBuffer) BatchRead(_ context.Context, _ int) [][]byte
  • func (d *DoubleBuffer) BlockingRead(ctx context.Context) (DataChunk, error)
  • func (d *DoubleBuffer) Close()
  • func (d *DoubleBuffer) RegisterCallback(_ DataCallBack) UnregisterFunc
  • func (d *DoubleBuffer) RegisterReadMode(readMode chanjet.ReadMode) error
  • func (d *DoubleBuffer) Write(p []byte) error
• type MinHeap
  • func (m MinHeap) Len() int
  • func (m MinHeap) Less(i, j int) bool
  • func (m *MinHeap) Pop() interface{}
  • func (m *MinHeap) Push(x interface{})
  • func (m MinHeap) Swap(i, j int)
• type MinHeapItem
• type Options
  • func WithMetrics(collector chanjet.CollectorType) Options
• type ReadBuffer
• type SmartBuffer
  • func (s *SmartBuffer) Close()
  • func (s *SmartBuffer) Release(data []byte)
• type UnregisterFunc
• type WaiterManager
  • func (w *WaiterManager) Close()
• type WrapHeap
  • func NewWrapHeap() *WrapHeap
  • func (h *WrapHeap) Len() int
  • func (h *WrapHeap) PeekFirst() *MinHeapItem
  • func (h *WrapHeap) Pick() *MinHeapItem
  • func (h *WrapHeap) Push(item *MinHeapItem)

Constants

const (
	SmallDataThreshold      = 1024            // Small data threshold (<1KB)
	LargeDataThreshold      = 32 * 1024       // Big data threshold (>32KB)
	MediumDataCacheDuration = 5 * time.Second // Cache time for medium-sized data
	SwitchCheckInterval     = 5 * time.Millisecond
)

Types

type BufferItem

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

type DataCallBack

type DataCallBack func(data []byte)

type DataChunk

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

func (*DataChunk) Bytes

func (d *DataChunk) Bytes() []byte

func (*DataChunk) Release

func (d *DataChunk) Release()

type DoubleBuffer

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

DoubleBuffer Double buffer design

func NewDoubleBuffer

func NewDoubleBuffer(size int32, sc *config.SwitchCondition, opts ...Options) (*DoubleBuffer, error)

func (*DoubleBuffer) BatchRead

func (d *DoubleBuffer) BatchRead(_ context.Context, _ int) [][]byte

func (*DoubleBuffer) BlockingRead

func (d *DoubleBuffer) BlockingRead(ctx context.Context) (DataChunk, error)

BlockingRead Blocking reading requires passing in a Context with timeout control. If there is data, read the data immediately and return. If there is no data, wait for new data in a blocking manner until the context times out or the channel is closed. It will not block forever.

func (*DoubleBuffer) Close

func (d *DoubleBuffer) Close()

func (*DoubleBuffer) RegisterCallback

func (d *DoubleBuffer) RegisterCallback(_ DataCallBack) UnregisterFunc

func (*DoubleBuffer) RegisterReadMode

func (d *DoubleBuffer) RegisterReadMode(readMode chanjet.ReadMode) error

func (*DoubleBuffer) Write

func (d *DoubleBuffer) Write(p []byte) error

type MinHeap

type MinHeap []*MinHeapItem

func (MinHeap) Len

func (m MinHeap) Len() int

func (MinHeap) Less

func (m MinHeap) Less(i, j int) bool

func (*MinHeap) Pop

func (m *MinHeap) Pop() interface{}

func (*MinHeap) Push

func (m *MinHeap) Push(x interface{})

func (MinHeap) Swap

func (m MinHeap) Swap(i, j int)

type MinHeapItem

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

type Options

type Options func(buffer *DoubleBuffer) error

func WithMetrics

func WithMetrics(collector chanjet.CollectorType) Options

WithMetrics Enable indicator collection and specify the collector type

type ReadBuffer

type ReadBuffer interface {
	RegisterReadMode(readMode chanjet.ReadMode) error
	BlockingRead(ctx context.Context) ([]byte, error)
	RegisterCallback(cb DataCallBack) UnregisterFunc
	BatchRead(ctx context.Context, batchSize int) [][]byte
}

type SmartBuffer

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

func (*SmartBuffer) Close

func (s *SmartBuffer) Close()

func (*SmartBuffer) Release

func (s *SmartBuffer) Release(data []byte)

Release the data read by the zero-copy API

type UnregisterFunc

type UnregisterFunc func()

type WaiterManager

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

func (*WaiterManager) Close

func (w *WaiterManager) Close()

type WrapHeap

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

func NewWrapHeap

func NewWrapHeap() *WrapHeap

func (*WrapHeap) Len

func (h *WrapHeap) Len() int

func (*WrapHeap) PeekFirst

func (h *WrapHeap) PeekFirst() *MinHeapItem

func (*WrapHeap) Pick

func (h *WrapHeap) Pick() *MinHeapItem

func (*WrapHeap) Push

func (h *WrapHeap) Push(item *MinHeapItem)