kinesisdatacounter

package module

Go to main page

Versions in this module

v0

v0.3.1

Nov 4, 2022

v0.3.0

Nov 4, 2022

v0.2.0

Dec 1, 2021

Changes in this version

type CounterConfig

+ TargetExpr string

v0.1.4

Nov 24, 2021

v0.1.3

Nov 24, 2021

v0.1.2

Nov 24, 2021

Changes in this version

type KinesisTimeWindow

+ func (w *KinesisTimeWindow) String() string

v0.1.1

Nov 24, 2021

v0.1.0

Nov 21, 2021

Changes in this version

+ func CounterTypeValuesString() string

+ type ARN struct

+ func (arn *ARN) IsAmbiguous() bool

+ func (arn *ARN) IsKinesisDataStream() bool

+ func (arn *ARN) MarshalText() ([]byte, error)

+ func (arn *ARN) Match(other string) bool

+ func (arn *ARN) Set(text string) error

+ func (arn *ARN) StreamName() string

+ func (arn *ARN) UnmarshalText(text []byte) error

+ type App struct

+ func New(cfg *Config) (*App, error)

+ func NewWithClient(cfg *Config, kinesisClient KinesisClient, firehoseClient FirehoseClient) *App

+ func (app *App) Handler(ctx context.Context, event *KinesisTimeWindowEvent) (*TimeWindowEventResponse, error)

+ func (app *App) Run(ctx context.Context, streamName string, tumblingWindow time.Duration) error

+ func (app *App) SetIgnorePutRecord(f bool)

+ func (app *App) SetOutput(w io.Writer)

+ func (app *App) SetVersion(version string)

+ type BatchItemFailure struct

+ ItemIdentifier string

+ type Config struct

+ Counters []*CounterConfig

+ RequiredVersion string

+ func NewDefaultConfig() *Config

+ func (c *Config) ValidateVersion(version string) error

+ func (cfg *Config) Load(path string) error

+ func (cfg *Config) Restrict() error

+ type CounterConfig struct

+ AggregateStreamArn *ARN

+ CounterType CounterType

+ ID string

+ InputStreamARN *ARN

+ JQExpr string

+ OutputStreamARN *ARN

+ SipHashKeyHex string

+ TargetColumn string

+ func NewDefaultCounterConfig() *CounterConfig

+ func (cfg *CounterConfig) Restrict() error

+ func (cfg *CounterConfig) SetFlags(f *flag.FlagSet) error

+ type CounterState struct

+ Base64HLLPP string

+ CounterType CounterType

+ RowCount int64

+ type CounterStates map[string]map[string]*CounterState

+ func (s CounterStates) MergeInto(other CounterStates)

+ type CounterType int

+ const ApproxCountDistinct

+ const Count

+ func CounterTypeString(s string) (CounterType, error)

+ func CounterTypeValues() []CounterType

+ func (i *CounterType) UnmarshalJSON(data []byte) error

+ func (i *CounterType) UnmarshalYAML(unmarshal func(interface{}) error) error

+ func (i CounterType) IsACounterType() bool

+ func (i CounterType) MarshalJSON() ([]byte, error)

+ func (i CounterType) MarshalYAML() (interface{}, error)

+ func (i CounterType) String() string

+ func (t *CounterType) Set(str string) error

+ type FirehoseClient interface

+ PutRecord func(ctx context.Context, params *firehose.PutRecordInput, ...) (*firehose.PutRecordOutput, error)

+ type IntermediateRecord struct

+ CounterID string

+ CounterType CounterType

+ CounterVersion string

+ EventSourceARN string

+ ShardID string

+ State *CounterState

+ Window *KinesisTimeWindow

+ type KinesisClient interface

+ DescribeStream func(ctx context.Context, params *kinesis.DescribeStreamInput, ...) (*kinesis.DescribeStreamOutput, error)

+ GetRecords func(ctx context.Context, params *kinesis.GetRecordsInput, ...) (*kinesis.GetRecordsOutput, error)

+ GetShardIterator func(ctx context.Context, params *kinesis.GetShardIteratorInput, ...) (*kinesis.GetShardIteratorOutput, error)

+ PutRecord func(ctx context.Context, params *kinesis.PutRecordInput, ...) (*kinesis.PutRecordOutput, error)

+ type KinesisTimeWindow struct

+ End time.Time

+ Start time.Time

+ type KinesisTimeWindowEvent struct

+ EventSourceArn string

+ IsFinalInvokeForWindow bool

+ IsWindowTerminatedEarly bool

+ Records []events.KinesisEventRecord

+ ShardID string

+ State CounterStates

+ Window *KinesisTimeWindow

+ type TimeWindowEventResponse struct

+ BatchItemFailures []BatchItemFailure

+ State CounterStates

+ func (resp *TimeWindowEventResponse) AddBatchItemFailures(items ...BatchItemFailure)

+ func (resp *TimeWindowEventResponse) MergeInto(other *TimeWindowEventResponse)

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL