concurrentlineprocessor

Jan 5, 2026

Changes in this version

+ func DrainData(r io.Reader)

type Chunk

+ func (chunk *Chunk) Grow(n int)

type Option

+ func WithReaders(readers ...io.ReadCloser) Option

Dec 15, 2025

Changes in this version

+ const BaseBinary

+ const BaseSI

+ func Lines(l []byte, rawLine bool) iter.Seq[[]byte]

+ func Ptr[T any](v T) *T

type Chunk

+ func (chunk *Chunk) Write(src []byte) (int, error)

+ func (chunk *Chunk) WriteByte(b byte) error

+ type ChunkDetails struct

+ ChunkID int

+ ReaderID int

+ type DataProcessor func(b []byte, info *ChunkDetails, out io.Writer) error

type Option

+ func WithCustomChunkProcessor(c DataProcessor) Option

Nov 23, 2025

Changes in this version

+ var KB = 1024

+ func EnsureNewLineAtEnd(chunk *Chunk)

+ func FormatDuration(d time.Duration) string

+ type LineDetails struct

+ ChunkID int

+ ReaderID int

type Metrics

+ BytesWritten int64

type Option

+ func WithContext(ctx context.Context) Option

Oct 20, 2025

Changes in this version

+ func Filter[T any](arr []T, keep func(T) bool) []T

type Option

+ func WithMultiReaders(readers ...io.ReadCloser) Option

Jul 21, 2025

Changes in this version

+ func PrintSummaryPeriodically(p *concurrentLineProcessor, now time.Time)

Jul 17, 2025

Changes in this version

type Metrics

+ BytesTransformed int64

+ RowsWritten int64

Jul 11, 2025

Changes in this version

+ func AppendNewLine(b *[]byte)

+ func FormatBytes(size int) string

type ConcurrentLineProcessor

+ func (p *ConcurrentLineProcessor) Summary() string

type Option

+ func WithChannelSize(size int) Option

Jul 4, 2025

Changes in this version

+ func ExitOnError(err error)

+ func IfNull[T any](org *T, def T) T

Jul 4, 2025

Changes in this version

+ var DefaultChunkSize = 1024 * 30

+ var DefaultWorkers = runtime.NumCPU()

+ var Files = []string

+ func ErrWithDebugStack(err error) error

+ func ExistOnError(err error)

+ func IFNull[T any](org *T, def T) T

+ func PrintAsJsonString(d any)

+ func WithNewLine(data []byte) []byte

+ func WithOpts(p *ConcurrentLineProcessor, opts ...Option)

+ type Chunk struct

+ type ConcurrentLineProcessor struct

+ func NewConcurrentLineProcessor(r io.Reader, opts ...Option) *ConcurrentLineProcessor

+ func (p *ConcurrentLineProcessor) Metrics() Metrics

+ func (p *ConcurrentLineProcessor) Read(b []byte) (int, error)

+ func (p *ConcurrentLineProcessor) RowsRead() int

+ type LineProcessor func([]byte) ([]byte, error)

+ type Metrics struct

+ BytesRead int64

+ RowsRead int64

+ TimeTook string

+ TransformedBytes int64

+ type Option func(*ConcurrentLineProcessor)

+ func WithChunkSize(size int) Option

+ func WithCustomLineProcessor(c LineProcessor) Option

+ func WithRowsReadLimit(limit int) Option

+ func WithWorkers(n int) Option