kinesis

README

go-kinesis

Batch producer for Kinesis built on top of the official Go AWS SDK.

License

MIT

Documentation

Overview

Package kinesis implements a batch producer built on top of the official AWS SDK.

Package kinesis implements a batch producer built on top of the official AWS SDK.

Index

• Variables
• type Config
• type FirehoseConfig
• type FirehoseProducer
  • func NewFirehose(config FirehoseConfig) *FirehoseProducer
  • func (p *FirehoseProducer) Put(data []byte) error
  • func (p *FirehoseProducer) Start()
  • func (p *FirehoseProducer) Stop()
• type KinesisProducer
  • func New(config Config) *KinesisProducer
  • func (p *KinesisProducer) Put(data []byte, partitionKey string) error
  • func (p *KinesisProducer) Start()
  • func (p *KinesisProducer) Stop()
• type Producer

Variables

var (
	ErrRecordSizeExceeded = errors.New("kinesis: record size exceeded")
)

Errors.

Types

type Config

type Config struct {
	// StreamName is the Kinesis stream.
	StreamName string

	// FlushInterval is a regular interval for flushing the buffer. Defaults to 1s.
	FlushInterval time.Duration

	// BufferSize determines the batch request size. Must not exceed 500. Defaults to 500.
	BufferSize int

	// BacklogSize determines the channel capacity before Put() will begin blocking. Defaults to 500.
	BacklogSize int

	// Backoff determines the backoff strategy for record failures.
	Backoff backoff.Backoff

	// Logger is the logger used. Defaults to log.Log.
	Logger log.Interface

	// Client is the Kinesis API implementation.
	Client kinesisiface.KinesisAPI
}

type FirehoseConfig

type FirehoseConfig struct {
	// StreamName is the Firehose stream.
	StreamName string

	// FlushInterval is a regular interval for flushing the buffer. Defaults to 1s.
	FlushInterval time.Duration

	// BufferSize determines the batch request size. Must not exceed 500. Defaults to 500.
	BufferSize int

	// BacklogSize determines the channel capacity before Put() will begin blocking. Defaults to 500.
	BacklogSize int

	// Backoff determines the backoff strategy for record failures.
	Backoff backoff.Backoff

	// Client is the Firehose API implementation.
	Client firehoseiface.FirehoseAPI
}

type FirehoseProducer

type FirehoseProducer struct {
	FirehoseConfig
	// contains filtered or unexported fields
}

FirehoseProducer batches records.

func NewFirehose

func NewFirehose(config FirehoseConfig) *FirehoseProducer

NewFirehose producer with the given config.

func (*FirehoseProducer) Put

func (p *FirehoseProducer) Put(data []byte) error

Put record `data`. This method is thread-safe.

func (*FirehoseProducer) Start

func (p *FirehoseProducer) Start()

Start the producer.

func (*FirehoseProducer) Stop

func (p *FirehoseProducer) Stop()

Stop the producer. Flushes any in-flight data.

type KinesisProducer

type KinesisProducer struct {
	Config
	// contains filtered or unexported fields
}

Producer batches records.

func New

func New(config Config) *KinesisProducer

New producer with the given config.

func (*KinesisProducer) Put

func (p *KinesisProducer) Put(data []byte, partitionKey string) error

Put record `data` using `partitionKey`. This method is thread-safe.

func (*KinesisProducer) Start

func (p *KinesisProducer) Start()

Start the producer.

func (*KinesisProducer) Stop

func (p *KinesisProducer) Stop()

Stop the producer. Flushes any in-flight data.

type Producer

type Producer interface {
	Put(data []byte) error
	Start()
	Stop()
}