eventlog

Documentation

Index

• Constants
• Variables
• func New(cfg StorageConfig, fss ...afero.Fs) (*eventManager, error)
• func NewDummy() *dummyEventManager
• type Event
  • func (e Event) IntoProto() *proto.FetchEventsResponse
• type EventManager
• type EventPusher
• type EventSubscriber
• type EventType
• type EventlogPosition
• type File
• type StorageConfig
• type SubscribeOption
  • func WithActiveLog() SubscribeOption
  • func WithPosition(position EventlogPosition) SubscribeOption
  • func WithSkipEventAtPosition(skip bool) SubscribeOption
• type Subscription
  • func (s *Subscription) Close() <-chan struct{}
  • func (s *Subscription) Events() <-chan Event
• type WriteSyncCloser

Constants

const (
	MagicHI uint8 = 0xAA
	MagicLO uint8 = 0x55
)

Variables

var ErrAlreadySubscribed = errors.New("already subscribed")

var ErrNilEvent = errors.New("event is nil")

var ErrNotFound = errors.New("not found")

var ErrServiceStopped = errors.New("service stopped")

var ErrSubscriptionStopped = errors.New("subscription stopped")

Functions

func New

func New(cfg StorageConfig, fss ...afero.Fs) (*eventManager, error)

New initializes and starts the event log manager

func NewDummy

func NewDummy() *dummyEventManager

Types

type Event

type Event struct {
	Type      EventType `json:"type"`
	Timestamp int64     `json:"ts"`
	LogID     string    `json:"log_id"`
	Offset    int64     `json:"offset"`
	Data      []byte    `json:"data"`
}

func (Event) IntoProto

func (e Event) IntoProto() *proto.FetchEventsResponse

type EventManager

type EventManager interface {
	EventPusher
	EventSubscriber
	control.ServiceController
}

type EventPusher

type EventPusher interface {
	Push(eventType EventType, data interface{}) error
}

type EventSubscriber

type EventSubscriber interface {
	Subscribe(ctx context.Context, subscriberId string, opts ...SubscribeOption) (*Subscription, error)
	Unsubscribe(ctx context.Context, subscriberId string) error
}

type EventType

type EventType int32

const (
	Unspecified      EventType = EventType(proto.EventType_Unspecified)
	PeerAdd          EventType = EventType(proto.EventType_PeerAdd)
	PeerRemove       EventType = EventType(proto.EventType_PeerRemove)
	PeerUpdate       EventType = EventType(proto.EventType_PeerUpdate)
	PeerTraffic      EventType = EventType(proto.EventType_PeerTraffic)
	PeerFirstConnect EventType = EventType(proto.EventType_PeerFirstConnect)
)

type EventlogPosition

type EventlogPosition struct {
	LogID  string
	Offset int64
}

EventlogPosition describe where we want to start reading the log. Zero offset means the beginning of the file. Empty logID means the beginning of the whole journal.

type File

type File interface {
	io.Reader
	io.Writer
	io.Seeker
	io.Closer
	Stat() (os.FileInfo, error)
	Sync() error
}

type StorageConfig

type StorageConfig struct {
	// path to a log dir
	Dir string `json:"dir"`
	// number of files to maintain
	MaxFiles int `json:"max_files"`
	// for how long we want to write to a single logfile
	Period time.Duration `json:"period"`
	// how many bytes we want to write to a single logfile
	Size int64 `json:"size"`
}

type SubscribeOption

type SubscribeOption func(opts *subscribeOptions) error

func WithActiveLog

func WithActiveLog() SubscribeOption

Started from Active log disregarding the Position

func WithPosition

func WithPosition(position EventlogPosition) SubscribeOption

Used only if ActiveLog is not defined

func WithSkipEventAtPosition

func WithSkipEventAtPosition(skip bool) SubscribeOption

type Subscription

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

func (*Subscription) Close

func (s *Subscription) Close() <-chan struct{}

Close the subscriber. Might be called multiple times.

func (*Subscription) Events

func (s *Subscription) Events() <-chan Event

type WriteSyncCloser

type WriteSyncCloser interface {
	io.Writer
	io.Closer
	Sync() error
}