types

type Action interface {
	// PerformAction does something with matched messages.
	PerformAction(ctx context.Context, messages []Message) error
}

type Fields map[string]string

type Formatter interface {
	// Format formats the message.
	Format(*bytes.Buffer, Message) error
}

type FormatterFunc func(*bytes.Buffer, Message) error

type Matcher interface {
	MatchMessage(message Message) bool
}

type Message struct {
	Timestamp time.Time `json:"ts"`                  // Timestamp of the event.
	Cursor    string    `json:"cursor,omitempty"`    // Cursor position of the message.
	Fields    Fields    `json:"fields,omitempty"`    // Fields contains message fields.
	Source    Source    `json:"source,omitempty"`    // Source the message was read from.
	ReaderID  ReaderID  `json:"reader_id,omitempty"` // ReaderID that read the message.
}

type Persister interface {
	// LoadState loads the reader state. When the state doees not exist, it must
	// return no error.
	LoadState(context.Context, ReaderID) (State, error)
	// SaveSave saves the reader state.
	SaveState(context.Context, ReaderID, State) error
}

type Processor interface {
	// ProcessMessage processes the message read.
	ProcessMessage(context.Context, Message) error
	Tick(context.Context, time.Time) error
}

type ReadLogsParams struct {
	State State          // State for resuming reading.
	Ch    chan<- Message // Ch is a channel to write the messages to.
}

type Reader interface {
	// ReaderID returns the reader's ID.
	ReaderID() ReaderID
	// ReadLogs reads logs until context is done, or an error is encountered.
	// Implementations must not close the ReadLogsParams.Ch as that is done
	// conditionally in Watcher.
	ReadLogs(context.Context, ReadLogsParams) error
}

type ReaderID string

type ReaderParams struct {
	ReaderID ReaderID
	Logger   log.Logger
}

type Source int

type State struct {
	// Timestamp is the last timestamp read.
	Timestamp time.Time `json:"time"`
	// NumMessages is the number of messages read with the same Timestamp.
	NumMessages int `json:"num_messages"`
	// Cursor is the current cursor, if any.
	Cursor string `json:"cursor,omitempty"`
}