tailer

Documentation

Index

• Constants
• Variables
• type TailMessageHook
• type Tailer
  • func NewTailer(brokers []string, topic string, size int, codec goka.Codec) (*Tailer, error)
  • func (t *Tailer) IterateReverse(maxOffset int64, visit TailerVisiter) error
  • func (t *Tailer) Read(num int64, offset int64) ([]*TailerItem, error)
  • func (t *Tailer) RegisterConsumeHook(tailHook TailMessageHook)
  • func (t *Tailer) Start() error
  • func (t *Tailer) Stop()
• type TailerItem
• type TailerVisiter

Constants

const AllItems = -1

AllItems indicates all items from the tailer should be read

const (
	EndOffset int64 = -1
)

EndOffset specifies the largest offset. It is used to tell IterateReverse to return all items independent of their offset

Variables

var (
	// ErrStop indicates the iteration should be stopped. This can be returned from the iterate-Visitor
	ErrStop = errors.New("iteration stopped")
)

Types

type TailMessageHook

type TailMessageHook func(item *TailerItem)

TailMessageHook is called on every message being added to the tailer

type Tailer

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

Tailer updates messages from a topic and keeps them in a ring buffer for reverse iteration

func NewTailer

func NewTailer(brokers []string, topic string, size int, codec goka.Codec) (*Tailer, error)

NewTailer creates a new tailer for passed topic, size and codec

func (*Tailer) IterateReverse

func (t *Tailer) IterateReverse(maxOffset int64, visit TailerVisiter) error

IterateReverse iterates over all items ignoring items having bigger offset than maxOffset (or all, if EndOffset is given)

func (*Tailer) Read

func (t *Tailer) Read(num int64, offset int64) ([]*TailerItem, error)

func (*Tailer) RegisterConsumeHook

func (t *Tailer) RegisterConsumeHook(tailHook TailMessageHook)

RegisterConsumeHook sets the callback that will be called on every message being added to the tailer

func (*Tailer) Start

func (t *Tailer) Start() error

Start starts the tailer

func (*Tailer) Stop

func (t *Tailer) Stop()

Stop the tailer

type TailerItem

type TailerItem struct {
	// Key is the key of the original message
	Key string
	// Value is the decoded value. This will not be nil, as the tailer ignores nils
	Value interface{}
	// Offset is the message's offset
	Offset int64
}

TailerItem represents a decoded messagei n the tailer's ring buffer

type TailerVisiter

type TailerVisiter func(item *TailerItem) error

TailerVisiter is called on reverseiterate for every message