kinesis

Documentation

Index

• Constants
• func CreateStream(streamName string, client *kinesis.Kinesis) error
• func New(lgr log.Logger, registry metrics.Registry, cp cpclient.Coordinator, ...) providers.Provider
• func NewClient(src *KinesisSource) (*kinesis.Kinesis, error)
• func Prepare(img string) (string, error)
• func PutRecord(src *KinesisSource, data []byte, key string) error
• type KinesisSource
  • func MustSource() *KinesisSource
  • func SourceRecipe() (*KinesisSource, error)
  • func (k *KinesisSource) GetProviderType() abstract.ProviderType
  • func (k *KinesisSource) IsAppendOnly() bool
  • func (k *KinesisSource) IsSource()
  • func (k *KinesisSource) Validate() error
  • func (k *KinesisSource) WithDefaults()
• type Provider
  • func (p *Provider) Activate(context.Context, *model.TransferOperation, abstract.TableMap, ...) error
  • func (p *Provider) Source() (abstract.Source, error)
  • func (p *Provider) Type() abstract.ProviderType
• type Source
  • func NewSource(transferID string, cp coordinator.Coordinator, cfg *KinesisSource, ...) (*Source, error)
  • func (s *Source) Fetch() ([]abstract.ChangeItem, error)
  • func (s *Source) Run(sink abstract.AsyncSink) error
  • func (s *Source) Stop()

Constants

const (
	ExpectedBitLength = 186
	SequenceMask      = (1 << 4) - 1
)

const ProviderType = abstract.ProviderType("kinesis")

Functions

func CreateStream

func CreateStream(streamName string, client *kinesis.Kinesis) error

func New

func New(lgr log.Logger, registry metrics.Registry, cp cpclient.Coordinator, transfer *model.Transfer) providers.Provider

func NewClient

func NewClient(src *KinesisSource) (*kinesis.Kinesis, error)

func Prepare

func Prepare(img string) (string, error)

func PutRecord

func PutRecord(src *KinesisSource, data []byte, key string) error

Types

type KinesisSource

type KinesisSource struct {
	Endpoint     string
	Region       string
	Stream       string
	BufferSize   int
	AccessKey    string
	SecretKey    model.SecretString
	ParserConfig map[string]interface{}
}

func MustSource

func MustSource() *KinesisSource

func SourceRecipe

func SourceRecipe() (*KinesisSource, error)

func (*KinesisSource) GetProviderType

func (k *KinesisSource) GetProviderType() abstract.ProviderType

func (*KinesisSource) IsAppendOnly

func (k *KinesisSource) IsAppendOnly() bool

func (*KinesisSource) IsSource

func (k *KinesisSource) IsSource()

func (*KinesisSource) Validate

func (k *KinesisSource) Validate() error

func (*KinesisSource) WithDefaults

func (k *KinesisSource) WithDefaults()

type Provider

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

func (*Provider) Activate

func (p *Provider) Activate(context.Context, *model.TransferOperation, abstract.TableMap, providers.ActivateCallbacks) error

func (*Provider) Source

func (p *Provider) Source() (abstract.Source, error)

func (*Provider) Type

func (p *Provider) Type() abstract.ProviderType

type Source

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

func NewSource

func NewSource(
	transferID string,
	cp coordinator.Coordinator,
	cfg *KinesisSource,
	logger log.Logger,
	registry metrics.Registry,
) (*Source, error)

func (*Source) Fetch

func (s *Source) Fetch() ([]abstract.ChangeItem, error)

func (*Source) Run

func (s *Source) Run(sink abstract.AsyncSink) error

func (*Source) Stop

func (s *Source) Stop()