firequeue

README

firequeue

firequeue is to ensure putting items to Amazon Kinesis Data Firehose with an in-memory queue

Synopsis

import (
    "context"

    "github.com/aws/aws-sdk-go/aws/session"
    "github.com/aws/aws-sdk-go/service/firehose"
)

func main() {
    sess := session.Must(session.NewSession())
    fh := firehose.New(sess)
    fq := firequeue.New(fh, "DELIVERY_STREAM_NAME")

    ctx := context.Background()
    go fq.Loop(ctx) // We should start looping before sending items

    err := fq.Enqueue(&firehose.Record{...})
    ...
}

Description

The firequeue utilizes an in-memory queue to ensure input to Amazon Kinesis Data Firehose. When the looping process is cancelled by the context, the firequeue wait for the queue to be empty and then exit the loop.

Installation

% go get github.com/natureglobal/firequeue

Author

Songmu

Documentation

Index

• type Option
  • func BatchInterval(period time.Duration) Option
  • func BatchSize(length int) Option
  • func ErrorHandler(fn func(error, []*firehose.Record)) Option
  • func MaxQueueLength(length int) Option
  • func Parallel(i int) Option
• type Queue
  • func New(fh firehoseiface.FirehoseAPI, DeliveryStreamName string, opts ...Option) *Queue
  • func (q *Queue) Enqueue(r *firehose.Record) error
  • func (q *Queue) Loop(ctx context.Context) error
  • func (q *Queue) Stats() Stats
• type Stats

Types

type Option

type Option func(*Queue)

Option is a type for constructor options

func BatchInterval

func BatchInterval(period time.Duration) Option

BatchInterval is an option to specify batch interval

func BatchSize

func BatchSize(length int) Option

BatchSize is an option to specify batch size

func ErrorHandler

func ErrorHandler(fn func(error, []*firehose.Record)) Option

ErrorHandler is an option to specify the error handler

func MaxQueueLength

func MaxQueueLength(length int) Option

MaxQueueLength is an option to specify max length of in-memory queue

func Parallel

func Parallel(i int) Option

Parallel is an option to specify the number of parallelism

type Queue

type Queue struct {
	DeliveryStreamName *string
	// contains filtered or unexported fields
}

Queue manages a sending list for firehose

func New

func New(fh firehoseiface.FirehoseAPI, DeliveryStreamName string, opts ...Option) *Queue

New return new Queue

func (*Queue) Enqueue

func (q *Queue) Enqueue(r *firehose.Record) error

Enqueue add data to queue

func (*Queue) Loop

func (q *Queue) Loop(ctx context.Context) error

Loop proceeds jobs in queue

func (*Queue) Stats

func (q *Queue) Stats() Stats

Stats return queue stats

type Stats

type Stats struct {
	QueueLength      int
	BatchLength      int64
	Success          int64
	RetryCount       int64
	UnretryableError int64
	QueueFullError   int64
}

Stats contains firequeue statistics.