internal

Documentation

Index

• Variables
• func CreateKafkaTopic(topic string, addr net.Addr) (err error)
• func KafkaWriteWithBackoff(ctx context.Context, writer *segmentio.Writer, ...) error
• func ReadFileOrURL(name string, s3client s3iface.S3API) ([]byte, error)
• func WriteFileOrURL(name string, contents []byte, s3client s3iface.S3API) error
• type KafkaReader
  • func BlendKafka(in map[string]KafkaReader) KafkaReader
• type KafkaTestReader
  • func (r *KafkaTestReader) Close() error
  • func (r *KafkaTestReader) CommitMessages(ctx context.Context, msgs ...segmentio.Message) error
  • func (r *KafkaTestReader) FetchMessage(ctx context.Context) (segmentio.Message, error)
• type RetryReader
  • func (r RetryReader) FetchMessage(ctx context.Context) (segmentio.Message, error)

Variables

var FileOrURLNotFound = errors.New("file or url does not exist")

Functions

func CreateKafkaTopic

func CreateKafkaTopic(topic string, addr net.Addr) (err error)

func KafkaWriteWithBackoff

func KafkaWriteWithBackoff(ctx context.Context, writer *segmentio.Writer, log func(string, ...interface{}), interval, maxInterval time.Duration, messages ...segmentio.Message) error

func ReadFileOrURL

func ReadFileOrURL(name string, s3client s3iface.S3API) ([]byte, error)

ReadFileOrURL reads a path from the filesystem or an s3 URL. The s3client parameter is required if reading an s3 URL. ReadFileOrURL returns FileOrURLNotFound when the local filesystem path or remote s3 location is not found.

func WriteFileOrURL

func WriteFileOrURL(name string, contents []byte, s3client s3iface.S3API) error

Types

type KafkaReader

type KafkaReader interface {
	FetchMessage(ctx context.Context) (segmentio.Message, error)
	CommitMessages(ctx context.Context, msgs ...segmentio.Message) error
	io.Closer
}

func BlendKafka

func BlendKafka(in map[string]KafkaReader) KafkaReader

type KafkaTestReader

type KafkaTestReader struct {
	Queue               []segmentio.Message
	FetchOff, CommitOff int
	Closed              bool
}

KafkaTestReader is a testing implementation of the KafkaReader type.

func (*KafkaTestReader) Close

func (r *KafkaTestReader) Close() error

func (*KafkaTestReader) CommitMessages

func (r *KafkaTestReader) CommitMessages(ctx context.Context, msgs ...segmentio.Message) error

func (*KafkaTestReader) FetchMessage

func (r *KafkaTestReader) FetchMessage(ctx context.Context) (segmentio.Message, error)

type RetryReader

type RetryReader struct {
	*segmentio.Reader
}

RetryReader wraps a kafka reader with fetch retry logic.

func (RetryReader) FetchMessage

func (r RetryReader) FetchMessage(ctx context.Context) (segmentio.Message, error)