event

Documentation

Index

• type CSVHandler
  • func NewCSVHandler(awsSession *session.Session, s3Clients map[string]S3Client, ...) *CSVHandler
  • func (handler CSVHandler) Handle(ctx context.Context, event *DimensionsInserted) error
• type Consumer
  • func NewConsumer() *Consumer
  • func (consumer *Consumer) Close(ctx context.Context) (err error)
  • func (consumer *Consumer) Consume(ctx context.Context, messageConsumer kafka.IConsumerGroup, handler Handler, ...)
• type DimensionsInserted
  • func Unmarshal(message kafka.Message) (*DimensionsInserted, error)
• type Handler
• type ObservationWriter
• type S3Client
• type VaultClient

Types

type CSVHandler

type CSVHandler struct {
	AwsSession *session.Session
	// contains filtered or unexported fields
}

CSVHandler handles events to extract observations from CSV files.

func NewCSVHandler

func NewCSVHandler(awsSession *session.Session, s3Clients map[string]S3Client, vaultClient VaultClient, observationWriter ObservationWriter, vaultPath string) *CSVHandler

NewCSVHandler returns a new CSVHandler instance that uses the given file.FileGetter and Output producer.

func (CSVHandler) Handle

func (handler CSVHandler) Handle(ctx context.Context, event *DimensionsInserted) error

Handle takes a single event, and returns the observations gathered from the URL in the event.

type Consumer

type Consumer struct {
	Closing chan bool
	Closed  chan bool
}

Consumer consumes event messages.

func NewConsumer

func NewConsumer() *Consumer

NewConsumer returns a new consumer instance.

func (*Consumer) Close

func (consumer *Consumer) Close(ctx context.Context) (err error)

Close safely closes the consumer and releases all resources

func (*Consumer) Consume

func (consumer *Consumer) Consume(ctx context.Context, messageConsumer kafka.IConsumerGroup, handler Handler, errorReporter reporter.ErrorReporter)

Consume convert them to event instances, and pass the event to the provided handler.

type DimensionsInserted

type DimensionsInserted struct {
	FileURL    string `avro:"file_url"`
	InstanceID string `avro:"instance_id"`
}

DimensionsInserted is the structure of each event consumed by the observation extractor.

func Unmarshal

func Unmarshal(message kafka.Message) (*DimensionsInserted, error)

Unmarshal converts a event instance to []byte.

type Handler

type Handler interface {
	Handle(ctx context.Context, event *DimensionsInserted) error
}

Handler represents a handler for processing a single event.

type ObservationWriter

type ObservationWriter interface {
	WriteAll(ctx context.Context, observationReader observation.Reader, instanceID string)
}

ObservationWriter provides operations for observation output.

type S3Client

type S3Client interface {
	Get(key string) (io.ReadCloser, *int64, error)
	GetWithPSK(key string, psk []byte) (io.ReadCloser, *int64, error)
	Checker(ctx context.Context, state *healthcheck.CheckState) error
}

S3Client represents the S3 client from dp-s3 with the required methods

type VaultClient

type VaultClient interface {
	ReadKey(path, key string) (string, error)
	Checker(ctx context.Context, state *healthcheck.CheckState) error
}

VaultClient is an interface to represent methods called to action upon vault