event

Documentation

Index

• type AvroProducer
  • func NewAvroProducer(messageProducer chan []byte, marshaller Marshaller) *AvroProducer
  • func (producer *AvroProducer) CSVExported(event *CSVExported) error
• type CSVExported
• type Consumer
  • func NewConsumer() *Consumer
  • func (consumer *Consumer) Close(ctx context.Context) (err error)
  • func (consumer *Consumer) Consume(messageConsumer MessageConsumer, handler Handler, errorHandler errors.Handler)
• type DatasetAPI
• type ExportHandler
  • func NewExportHandler(filterStore FilterStore, observationStore ObservationStore, ...) *ExportHandler
  • func (handler *ExportHandler) Handle(ctx context.Context, event *FilterSubmitted) error
• type FileStore
• type FilterStore
• type FilterSubmitted
• type Handler
• type Marshaller
• type MessageConsumer
• type ObservationStore
• type Producer

Types

type AvroProducer

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

AvroProducer of output events.

func NewAvroProducer

func NewAvroProducer(messageProducer chan []byte, marshaller Marshaller) *AvroProducer

NewAvroProducer returns a new instance of AvroProducer.

func (*AvroProducer) CSVExported

func (producer *AvroProducer) CSVExported(event *CSVExported) error

CSVExported produces a new CSV exported event for the given filter output ID.

type CSVExported

type CSVExported struct {
	FilterID   string `avro:"filter_output_id"`
	FileURL    string `avro:"file_url"`
	InstanceID string `avro:"instance_id"`
	DatasetID  string `avro:"dataset_id"`
	Edition    string `avro:"edition"`
	Version    string `avro:"version"`
	Filename   string `avro:"filename"`
	RowCount   int32  `avro:"row_count"`
}

CSVExported provides event data for a single exported CSV

type Consumer

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

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(messageConsumer MessageConsumer, handler Handler, errorHandler errors.Handler)

Consume converts messages to event instances, and pass the event to the provided handler.

type DatasetAPI

type DatasetAPI interface {
	PutVersion(ctx context.Context, userAuthToken, serviceAuthToken, collectionID, datasetID, edition, version string, m dataset.Version) error
	GetVersion(ctx context.Context, userAuthToken, serviceAuthToken, downloadServiceAuthToken, collectionID, datasetID, edition, version string) (m dataset.Version, err error)
	GetInstance(ctx context.Context, userAuthToken, serviceAuthToken, collectionID, instanceID string) (m dataset.Instance, err error)
	GetMetadataURL(id, edition, version string) (url string)
	GetVersionMetadata(ctx context.Context, userAuthToken, serviceAuthToken, collectionID, id, edition, version string) (m dataset.Metadata, err error)
}

DatasetAPI contains functions to call the dataset API.

type ExportHandler

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

ExportHandler handles a single CSV export of a filtered dataset.

func NewExportHandler

func NewExportHandler(
	filterStore FilterStore,
	observationStore ObservationStore,
	fileStore FileStore,
	eventProducer Producer,
	datasetAPI DatasetAPI,
	cfg *config.Config,
) *ExportHandler

NewExportHandler returns a new instance using the given dependencies.

func (*ExportHandler) Handle

func (handler *ExportHandler) Handle(ctx context.Context, event *FilterSubmitted) error

Handle the export of a single filter output.

type FileStore

type FileStore interface {
	PutFile(ctx context.Context, reader io.Reader, filename string, isPublished bool) (url string, err error)
}

FileStore provides storage for filtered output files.

type FilterStore

type FilterStore interface {
	GetFilter(ctx context.Context, filterID string) (*filter.Model, error)
	PutCSVData(ctx context.Context, filterID string, downloadItem filter.Download) error
	PutStateAsEmpty(ctx context.Context, filterJobID string) error
	PutStateAsError(ctx context.Context, filterJobID string) error
}

FilterStore provides existing filter data.

type FilterSubmitted

type FilterSubmitted struct {
	FilterID   string `avro:"filter_output_id"`
	InstanceID string `avro:"instance_id"`
	DatasetID  string `avro:"dataset_id"`
	Edition    string `avro:"edition"`
	Version    string `avro:"version"`
}

FilterSubmitted is the structure of each event consumed.

type Handler

type Handler interface {
	Handle(ctx context.Context, filterSubmittedEvent *FilterSubmitted) error
}

Handler represents a handler for processing a single event.

type Marshaller

type Marshaller interface {
	Marshal(s interface{}) ([]byte, error)
}

Marshaller marshals events into messages.

type MessageConsumer

type MessageConsumer interface {
	Channels() *kafka.ConsumerGroupChannels
	CommitAndRelease(kafka.Message)
}

MessageConsumer provides a generic interface for consuming []byte messages

type ObservationStore

type ObservationStore interface {
	StreamCSVRows(ctx context.Context, instanceID, filterID string, filters *observation.DimensionFilters, limit *int) (observation.StreamRowReader, error)
}

ObservationStore provides filtered observation data in CSV rows.

type Producer

type Producer interface {
	CSVExported(e *CSVExported) error
}

Producer handles producing output events.