consumer

Documentation

Index

• type Config
• type KinesisPipe
  • func NewKinesisPipe(kinesis kinesisiface.KinesisAPI, dynamodb dynamodbiface.DynamoDBAPI, ...) (*KinesisPipe, error)
  • func (c *KinesisPipe) Close()
  • func (c *KinesisPipe) ReadChannel() <-chan *Result
• type Result
• type ResultPipe
• type StandardInputPipe
  • func NewStandardInputPipe() *StandardInputPipe
  • func (c *StandardInputPipe) Close()
  • func (c *StandardInputPipe) ReadChannel() <-chan *Result

Types

type Config

type Config struct {
	// ApplicationName is the name that kinsumer uses to communicate with other clients
	// consuming the same kinesis stream
	ApplicationName string

	// StreamName is the name of the stream that is being consumed from
	StreamName string

	// (Optional) Time for Kinsumer to sleep if there are no new records
	ThrottleDelay string

	// (Optional) Delay before the checkpoint for each shard is committed to the database
	CommitFrequency string

	// (Optional) How frequently the list of shards are checked
	ShardCheckFrequency string

	// (Optional) Size of the internal buffer for kinesis events
	BufferSize int
}

Config is used to set configuration variables for the Consumer

type KinesisPipe

type KinesisPipe struct {
	// C is used to read records off the kinsumer queue
	C <-chan *Result

	sync.WaitGroup
	// contains filtered or unexported fields
}

KinesisPipe is a ResultPipe that consumes globs of events from Kinesis.

func NewKinesisPipe

func NewKinesisPipe(kinesis kinesisiface.KinesisAPI, dynamodb dynamodbiface.DynamoDBAPI, stats statsd.StatSender, config Config) (*KinesisPipe, error)

NewKinesisPipe returns a newly created KinesisPipe.

func (*KinesisPipe) Close

func (c *KinesisPipe) Close()

Close closes down Kinesis consumption.

func (*KinesisPipe) ReadChannel

func (c *KinesisPipe) ReadChannel() <-chan *Result

ReadChannel provides Results which are base-64 encoded, compressed lists of JSON records.

type Result

type Result struct {
	Data  []byte
	Error error
}

Result is the next data/error to be consumed from the kinsumer or standard input.

type ResultPipe

type ResultPipe interface {
	// ReadChannel provides a channel from which the Results are read.
	ReadChannel() <-chan *Result

	// Close cleans up any resources associated with the pipe.
	Close()
}

ResultPipe consumes input from somewhere and provides Results through its ReadChannel.

type StandardInputPipe

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

StandardInputPipe is a ResultPipe that consumes plaintext events from standard input.

func NewStandardInputPipe

func NewStandardInputPipe() *StandardInputPipe

NewStandardInputPipe sets up a StandardInputPipe.

func (*StandardInputPipe) Close

func (c *StandardInputPipe) Close()

Close does nothing, as standard input closes automatically on EOF.

func (*StandardInputPipe) ReadChannel

func (c *StandardInputPipe) ReadChannel() <-chan *Result

ReadChannel provides results which are single, uncompressed, decoded events.