reader

Documentation

Index

• func ReadOnceAsync(stream io.Reader, buffer []byte) (int, error)
• func ReadUntilFullAsync(ctx context.Context, stream io.Reader, buffer []byte) (int, error)
• type Reader
  • func NewReader(client *http.Client, logger *logr.Logger, options *ReaderOptions, ...) *Reader
  • func (r *Reader) Close() error
  • func (r *Reader) GetStream(ctx context.Context) (io.ReadCloser, error)
  • func (r *Reader) ProcessBatch() error
  • func (r *Reader) ProcessLine(lineData []byte, row int64, bytesConsumed int64) error
  • func (r *Reader) ReadStream(ctx context.Context, stream io.Reader) error
• type ReaderOptions
  • func DefaultReaderOptions() *ReaderOptions
• type StreamSource

Functions

func ReadOnceAsync

func ReadOnceAsync(stream io.Reader, buffer []byte) (int, error)

ReadOnceAsync reads from the stream once

func ReadUntilFullAsync

func ReadUntilFullAsync(ctx context.Context, stream io.Reader, buffer []byte) (int, error)

ReadUntilFullAsync reads from the stream until the buffer is full or the stream ends

Types

type Reader

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

Reader reads and validates lines from a stream

func NewReader

func NewReader(client *http.Client, logger *logr.Logger, options *ReaderOptions, baseURL string) *Reader

NewReader creates a new reader

func (*Reader) Close

func (r *Reader) Close() error

Close cleans up resources

func (*Reader) GetStream

func (r *Reader) GetStream(ctx context.Context) (io.ReadCloser, error)

GetStream returns a stream based on the configured source

func (*Reader) ProcessBatch

func (r *Reader) ProcessBatch() error

ProcessBatch simulates batch processing to help trigger the issue

func (*Reader) ProcessLine

func (r *Reader) ProcessLine(lineData []byte, row int64, bytesConsumed int64) error

ProcessLine validates a line against the expected value

func (*Reader) ReadStream

func (r *Reader) ReadStream(ctx context.Context, stream io.Reader) error

ReadStream reads and processes the stream

type ReaderOptions

type ReaderOptions struct {
	BatchSize    int
	ChunkSize    int
	Delay        time.Duration
	FillBuffer   bool
	StreamSource StreamSource
}

ReaderOptions configures the reader

func DefaultReaderOptions

func DefaultReaderOptions() *ReaderOptions

DefaultReaderOptions returns default options

type StreamSource

type StreamSource int

StreamSource defines the source of the stream

const (
	// StreamSourceHTTP retrieves the stream from an HTTP server
	StreamSourceHTTP StreamSource = iota
	// StreamSourceLocal creates a local stream
	StreamSourceLocal
)