runner

Documentation

Index

• Constants
• type AWSConfig
  • func NewAWSConfig(region, profile, stream string) *AWSConfig
• type Record
• type Runner
  • func New(rbuf, ebuf int) *Runner
  • func (r *Runner) CheckShards() error
  • func (r *Runner) Configure(a *AWSConfig)
  • func (r *Runner) ConfigureWithMetrics(a *AWSConfig)
  • func (r *Runner) Run()
  • func (r *Runner) Stop()
  • func (r *Runner) WithAsync(async bool) *Runner
  • func (r *Runner) WithDebug(debug bool) *Runner
  • func (r *Runner) WithIntervals(intervalCheckShards, intervalCheckRecords time.Duration) *Runner

Constants

const (
	RecordByteLimit                    = 1024 * 1024
	PutRecordsLengthLimit              = 500
	PutRecordsByteLimit                = 5 * 1024 * 1024
	PutRecordsByteLimitPerMSPerShard   = 1024 // assuming 1kb per ms
	PutRecordsLengthLimitPerMSPerShard = 1
)

Types

type AWSConfig

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

func NewAWSConfig

func NewAWSConfig(region, profile, stream string) *AWSConfig

type Record

type Record struct {
	Data         []byte
	PartitionKey string
}

type Runner

type Runner struct {

	// Message channel to send records to
	RecordChan chan Record

	Metrics *metrics.Metrics
	// contains filtered or unexported fields
}

func New

func New(rbuf, ebuf int) *Runner

New returns a new Runner struct, with provided configurations. stream parameter is for setting the stream name, rbuf and ebuf are channel buffer sizes rbuf is for RecordChan that can help us achieve higher burst traffic. ebuf is for internal batch of PutRecords call.

func (*Runner) CheckShards

func (r *Runner) CheckShards() error

CheckShards for checking the number of open shards available in the given kinesis stream

func (*Runner) Configure

func (r *Runner) Configure(a *AWSConfig)

Configure should be called after New to configure the underlying kinesis client

func (*Runner) ConfigureWithMetrics

func (r *Runner) ConfigureWithMetrics(a *AWSConfig)

ConfigureWithMetrics enables the Runner with metrics and should be called after New to configure the underlying kinesis client

func (*Runner) Run

func (r *Runner) Run()

func (*Runner) Stop

func (r *Runner) Stop()

func (*Runner) WithAsync

func (r *Runner) WithAsync(async bool) *Runner

WithAsync enables the async flag to not wait for PutRecords Response, and send the data asynchronously.

func (*Runner) WithDebug

func (r *Runner) WithDebug(debug bool) *Runner

WithDebug enables the debug flag, writes metrics to logger.

func (*Runner) WithIntervals

func (r *Runner) WithIntervals(intervalCheckShards, intervalCheckRecords time.Duration) *Runner

WithIntervals allows to set intervals, for checking shards and checking records.