service

Documentation

Index

• type DatasetAPI
• type EventConsumer
• type ExternalServiceList
  • func NewServiceList(initialiser Initialiser) *ExternalServiceList
  • func (e *ExternalServiceList) GetDatasetAPIClient(cfg *config.Config) (DatasetAPI, error)
  • func (e *ExternalServiceList) GetFileStore(cfg *config.Config) (fileStore *file.Store, err error)
  • func (e *ExternalServiceList) GetFilterStore(cfg *config.Config, serviceAuthToken string) (FilterStore, error)
  • func (e *ExternalServiceList) GetHTTPServer(bindAddr string, router http.Handler) HTTPServer
  • func (e *ExternalServiceList) GetHealthCheck(cfg *config.Config, buildTime, gitCommit, version string) (HealthChecker, error)
  • func (e *ExternalServiceList) GetKafkaConsumer(ctx context.Context, cfg *config.Config) (dpkafka.IConsumerGroup, error)
  • func (e *ExternalServiceList) GetKafkaProducer(ctx context.Context, cfg *config.Config) (dpkafka.IProducer, error)
  • func (e *ExternalServiceList) GetObservationStore(ctx context.Context) (observationStore *graph.DB, err error)
• type FilterStore
• type HTTPServer
• type HealthChecker
• type Init
  • func (e *Init) DoGetDatasetAPIClient(cfg *config.Config) DatasetAPI
  • func (e *Init) DoGetFileStore(cfg *config.Config) (fileStore *file.Store, err error)
  • func (e *Init) DoGetFilterStore(cfg *config.Config, serviceAuthToken string) FilterStore
  • func (e *Init) DoGetHTTPServer(bindAddr string, router http.Handler) HTTPServer
  • func (e *Init) DoGetHealthCheck(cfg *config.Config, buildTime, gitCommit, version string) (HealthChecker, error)
  • func (e *Init) DoGetKafkaConsumer(ctx context.Context, kafkaCfg *config.KafkaConfig) (dpkafka.IConsumerGroup, error)
  • func (e *Init) DoGetKafkaProducer(ctx context.Context, cfg *config.Config) (dpkafka.IProducer, error)
  • func (e *Init) DoGetObservationStore(ctx context.Context) (observationStore *graph.DB, err error)
• type Initialiser
• type Service
  • func Run(ctx context.Context, serviceList *ExternalServiceList, ...) (*Service, error)
  • func (svc *Service) Close(ctx context.Context) error

Types

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, ifMatch string) (m dataset.Instance, eTag string, 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)
	GetOptions(ctx context.Context, userAuthToken, serviceAuthToken, collectionID, id, edition, version, dimension string, q *dataset.QueryParams) (m dataset.Options, err error)
	GetVersionDimensions(ctx context.Context, userAuthToken, serviceAuthToken, collectionID, id, edition, version string) (m dataset.VersionDimensions, err error)
	GetOptionsInBatches(ctx context.Context, userAuthToken, serviceAuthToken, collectionID, id, edition, version, dimension string, batchSize, maxWorkers int) (m dataset.Options, err error)
	Checker(context.Context, *healthcheck.CheckState) error
}

DatasetAPI contains functions to call the dataset API.

type EventConsumer

type EventConsumer interface {
	Close(ctx context.Context) (err error)
}

EventConsumer defines the required methods from event Consumer

type ExternalServiceList

type ExternalServiceList struct {
	HealthCheck      bool
	KafkaConsumer    bool
	KafkaProducer    bool
	DatasetAPI       bool
	FilterStore      bool
	ObservationStore bool
	FileStore        bool
	Init             Initialiser
}

ExternalServiceList holds the initialiser and initialisation state of external services.

func NewServiceList

func NewServiceList(initialiser Initialiser) *ExternalServiceList

NewServiceList creates a new service list with the provided initialiser

func (*ExternalServiceList) GetDatasetAPIClient

func (e *ExternalServiceList) GetDatasetAPIClient(cfg *config.Config) (DatasetAPI, error)

GetDatasetAPIClient gets and initialises the DatasetAPI Client

func (*ExternalServiceList) GetFileStore

func (e *ExternalServiceList) GetFileStore(cfg *config.Config) (fileStore *file.Store, err error)

GetFileStore returns an initialised connection to file store

func (*ExternalServiceList) GetFilterStore

func (e *ExternalServiceList) GetFilterStore(cfg *config.Config, serviceAuthToken string) (FilterStore, error)

GetFilterStore returns a filter client

func (*ExternalServiceList) GetHTTPServer

func (e *ExternalServiceList) GetHTTPServer(bindAddr string, router http.Handler) HTTPServer

GetHTTPServer creates an http server and sets the Server flag to true

func (*ExternalServiceList) GetHealthCheck

func (e *ExternalServiceList) GetHealthCheck(cfg *config.Config, buildTime, gitCommit, version string) (HealthChecker, error)

GetHealthCheck creates a healthcheck with versionInfo and sets the HealthCheck flag to true

func (*ExternalServiceList) GetKafkaConsumer

func (e *ExternalServiceList) GetKafkaConsumer(ctx context.Context, cfg *config.Config) (dpkafka.IConsumerGroup, error)

GetKafkaConsumer creates a Kafka consumer and sets the consumer flag to true

func (*ExternalServiceList) GetKafkaProducer

func (e *ExternalServiceList) GetKafkaProducer(ctx context.Context, cfg *config.Config) (dpkafka.IProducer, error)

GetKafkaConsumer creates a Kafka producer and sets the producer flag to true

func (*ExternalServiceList) GetObservationStore

func (e *ExternalServiceList) GetObservationStore(ctx context.Context) (observationStore *graph.DB, err error)

GetObservationStore returns an initialised connection to observation store (graph database)

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
	Checker(context.Context, *healthcheck.CheckState) error
}

FilterStore provides existing filter data.

type HTTPServer

type HTTPServer interface {
	ListenAndServe() error
	Shutdown(ctx context.Context) error
}

HTTPServer defines the required methods from the HTTP server

type HealthChecker

type HealthChecker interface {
	Handler(w http.ResponseWriter, req *http.Request)
	Start(ctx context.Context)
	Stop()
	AddCheck(name string, checker healthcheck.Checker) (err error)
}

HealthChecker defines the required methods from Healthcheck

type Init

type Init struct{}

Init implements the Initialiser interface to initialise dependencies

func (*Init) DoGetDatasetAPIClient

func (e *Init) DoGetDatasetAPIClient(cfg *config.Config) DatasetAPI

DoGetDatasetAPIClient creates a datasetAPI client

func (*Init) DoGetFileStore

func (e *Init) DoGetFileStore(cfg *config.Config) (fileStore *file.Store, err error)

DoGetFileStore creates a connection to s3

func (*Init) DoGetFilterStore

func (e *Init) DoGetFilterStore(cfg *config.Config, serviceAuthToken string) FilterStore

DoGetFilterStore creates a connection to the Filter data store

func (*Init) DoGetHTTPServer

func (e *Init) DoGetHTTPServer(bindAddr string, router http.Handler) HTTPServer

DoGetHTTPServer creates an HTTP Server with the provided bind address and router

func (*Init) DoGetHealthCheck

func (e *Init) DoGetHealthCheck(cfg *config.Config, buildTime, gitCommit, version string) (HealthChecker, error)

DoGetHealthCheck creates a healthcheck with versionInfo

func (*Init) DoGetKafkaConsumer

func (e *Init) DoGetKafkaConsumer(ctx context.Context, kafkaCfg *config.KafkaConfig) (dpkafka.IConsumerGroup, error)

DoGetKafkaConsumer returns a Kafka Consumer group

func (*Init) DoGetKafkaProducer

func (e *Init) DoGetKafkaProducer(ctx context.Context, cfg *config.Config) (dpkafka.IProducer, error)

DoGetKafkaProducer creates a kafka producer

func (*Init) DoGetObservationStore

func (e *Init) DoGetObservationStore(ctx context.Context) (observationStore *graph.DB, err error)

DoGetObservationStore creates a graph DB client

type Initialiser

type Initialiser interface {
	DoGetDatasetAPIClient(cfg *config.Config) DatasetAPI
	DoGetFilterStore(cfg *config.Config, serviceAuthToken string) FilterStore
	DoGetObservationStore(ctx context.Context) (observationStore *graph.DB, err error)
	DoGetFileStore(cfg *config.Config) (fileStore *file.Store, err error)
	DoGetHTTPServer(bindAddr string, router http.Handler) HTTPServer
	DoGetHealthCheck(cfg *config.Config, buildTime, gitCommit, version string) (HealthChecker, error)
	DoGetKafkaConsumer(ctx context.Context, kafkaCfg *config.KafkaConfig) (kafka.IConsumerGroup, error)
	DoGetKafkaProducer(ctx context.Context, cfg *config.Config) (kafka.IProducer, error)
}

Initialiser defines the methods to initialise external services

type Service

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

Service contains all the configs, server and clients to run the event handler service

func Run

func Run(ctx context.Context, serviceList *ExternalServiceList, buildTime, gitCommit, version string, svcErrors chan error) (*Service, error)

Run the service

func (*Service) Close

func (svc *Service) Close(ctx context.Context) error

Close gracefully shuts the service down in the required order, with timeout