resources

Documentation

Index

• func CompressedCsvReader(path string, readerHandler func(reader *csv.Reader) error, opts ...CsvOption) error
• func CompressedCsvWriter(path string, writerHandler func(writer *csv.Writer) error, opts ...CsvOption) error
• func GetTempFilePath() (string, error)
• func NewCsvReader(r io.Reader) *csv.Reader
• func NewCsvWriter(w io.Writer) *csv.Writer
• func OpenTempFile() (io.WriteCloser, string, error)
• func Pipeline(src chan interface{}, pipes ...Pipe) chan interface{}
• func ReadAllLines(csvReader *csv.Reader, batchSize int, ...) error
• func SplitEntityName(fullName, entityType string) (string, int32, error)
• type Batcher
  • func NewBatcher(batchSize int) *Batcher
  • func (b *Batcher) Map(src chan interface{}) chan interface{}
• type CsvOption
  • func WithCloseErrorHandler(handler func(err error)) CsvOption
  • func WithSeparator(sep rune) CsvOption
• type JsonUnmarshaler
  • func NewJsonUnmarshaler(makePtr func() interface{}) *JsonUnmarshaler
  • func (u *JsonUnmarshaler) Goroutines(count int) *JsonUnmarshaler
  • func (u *JsonUnmarshaler) Map(src chan interface{}) chan interface{}
  • func (u *JsonUnmarshaler) OnError(f func(err error)) *JsonUnmarshaler
  • func (u *JsonUnmarshaler) Unmarshal(unmarshal func(bytes []byte, ptr interface{}) error) *JsonUnmarshaler
• type LineScanner
  • func NewLineScanner(filePattern string, maxBuffSize int) *LineScanner
  • func (s *LineScanner) Lines() chan interface{}
  • func (s *LineScanner) OnError(f func(err error)) *LineScanner
  • func (s *LineScanner) Reader(makeReader func(f *os.File) (io.Reader, error)) *LineScanner
  • func (s *LineScanner) Run() []Result
  • func (s *LineScanner) Skip(lineCount int) *LineScanner
• type Pipe
• type Result

Functions

func CompressedCsvReader

func CompressedCsvReader(path string, readerHandler func(reader *csv.Reader) error, opts ...CsvOption) error

func CompressedCsvWriter

func CompressedCsvWriter(path string, writerHandler func(writer *csv.Writer) error, opts ...CsvOption) error

func GetTempFilePath

func GetTempFilePath() (string, error)

func NewCsvReader

func NewCsvReader(r io.Reader) *csv.Reader

func NewCsvWriter

func NewCsvWriter(w io.Writer) *csv.Writer

func OpenTempFile

func OpenTempFile() (io.WriteCloser, string, error)

func Pipeline

func Pipeline(src chan interface{}, pipes ...Pipe) chan interface{}

func ReadAllLines

func ReadAllLines(
	csvReader *csv.Reader,
	batchSize int,
	onBatch func(batch [][]string, lastReadCount int, totalRead int) error,
	onError func(err error) bool,
) error

func SplitEntityName

func SplitEntityName(fullName, entityType string) (string, int32, error)

Types

type Batcher

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

func NewBatcher

func NewBatcher(batchSize int) *Batcher

func (*Batcher) Map

func (b *Batcher) Map(src chan interface{}) chan interface{}

type CsvOption

type CsvOption func(opts *csvOpts)

func WithCloseErrorHandler

func WithCloseErrorHandler(handler func(err error)) CsvOption

func WithSeparator

func WithSeparator(sep rune) CsvOption

type JsonUnmarshaler

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

func NewJsonUnmarshaler

func NewJsonUnmarshaler(makePtr func() interface{}) *JsonUnmarshaler

func (*JsonUnmarshaler) Goroutines

func (u *JsonUnmarshaler) Goroutines(count int) *JsonUnmarshaler

func (*JsonUnmarshaler) Map

func (u *JsonUnmarshaler) Map(src chan interface{}) chan interface{}

func (*JsonUnmarshaler) OnError

func (u *JsonUnmarshaler) OnError(f func(err error)) *JsonUnmarshaler

func (*JsonUnmarshaler) Unmarshal

func (u *JsonUnmarshaler) Unmarshal(unmarshal func(bytes []byte, ptr interface{}) error) *JsonUnmarshaler

type LineScanner

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

func NewLineScanner

func NewLineScanner(filePattern string, maxBuffSize int) *LineScanner

func (*LineScanner) Lines

func (s *LineScanner) Lines() chan interface{}

func (*LineScanner) OnError

func (s *LineScanner) OnError(f func(err error)) *LineScanner

func (*LineScanner) Reader

func (s *LineScanner) Reader(makeReader func(f *os.File) (io.Reader, error)) *LineScanner

func (*LineScanner) Run

func (s *LineScanner) Run() []Result

iterate from 0 to n, replace index in 'filePattern' and tries to read file it closes channel 'ch' and return if file is not exists

func (*LineScanner) Skip

func (s *LineScanner) Skip(lineCount int) *LineScanner

type Pipe

type Pipe interface {
	Map(src chan interface{}) chan interface{}
}

type Result

type Result struct {
	Errors      map[int]string `json:",omitempty"`
	Lines       int
	ErrorsCount int
	FileName    string
}