service

Documentation

Index

• Variables
• func GetKafkaProducer(ctx context.Context, cfg *config.Config) (kafka.IProducer, error)
• type CantabularClient
• type DatasetAPIClient
• type Datastore
• type Generator
• type HTTPServer
• type HealthChecker
• type MetadataClient
• type PopulationTypesAPIClient
• type Responder
• type Service
  • func New() *Service
  • func (svc *Service) Close(ctx context.Context) error
  • func (svc *Service) Init(ctx context.Context, cfg *config.Config, buildTime, gitCommit, version string) error
  • func (svc *Service) Start(ctx context.Context, svcErrors chan error)

Variables

var GetCantabularClient = func(cfg *config.Config) CantabularClient {
	return cantabular.NewClient(
		cantabular.Config{
			Host:           cfg.CantabularURL,
			ExtApiHost:     cfg.CantabularExtURL,
			GraphQLTimeout: cfg.DefaultRequestTimeout,
		},
		dphttp.NewClient(),
		nil,
	)
}

GetCantabularClient gets and initialises the Cantabular Client

var GetDatasetAPIClient = func(cfg *config.Config) DatasetAPIClient {
	return dataset.NewAPIClient(cfg.DatasetAPIURL)
}

GetDatasetAPIClient gets and initialises the DatasetAPI Client

var GetGenerator = func() Generator {
	return generator.New()
}

var GetHTTPServer = func(bindAddr string, router http.Handler) HTTPServer {
	otelHandler := otelhttp.NewHandler(router, "/")
	s := dphttp.NewServer(bindAddr, otelHandler)
	s.HandleOSSignals = false
	return s
}

GetHTTPServer creates an http server and sets the Server

var GetHealthCheck = func(cfg *config.Config, buildTime, gitCommit, version string) (HealthChecker, error) {
	versionInfo, err := healthcheck.NewVersionInfo(buildTime, gitCommit, version)
	if err != nil {
		return nil, fmt.Errorf("failed to get version info: %w", err)
	}

	hc := healthcheck.New(
		versionInfo,
		cfg.HealthCheckCriticalTimeout,
		cfg.HealthCheckInterval,
	)
	return &hc, nil
}

GetHealthCheck creates a healthcheck with versionInfo

var GetMetadataClient = func(cfg *config.Config) MetadataClient {

	return cantabularmetadata.NewClient(cantabularmetadata.Config{
		Host:           cfg.MetadataAPIURL,
		GraphQLTimeout: cfg.DefaultRequestTimeout,
	}, dphttp.NewClient())
}

var GetMongoDB = func(ctx context.Context, cfg *config.Config, g Generator) (Datastore, error) {
	return mongo.NewClient(ctx, g, mongo.Config{
		MongoDriverConfig:       cfg.Mongo,
		FilterAPIURL:            cfg.FilterAPIURL,
		FiltersCollection:       cfg.FiltersCollection,
		FilterOutputsCollection: cfg.FilterOutputsCollection,
	})
}

var GetPopulationClient = func(cfg *config.Config) (PopulationTypesAPIClient, error) {
	return population.NewClient(cfg.PopulationTypesAPIURL)
}

GetPopulationClient gets and initialises the PopultaionTypesAPI Client

var GetResponder = func() Responder {
	return responder.New()
}

GetResponder gets a http request responder

Functions

func GetKafkaProducer

func GetKafkaProducer(ctx context.Context, cfg *config.Config) (kafka.IProducer, error)

GetKafkaProducer creates a Kafka producer. Currently just for POST filters/{id}/submit

Types

type CantabularClient

type CantabularClient interface {
	GetDimensionOptions(context.Context, cantabular.GetDimensionOptionsRequest) (*cantabular.GetDimensionOptionsResponse, error)
	StaticDatasetQuery(context.Context, cantabular.StaticDatasetQueryRequest) (*cantabular.StaticDatasetQuery, error)
	GetGeographyDimensionsInBatches(ctx context.Context, datasetID string, batchSize, maxWorkers int) (*gql.Dataset, error)
	GetDimensionsByName(context.Context, cantabular.GetDimensionsByNameRequest) (*cantabular.GetDimensionsResponse, error)
	GetArea(context.Context, cantabular.GetAreaRequest) (*cantabular.GetAreaResponse, error)
	StatusCode(error) int
	Checker(context.Context, *healthcheck.CheckState) error
	CheckerAPIExt(context.Context, *healthcheck.CheckState) error
	GetCategorisations(context.Context, cantabular.GetCategorisationsRequest) (*cantabular.GetCategorisationsResponse, error)
	CheckQueryCount(context.Context, cantabular.StaticDatasetQueryRequest) (int, error)
	StaticDatasetQueryStreamJSON(context.Context, cantabular.StaticDatasetQueryRequest, stream.Consumer) (cantabular.GetObservationsResponse, error)
}

type DatasetAPIClient

type DatasetAPIClient interface {
	GetVersion(ctx context.Context, userAuthToken, svcAuthToken, downloadSvcAuthToken, collectionID, datasetID, edition, version string) (dataset.Version, error)
	GetOptionsInBatches(ctx context.Context, userAuthToken, serviceAuthToken, collectionID, id, edition, version, dimension string, batchSize, maxWorkers int) (dataset.Options, error)
	GetMetadataURL(id, edition, version string) string
	Checker(context.Context, *healthcheck.CheckState) error
	GetDatasetCurrentAndNext(ctx context.Context, userAuthToken, serviceAuthToken, collectionID, datasetID string) (m dataset.Dataset, err error)
}

type Datastore

type Datastore interface {
	CreateFilter(context.Context, *model.Filter) error
	GetFilter(context.Context, string) (*model.Filter, error)
	UpdateFilterOutput(context.Context, *model.FilterOutput) error
	CreateFilterOutput(context.Context, *model.FilterOutput) error
	GetFilterOutput(context.Context, string) (*model.FilterOutput, error)
	AddFilterOutputEvent(context.Context, string, *model.Event) error
	GetFilterDimensions(context.Context, string, int, int) ([]model.Dimension, int, error)
	GetFilterDimension(ctx context.Context, fID, dimName string) (model.Dimension, error)
	GetFilterDimensionOptions(context.Context, string, string, int, int) ([]string, int, string, error)
	DeleteFilterDimensionOptions(context.Context, string, string) (string, error)
	DeleteFilterDimension(context.Context, string, string) (string, error)
	AddFilterDimension(ctx context.Context, s string, dimension model.Dimension) error
	UpdateFilterDimension(ctx context.Context, filterID string, dimensionName string, dimension model.Dimension, currentETag string) (eTag string, err error)
	RemoveFilterDimensionOption(ctx context.Context, filterID, dimension, option, currentETag string) (eTag string, err error)
	Checker(context.Context, *healthcheck.CheckState) error
	Conn() *mongo.MongoConnection
}

Datastore is the interface for interacting with the storage backend

type Generator

type Generator interface {
	PSK() ([]byte, error)
	UUID() (uuid.UUID, error)
	Timestamp() time.Time
	UniqueTimestamp() primitive.Timestamp
	URL(host, path string, args ...interface{}) string
}

Generator is the interface for generating dynamic tokens and timestamps

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(http.ResponseWriter, *http.Request)
	Start(context.Context)
	Stop()
	AddAndGetCheck(name string, checker healthcheck.Checker) (*healthcheck.Check, error)
	Subscribe(healthcheck.Subscriber, ...*healthcheck.Check)
}

HealthChecker defines the required methods from Healthcheck

type MetadataClient

type MetadataClient interface {
	GetDefaultClassification(ctx context.Context, req cantabularmetadata.GetDefaultClassificationRequest) (*cantabularmetadata.GetDefaultClassificationResponse, error)
}

type PopulationTypesAPIClient

type PopulationTypesAPIClient interface {
	Checker(context.Context, *healthcheck.CheckState) error
	GetCategorisations(context.Context, population.GetCategorisationsInput) (population.GetCategorisationsResponse, error)
	GetPopulationTypeMetadata(context.Context, population.GetPopulationTypeMetadataInput) (population.GetPopulationTypeMetadataResponse, error)
}

type Responder

type Responder interface {
	JSON(context.Context, http.ResponseWriter, int, interface{})
	Error(context.Context, http.ResponseWriter, int, error)
	StatusCode(http.ResponseWriter, int)
	Bytes(context.Context, http.ResponseWriter, int, []byte)
	Errors(context.Context, http.ResponseWriter, int, []error)
}

Responder handles responding to http requests

type Service

type Service struct {
	Cfg         *config.Config
	Server      HTTPServer
	HealthCheck HealthChecker
	API         *api.API

	Producer kafka.IProducer
	// contains filtered or unexported fields
}

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

func New

func New() *Service

func (*Service) Close

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

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

func (*Service) Init

func (svc *Service) Init(ctx context.Context, cfg *config.Config, buildTime, gitCommit, version string) error

Init initialises the service and it's dependencies

func (*Service) Start

func (svc *Service) Start(ctx context.Context, svcErrors chan error)

Start the service