kcl

README

kinesis_client_library

AWS kinesis client library implemented in go

Documentation

Overview

Package kcl provides a kinesis client library (KCL) implementation for Amazon Web Services. The primary goals of this package are to provide a performant Go KCL while keeping the library idomatic (and readable) and easy to use.

The AWS SDK is configured based on the AWS SDK's default configuration: http://godoc.org/github.com/awslabs/aws-sdk-go/aws#pkg-constants. This means that AWS credentials are loaded using the credentials.NewChainCredentials method. The quickest way to get started using this package is to set the following environmental variables: Access Key ID: AWS_ACCESS_KEY_ID or AWS_ACCESS_KEY, Secret Access Key: AWS_SECRET_ACCESS_KEY or AWS_SECRET_KEY, and region: AWS_REGION.

Please report/log issues or feature requests here: https://github.com/suicidejack/kinesis_client_library/issues.

Index

• type Config
• type KinesisRecord
• type StreamConsumer
  • func NewStreamConsumer(opts *Config) (consumer *StreamConsumer, err error)
  • func (c *StreamConsumer) Consume() chan *KinesisRecord
  • func (c *StreamConsumer) Shutdown() (err error)
  • func (c *StreamConsumer) Start() (err error)
  • func (c *StreamConsumer) ValidateStream() error

Types

type Config

type Config struct {
	// [REQUIRED] the application's name - it will be the name of the dynamodb
	// table that tracks this applications name
	ConsumerGroup string `json:"ConsumerGroup"`

	// [REQUIRED] the name of stream to consume from
	StreamName string `json:"streamName"`

	// run the AWS client in debug mode
	AWSDebugMode   bool                   `json:"awsDebugMode"`
	NumRecords     int64                  `json:"numRecords"`
	BufferSize     int                    `json:"bufferSize"`
	QueryFrequency *goprimitives.Duration `json:"queryFrequency"`
	ReadCapacity   int64                  `json:"readCapacity"`
	WriteCapacity  int64                  `json:"writeCapacity"`
	// WorkerID a unique ID for this consumer
	WorkerID string `json:"workerID"`
	// ConsumerExpirationSeconds the amount of time that a consumer has to checkpoint before
	// another consumer takes over consuming from that shard
	ConsumerExpirationSeconds int64 `json:"consumerExpirationSeconds"`
	// IteratorType valid options are "LATEST" or "TRIM_HORIZON" defaults to "LATEST"
	IteratorType string `json:"iteratorType"`
}

Config for a kinesis producer (high level, stream, etc.)

type KinesisRecord

type KinesisRecord struct {
	Data       []byte
	Checkpoint func()
}

KinesisRecord TODO

type StreamConsumer

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

StreamConsumer TODO

func NewStreamConsumer

func NewStreamConsumer(opts *Config) (consumer *StreamConsumer, err error)

NewStreamConsumer TODO

func (*StreamConsumer) Consume

func (c *StreamConsumer) Consume() chan *KinesisRecord

Consume TODO

func (*StreamConsumer) Shutdown

func (c *StreamConsumer) Shutdown() (err error)

Shutdown TODO

func (*StreamConsumer) Start

func (c *StreamConsumer) Start() (err error)

Start TODO

func (*StreamConsumer) ValidateStream

func (c *StreamConsumer) ValidateStream() error

ValidateStream checks if the stream exists (or even if you have a valid AWS connection) and returns and error if the stream does not exist