spooler

Documentation

Overview

Extract ids events from a logging directory. Scans directory for new files and adds them to the processing list.

Index

• Constants
• type ErrorCode
• type FileWatcher
  • func (w *FileWatcher) Add(name string) error
  • func (w *FileWatcher) Close() error
  • func (w *FileWatcher) ErrorC() <-chan error
  • func (w *FileWatcher) EventC() <-chan fsnotify.Event
• type Marker
• type SpoolError
• type Spooler
  • func NewSpooler(logdir, basename string) Spooler
• type Watcher

Constants

const FileNameBufferLen = 512

Maximum number of files to enqueue

Types

type ErrorCode

type ErrorCode int

Error codes produced by Spooler function

const (
	// Unable to read information about file that has been created
	// in spooler directory
	E_WatcherStat ErrorCode = iota
	// Error passed down from fsnotify
	E_Watcher
	// Unable to list files in the spooler directory
	E_WatcherSpoolerDirectoryList
	// Error passed down from parser
	E_Parse
)

Sequence of error constants produced by Spooler function

type FileWatcher

type FileWatcher fsnotify.Watcher

Wrap fsnotify.Watcher struct in local structure so its possible to expose the Events and Errors channels via functions.

func (*FileWatcher) Add

func (w *FileWatcher) Add(name string) error

lift fsnotify.Watcher.Add

func (*FileWatcher) Close

func (w *FileWatcher) Close() error

lift fsnotify.Watcher.Close

func (*FileWatcher) ErrorC

func (w *FileWatcher) ErrorC() <-chan error

return fsnotify.Watcher.Errors channel

func (*FileWatcher) EventC

func (w *FileWatcher) EventC() <-chan fsnotify.Event

return fsnotify.Watcher.Events channel

type Marker

type Marker struct {
	File   string
	Offset int64
}

File and offset within the file that has been processed already. Assumes events in files which have in lexicographical order less than the indicated file have been processed.

type SpoolError

type SpoolError interface {
	error
	Code() ErrorCode
	NextError() error
}

Errors generated by spooler function, type primarily used for test code

type Spooler

type Spooler interface {
	// Asynchronously run the spooler. Starts a new goroutine.
	//
	// Set marker to the start location (i.e. file and offset
	// within file to start parsing).
	//
	// All errors are reported via ErrorC.  The channels will be
	// closed when finished.
	//
	// The marker channel will return exactly 1 result when the
	// process finishes which can be used for resuming spooling
	// from last processed record.
	Start(marker *Marker) (<-chan *u2.Record, <-chan *Marker)

	// Run the spooler. The read records are returned in the
	// results channel.
	//
	// SyncStart blocks until both the main process loop and the
	// file watcher have been terminated (either by error or call
	// to Stop).
	SyncStart(marker *Marker, results chan<- *u2.Record) (*Marker, error)

	// Stops processing events and release resources.
	//
	// When graceful is true, stops watching for new files and
	// then processes all events available (i.e. it runs in batch
	// mode). When false, it will sharply end the processing
	// mid-file.
	//
	// Returns false when it was not possible to processes the
	// request.
	Stop(graceful bool) bool

	// Used by Start() to return errors
	ErrorC() <-chan error
}

func NewSpooler

func NewSpooler(logdir, basename string) Spooler

Create a new instance of a spooler for a given logdir. Searches for files which start with basename in logidr.

type Watcher

type Watcher interface {
	Add(name string) error
	Close() error
	EventC() <-chan fsnotify.Event
	ErrorC() <-chan error
}

Stubbed interface of fsnotify.Watcher, to support testing