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 func(item interface{}, kafkaOffset int64) error) error
  • func (t *Tailer) Read(num int64, offset int64) ([]interface{}, error)
  • func (t *Tailer) RegisterConsumeHook(tailHook TailMessageHook)
  • func (t *Tailer) Start() error
  • func (t *Tailer) Stop()

Constants

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 (
	StopErr = errors.New("Iteration stopped.")
)

Types

type TailMessageHook

type TailMessageHook func(message *sarama.ConsumerMessage)

type Tailer

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

Tailer retrieves the last n messages from a given topic

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 func(item interface{}, kafkaOffset int64) error) 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) ([]interface{}, error)

func (*Tailer) RegisterConsumeHook

func (t *Tailer) RegisterConsumeHook(tailHook TailMessageHook)

RegisterConsumeHook registers a TailMessageHook hook

func (*Tailer) Start

func (t *Tailer) Start() error

Start starts the tailer

func (*Tailer) Stop

func (t *Tailer) Stop()

Stop the tailer