input

Documentation

Index

• type InputOption
  • func WithModifyScan(modifyScan ModifyScan) InputOption
• type ModifyScan
• type S3Input
  • func NewS3Input(cfg S3InputConfig, opts ...InputOption) *S3Input
  • func (i *S3Input) Read(container *gobulk.Container) (map[string][]byte, error)
  • func (i *S3Input) Scan(ctx context.Context, marker *gobulk.Container, ...)
  • func (i *S3Input) Setup() error
• type S3InputConfig

Types

type InputOption

type InputOption func(i *S3Input)

InputOption is the base form of an option parameter for S3Input.

func WithModifyScan

func WithModifyScan(modifyScan ModifyScan) InputOption

WithModifyScan enhanced the S3Input with the passed modifyScan.

type ModifyScan

type ModifyScan func(output *s3.ListObjectsOutput) *s3.ListObjectsOutput

ModifyScan is the form of the func to be used as an optional parameter for the S3Input that is called between getting list of objects from S3 and sending them to the contCh. Could be used to filter result objects or do other needed stuff.

type S3Input

type S3Input struct {
	gobulk.BaseStorage
	Cfg S3InputConfig
	// contains filtered or unexported fields
}

S3Input represents an input that reads files from an AWS S3 Bucket.

func NewS3Input

func NewS3Input(cfg S3InputConfig, opts ...InputOption) *S3Input

NewS3Input returns a new instance of the S3Input.

func (*S3Input) Read

func (i *S3Input) Read(container *gobulk.Container) (map[string][]byte, error)

Read reads the raw data of a container.

func (*S3Input) Scan

func (i *S3Input) Scan(ctx context.Context, marker *gobulk.Container, contCh chan<- []*gobulk.Container, doneCh chan<- struct{}, errCh chan<- error)

Scan scans the S3 bucket for new containers and sends them to the channel. It starts the scan with the container defined as marker and stops either on an S3-interaction error, or when the context is cancelled, or when all the input containers are read.

func (*S3Input) Setup

func (i *S3Input) Setup() error

Setup contains the storage preparations like connection etc. Is called only once at the very beginning of the work with the storage. As for the S3Input, it checks whether the config for the input is proper by connecting and performing a simple S3 API call.

type S3InputConfig

type S3InputConfig struct {
	AwsCfg        *aws.Config
	Bucket        string `validate:"required"`
	Prefix        string
	ScanIntervals string
	ScanThrottle  int
	Delimiter     string
	EncodingType  string
	MaxKeys       int64 `validate:"lte=1000"` // AWS API allows to receive not more than 1000 items in a call
}

S3InputConfig represents the S3Input configurable fields model.