kinesis

Documentation

Index

• Variables
• func BuildStore(cfg StoreConfig) (consumer.Store, error)
• type LoggerAdapter
  • func (l *LoggerAdapter) Log(args ...interface{})
• type SessionConfig
• type StoreConfig
• type Subscriber
  • func NewSubscriber(ctx context.Context, cfg SessionConfig) (*Subscriber, error)
  • func (s *Subscriber) Subscribe(ctx context.Context, ...) error

Variables

var (
	ErrEmptyStreamName = errors.New("kinesis stream name can't be empty")
)

Functions

func BuildStore

func BuildStore(cfg StoreConfig) (consumer.Store, error)

Types

type LoggerAdapter

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

func (*LoggerAdapter) Log

func (l *LoggerAdapter) Log(args ...interface{})

type SessionConfig

type SessionConfig struct {
	Endpoint   string
	Region     string
	StreamName string `split_words:"true" default:"watchops"`
	Store      StoreConfig
	Timeout    time.Duration `default:"5s"`
	MaxRetries int           `split_words:"true" default:"3"`
}

type StoreConfig

type StoreConfig struct {
	Driver  string `default:"memory"`
	AppName string `split_words:"true" default:"watchops_consumer"`
	Redis   struct {
		Address  string
		Password string
		Username string
		DB       int
	}
	Postgres struct {
		TableName string `split_words:"true"`
		DSN       string
	}
	MySQL struct {
		TableName string `split_words:"true"`
		DSN       string
	}
}

type Subscriber

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

Subscriber is the kinesis subscriber.

func NewSubscriber

func NewSubscriber(ctx context.Context, cfg SessionConfig) (*Subscriber, error)

NewSubscriber creates a new instance of Subscriber.

func (*Subscriber) Subscribe

func (s *Subscriber) Subscribe(ctx context.Context, fn func(ctx context.Context, payload []byte, headers map[string][]string) error) error

Subscribe subscribes to the kinesis stream. nolint: exhaustivestruct