service

package

v1.5.0 Latest Latest Go to latest Published: Sep 1, 2022 License: MIT Imports: 22 Imported by: 0

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 Responder
type Service
- func New() *Service

Constants ¶

This section is empty.

Variables ¶

View Source

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

View Source

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

GetDatasetAPIClient gets and initialises the DatasetAPI Client

View Source

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

View Source

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

GetHTTPServer creates an http server and sets the Server

View Source

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

View Source

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

View Source

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

GetResponder gets a http request responder

Functions ¶

func GetKafkaProducer ¶ added in v1.2.0

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)
	GetGeographyDimensions(context.Context, cantabular.GetGeographyDimensionsRequest) (*cantabular.GetGeographyDimensionsResponse, error)
	SearchDimensions(ctx context.Context, req cantabular.SearchDimensionsRequest) (*cantabular.GetDimensionsResponse, error)
	StatusCode(error) int
	Checker(context.Context, *healthcheck.CheckState) error
	CheckerAPIExt(context.Context, *healthcheck.CheckState) error
}

type DatasetAPIClient ¶

type DatasetAPIClient interface {
	GetVersion(ctx context.Context, userAuthToken, svcAuthToken, downloadSvcAuthToken, collectionID, datasetID, edition, version string) (dataset.Version, error)
	GetVersionDimensions(ctx context.Context, userAuthToken, serviceAuthToken, collectionID, id, edition, version string) (dataset.VersionDimensions, error)
	GetOptionsInBatches(ctx context.Context, userAuthToken, serviceAuthToken, collectionID, id, edition, version, dimension string, batchSize, maxWorkers int) (dataset.Options, error)
	GetDatasetCurrentAndNext(ctx context.Context, userAuthToken, serviceAuthToken, collectionID, datasetID string) (dataset.Dataset, error)
	GetVersionMetadata(ctx context.Context, userAuthToken, serviceAuthToken, collectionID, id, edition, version string) (dataset.Metadata, error)
	Checker(context.Context, *healthcheck.CheckState) 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)
	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 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

Source Files ¶

View all Source files

Directories ¶

Path	Synopsis
mock

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL