registrar

Documentation

Index

• type AckEvent
  • func NewAckEvent(events []*core.EventDescriptor) *AckEvent
  • func (e *AckEvent) Process(state map[core.Stream]*FileState)
• type DeletedEvent
  • func NewDeletedEvent(stream core.Stream) *DeletedEvent
  • func (e *DeletedEvent) Process(state map[core.Stream]*FileState)
• type DiscoverEvent
  • func NewDiscoverEvent(stream core.Stream, source string, offset int64, fileinfo os.FileInfo) *DiscoverEvent
  • func (e *DiscoverEvent) Process(state map[core.Stream]*FileState)
• type EventProcessor
• type EventSpool
  • func (r *EventSpool) Add(event EventProcessor)
  • func (r *EventSpool) Close()
  • func (r *EventSpool) Send()
• type EventSpooler
• type FileIdentity
• type FileInfo
  • func NewFileInfo(fileinfo os.FileInfo) *FileInfo
  • func (fs *FileInfo) SameAs(info os.FileInfo) bool
  • func (fs *FileInfo) Stat() os.FileInfo
  • func (fs *FileInfo) Update(fileinfo os.FileInfo, identity *FileIdentity)
• type FileState
  • func (fs *FileState) Stat() os.FileInfo
  • func (fs *FileState) Update(fileinfo os.FileInfo, identity *FileIdentity)
• type FileStateOS
  • func (fs *FileStateOS) PopulateFileIds(info os.FileInfo)
  • func (fs *FileStateOS) SameAs(info os.FileInfo) bool
• type LoadPreviousFunc
• type Registrar
  • func NewRegistrar(pipeline *core.Pipeline, persistdir string) *Registrar
  • func (r *Registrar) Connect() EventSpooler
  • func (r *Registrar) LoadPrevious(callback_func LoadPreviousFunc) (have_previous bool, err error)
  • func (r *Registrar) Run()
• type Registrator
• type RenamedEvent
  • func NewRenamedEvent(stream core.Stream, source string) *RenamedEvent
  • func (e *RenamedEvent) Process(state map[core.Stream]*FileState)

Types

type AckEvent

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

AckEvent is a registrar ack event which triggers an update to the saved resume offsets for a file

func NewAckEvent

func NewAckEvent(events []*core.EventDescriptor) *AckEvent

NewAckEvent creates a new registrar ack event

func (*AckEvent) Process

func (e *AckEvent) Process(state map[core.Stream]*FileState)

Process persists the ack event into the registrar state by storing the offset

type DeletedEvent

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

func NewDeletedEvent

func NewDeletedEvent(stream core.Stream) *DeletedEvent

func (*DeletedEvent) Process

func (e *DeletedEvent) Process(state map[core.Stream]*FileState)

type DiscoverEvent

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

func NewDiscoverEvent

func NewDiscoverEvent(stream core.Stream, source string, offset int64, fileinfo os.FileInfo) *DiscoverEvent

func (*DiscoverEvent) Process

func (e *DiscoverEvent) Process(state map[core.Stream]*FileState)

type EventProcessor

type EventProcessor interface {
	Process(state map[core.Stream]*FileState)
}

type EventSpool

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

func (*EventSpool) Add

func (r *EventSpool) Add(event EventProcessor)

func (*EventSpool) Close

func (r *EventSpool) Close()

func (*EventSpool) Send

func (r *EventSpool) Send()

type EventSpooler

type EventSpooler interface {
	Close()
	Add(EventProcessor)
	Send()
}

type FileIdentity

type FileIdentity interface {
	SameAs(os.FileInfo) bool
	Stat() os.FileInfo
	Update(os.FileInfo, *FileIdentity)
}

type FileInfo

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

func NewFileInfo

func NewFileInfo(fileinfo os.FileInfo) *FileInfo

func (*FileInfo) SameAs

func (fs *FileInfo) SameAs(info os.FileInfo) bool

func (*FileInfo) Stat

func (fs *FileInfo) Stat() os.FileInfo

func (*FileInfo) Update

func (fs *FileInfo) Update(fileinfo os.FileInfo, identity *FileIdentity)

type FileState

type FileState struct {
	FileStateOS
	Source *string `json:"source,omitempty"`
	Offset int64   `json:"offset,omitempty"`
}

func (*FileState) Stat

func (fs *FileState) Stat() os.FileInfo

func (*FileState) Update

func (fs *FileState) Update(fileinfo os.FileInfo, identity *FileIdentity)

type FileStateOS

type FileStateOS struct {
	Inode  uint64 `json:"inode,omitempty"`
	Device uint64 `json:"device,omitempty"`
}

func (*FileStateOS) PopulateFileIds

func (fs *FileStateOS) PopulateFileIds(info os.FileInfo)

func (*FileStateOS) SameAs

func (fs *FileStateOS) SameAs(info os.FileInfo) bool

type LoadPreviousFunc

type LoadPreviousFunc func(string, *FileState) (core.Stream, error)

type Registrar

type Registrar struct {
	core.PipelineSegment

	sync.Mutex
	// contains filtered or unexported fields
}

func NewRegistrar

func NewRegistrar(pipeline *core.Pipeline, persistdir string) *Registrar

func (*Registrar) Connect

func (r *Registrar) Connect() EventSpooler

func (*Registrar) LoadPrevious

func (r *Registrar) LoadPrevious(callback_func LoadPreviousFunc) (have_previous bool, err error)

func (*Registrar) Run

func (r *Registrar) Run()

type Registrator

type Registrator interface {
	Connect() EventSpooler
	LoadPrevious(LoadPreviousFunc) (bool, error)
}

type RenamedEvent

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

func NewRenamedEvent

func NewRenamedEvent(stream core.Stream, source string) *RenamedEvent

func (*RenamedEvent) Process

func (e *RenamedEvent) Process(state map[core.Stream]*FileState)