multi_container

Documentation

Index

• Variables
• type ArchiveManager
  • func NewArchiveManager(rootDir, prefix string, compQueueBufSize, compPoolSize int) (am *ArchiveManager, err error)
  • func (am *ArchiveManager) CleanCompress() error
  • func (am *ArchiveManager) Close() error
  • func (am *ArchiveManager) Files(ctx context.Context, start, end time.Time) chan string
  • func (am *ArchiveManager) QueueCompression(filename string)
• type Container
  • func NewContainer[P generics.Ptr[RowType], RowType any](rootDir, prefix string, options ...Option) (c *Container[P, RowType], err error)
  • func (c *Container[P, RowType]) AcquireContainer() *sbt.Container[P, RowType]
  • func (c *Container[P, RowType]) AsIterable() generics.Iterable[containers.Tuple[*MultiContainerIteratorKey, P]]
  • func (c *Container[P, RowType]) Close() (err error)
  • func (c *Container[P, RowType]) Iter() *generics.Iterator[containers.Tuple[*MultiContainerIteratorKey, P]]
  • func (c *Container[P, RowType]) IterHandler(iter *generics.Iterator[containers.Tuple[*MultiContainerIteratorKey, P]])
  • func (c *Container[P, RowType]) IterRange(start, end time.Time) *generics.Iterator[containers.Tuple[*MultiContainerIteratorKey, P]]
  • func (c *Container[P, RowType]) ReleaseContainer()
• type MultiContainerFilenameParts
  • func NewMultiContainerFilenamePartsFromNow(prefix string) MultiContainerFilenameParts
  • func SplitMultiContainerFilename(filename string, tz *time.Location) (parts MultiContainerFilenameParts, err error)
  • func (p MultiContainerFilenameParts) Date() time.Time
  • func (p MultiContainerFilenameParts) Prefix() string
  • func (p MultiContainerFilenameParts) String() string
  • func (p MultiContainerFilenameParts) Unix() int64
• type MultiContainerIteratorKey
• type Option
  • func WithCompressionPoolSize(size int) Option
  • func WithCompressionScheduler(delaySec int) Option
  • func WithLog(l *log.Logger) Option
  • func WithOpenRead() Option
• type Options
  • func (o *Options) LogPrintf(format string, v ...interface{})

Variables

var ErrNoFileFound = errors.New("no file found")

Types

type ArchiveManager

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

ArchiveManager manages the archives of the multi container

func NewArchiveManager

func NewArchiveManager(rootDir, prefix string, compQueueBufSize, compPoolSize int) (am *ArchiveManager, err error)

NewArchiveManager creates a new archive manager

func (*ArchiveManager) CleanCompress

func (am *ArchiveManager) CleanCompress() error

CleanCompress does this to root dir:

• Ignore files with prefix and .sbt extension if there's a compressed version
• Compress any file with prefix and .sbt extension

func (*ArchiveManager) Close

func (am *ArchiveManager) Close() error

Close closes the archive

func (*ArchiveManager) Files

func (am *ArchiveManager) Files(ctx context.Context, start, end time.Time) chan string

Files returns the decompressed filename channel. This function panics on error, make sure to recover.

func (*ArchiveManager) QueueCompression

func (am *ArchiveManager) QueueCompression(filename string)

QueueCompression queues files for compression. filename should be the full path to the file (starting at rootDir)

type Container

type Container[P generics.Ptr[RowType], RowType any] struct {
	// contains filtered or unexported fields
}

Container is a AcquireContainer wrapper allowing data insertion in multiple serial files with archive control to save space.

It use a predefined filename format for identifying files. The format is <prefix>_<date 2006-01-02-15-04>_<unix time>.sbt

func NewContainer

func NewContainer[P generics.Ptr[RowType], RowType any](
	rootDir, prefix string, options ...Option,
) (c *Container[P, RowType], err error)

NewContainer creates a new Container and loads the last container file based on the mode

func (*Container[P, RowType]) AcquireContainer

func (c *Container[P, RowType]) AcquireContainer() *sbt.Container[P, RowType]

AcquireContainer returns the current container in a thread safe way. MAKE SURE to call ReleaseContainer when done, otherwise the container will be locked forever.

func (*Container[P, RowType]) AsIterable

func (c *Container[P, RowType]) AsIterable() generics.Iterable[containers.Tuple[*MultiContainerIteratorKey, P]]

func (*Container[P, RowType]) Close

func (c *Container[P, RowType]) Close() (err error)

Close closes the current container and the archive scheduler

func (*Container[P, RowType]) Iter

func (c *Container[P, RowType]) Iter() *generics.Iterator[containers.Tuple[*MultiContainerIteratorKey, P]]

Iter will begin iteration from the very first found file

func (*Container[P, RowType]) IterHandler

func (c *Container[P, RowType]) IterHandler(
	iter *generics.Iterator[containers.Tuple[*MultiContainerIteratorKey, P]],
)

func (*Container[P, RowType]) IterRange

func (c *Container[P, RowType]) IterRange(
	start, end time.Time,
) *generics.Iterator[containers.Tuple[*MultiContainerIteratorKey, P]]

IterRange will begin iteration from the first found file and will stop at the last found file

func (*Container[P, RowType]) ReleaseContainer

func (c *Container[P, RowType]) ReleaseContainer()

ReleaseContainer releases the current container.

type MultiContainerFilenameParts

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

func NewMultiContainerFilenamePartsFromNow

func NewMultiContainerFilenamePartsFromNow(prefix string) MultiContainerFilenameParts

NewMultiContainerFilenamePartsFromNow creates a new MultiContainerFilenameParts from now

func SplitMultiContainerFilename

func SplitMultiContainerFilename(filename string, tz *time.Location) (parts MultiContainerFilenameParts, err error)

SplitMultiContainerFilename splits a filename into parts

func (MultiContainerFilenameParts) Date

func (p MultiContainerFilenameParts) Date() time.Time

Date

func (MultiContainerFilenameParts) Prefix

func (p MultiContainerFilenameParts) Prefix() string

Prefix

func (MultiContainerFilenameParts) String

func (p MultiContainerFilenameParts) String() string

String returns the filename

func (MultiContainerFilenameParts) Unix

func (p MultiContainerFilenameParts) Unix() int64

Unix

type MultiContainerIteratorKey

type MultiContainerIteratorKey struct {
	Filename string
	RowId    int64
}

type Option

type Option func(*Options)

func WithCompressionPoolSize

func WithCompressionPoolSize(size int) Option

WithCompressionPoolSize sets the size of the compression worker pool. Defaults to NumCPU / 4.

func WithCompressionScheduler

func WithCompressionScheduler(delaySec int) Option

func WithLog

func WithLog(l *log.Logger) Option

func WithOpenRead

func WithOpenRead() Option

WithOpenRead sets the container to open the file for reading. Defaults to false.

type Options

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

func (*Options) LogPrintf

func (o *Options) LogPrintf(format string, v ...interface{})

LogPrintf