buffer

Documentation

Index

• type Buffer
  • func New(ctx context.Context, topic string, partition int32, config *Config) (*Buffer, error)
  • func (b *Buffer) Age() time.Duration
  • func (b *Buffer) Bytes() int64
  • func (b *Buffer) Close() error
  • func (b *Buffer) Flush() error
  • func (b *Buffer) Partition() int32
  • func (b *Buffer) Topic() string
  • func (b *Buffer) Write(data []byte) (int, error)
  • func (b *Buffer) WriteMessage(msg *sarama.ConsumerMessage) (int, error)
  • func (b *Buffer) Writes() int64
• type Config
  • func (c *Config) Validate() error
• type Flush
• type Reason

Types

type Buffer

type Buffer struct {
	*Config

	sync.RWMutex
	// contains filtered or unexported fields
}

Buffer represents a 1:N on-disk buffer.

func New

func New(ctx context.Context, topic string, partition int32, config *Config) (*Buffer, error)

New buffer for topic and partition. The path given is used for the base of the filenames created, which append ".{pid}.{id}.{fid}".

func (*Buffer) Age

func (b *Buffer) Age() time.Duration

func (*Buffer) Bytes

func (b *Buffer) Bytes() int64

Bytes returns the number of bytes made to the current file.

func (*Buffer) Close

func (b *Buffer) Close() error

Close the underlying file after flushing.

func (*Buffer) Flush

func (b *Buffer) Flush() error

Flush forces a flush.

func (*Buffer) Partition

func (b *Buffer) Partition() int32

func (*Buffer) Topic

func (b *Buffer) Topic() string

func (*Buffer) Write

func (b *Buffer) Write(data []byte) (int, error)

Write implements io.Writer.

func (*Buffer) WriteMessage

func (b *Buffer) WriteMessage(msg *sarama.ConsumerMessage) (int, error)

func (*Buffer) Writes

func (b *Buffer) Writes() int64

Writes returns the number of writes made to the current file.

type Config

type Config struct {
	TempDir       string        // Where to store temporary files
	FlushWrites   int64         // Flush after N writes, zero to disable
	FlushBytes    int64         // Flush after N bytes, zero to disable
	FlushInterval time.Duration // Flush after duration, zero to disable
	BufferSize    int           // Buffer size for writes
	Queue         chan *Flush   // Queue of flushed files
}

Config for disk buffer.

func (*Config) Validate

func (c *Config) Validate() error

Validate the configuration.

type Flush

type Flush struct {
	Reason     Reason
	Path       string
	Writes     int64
	Bytes      int64
	Opened     time.Time
	Closed     time.Time
	Age        time.Duration
	Topic      string
	Partition  int32
	LastOffset int64
	Ctx        context.Context
}

Flush represents a flushed file.

type Reason

type Reason string

Reason for flush.

const (
	Forced   Reason = "forced"
	Writes   Reason = "writes"
	Bytes    Reason = "bytes"
	Interval Reason = "interval"
)

Flush reasons.