log_buffer

Documentation

Index

• Constants
• Variables
• type LogBuffer
  • func NewLogBuffer(flushInterval time.Duration, ...) *LogBuffer
  • func (m *LogBuffer) AddToBuffer(partitionKey, data []byte, eventTsNs int64)
  • func (logBuffer *LogBuffer) LoopProcessLogData(startTreadTime time.Time, waitForDataFn func() bool, ...) (lastReadTime time.Time, err error)
  • func (m *LogBuffer) ReadFromBuffer(lastReadTime time.Time) (bufferCopy *bytes.Buffer)
  • func (m *LogBuffer) ReleaseMemory(b *bytes.Buffer)
  • func (m *LogBuffer) Shutdown()
• type MemBuffer
  • func (mb *MemBuffer) String() string
• type SealedBuffers
  • func (sbs *SealedBuffers) SealBuffer(startTime, stopTime time.Time, buf []byte, pos int) (newBuf []byte)

Constants

const BufferSize = 4 * 1024 * 1024

const PreviousBufferCount = 3

Variables

var (
	ResumeError = fmt.Errorf("resume")
)

Types

type LogBuffer

type LogBuffer struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

func NewLogBuffer

func NewLogBuffer(flushInterval time.Duration, flushFn func(startTime, stopTime time.Time, buf []byte), notifyFn func()) *LogBuffer

func (*LogBuffer) AddToBuffer

func (m *LogBuffer) AddToBuffer(partitionKey, data []byte, eventTsNs int64)

func (*LogBuffer) LoopProcessLogData

func (logBuffer *LogBuffer) LoopProcessLogData(
	startTreadTime time.Time,
	waitForDataFn func() bool,
	eachLogDataFn func(logEntry *filer_pb.LogEntry) error) (lastReadTime time.Time, err error)

func (*LogBuffer) ReadFromBuffer

func (m *LogBuffer) ReadFromBuffer(lastReadTime time.Time) (bufferCopy *bytes.Buffer)

func (*LogBuffer) ReleaseMemory

func (m *LogBuffer) ReleaseMemory(b *bytes.Buffer)

func (*LogBuffer) Shutdown

func (m *LogBuffer) Shutdown()

type MemBuffer

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

func (*MemBuffer) String

func (mb *MemBuffer) String() string

type SealedBuffers

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

func (*SealedBuffers) SealBuffer

func (sbs *SealedBuffers) SealBuffer(startTime, stopTime time.Time, buf []byte, pos int) (newBuf []byte)