service

Documentation

Index

• Variables
• type Closer
• type HTTPServer
• type HealthChecker
• type MongoDB
• 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) (err error)
  • func (svc *Service) Start(ctx context.Context, svcErrors chan error)

Variables

var GetFilterStore = func(cfg *config.Config) (datastore MongoDB, err error) {
	return mongo.CreateFilterStore(cfg.MongoConfig, cfg.Host)
}

GetFilterStore returns an initialised connection to filter store (mongo database)

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

GetHTTPServer returns an http server

var GetHealthCheck = func(version healthcheck.VersionInfo, criticalTimeout, interval time.Duration) HealthChecker {
	hc := healthcheck.New(version, criticalTimeout, interval)
	return &hc
}

GetHealthCheck returns a healthcheck

var GetProducer = func(ctx context.Context, cfg *config.Config, kafkaBrokers []string, topic string) (kafkaProducer kafka.IProducer, err error) {
	pConfig := &kafka.ProducerConfig{
		KafkaVersion:    &cfg.KafkaVersion,
		MaxMessageBytes: &cfg.KafkaMaxBytes,
	}
	if cfg.KafkaSecProtocol == "TLS" {
		pConfig.SecurityConfig = kafka.GetSecurityConfig(
			cfg.KafkaSecCACerts,
			cfg.KafkaSecClientCert,
			cfg.KafkaSecClientKey,
			cfg.KafkaSecSkipVerify,
		)
	}
	producerChannels := kafka.CreateProducerChannels()
	return kafka.NewProducer(ctx, kafkaBrokers, topic, producerChannels, pConfig)
}

GetProducer returns a kafka producer

Types

type Closer

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

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 MongoDB

type MongoDB interface {
	AddFilter(ctx context.Context, filter *models.Filter) (*models.Filter, error)
	GetFilter(ctx context.Context, filterID, eTagSelector string) (*models.Filter, error)
	UpdateFilter(ctx context.Context, updatedFilter *models.Filter, timestamp primitive.Timestamp, eTagSelector string, currentFilter *models.Filter) (newETag string, err error)
	GetFilterDimension(ctx context.Context, filterID string, name, eTagSelector string) (dimension *models.Dimension, err error)
	AddFilterDimension(ctx context.Context, filterID, name string, options []string, dimensions []models.Dimension, timestamp primitive.Timestamp, eTagSelector string, currentFilter *models.Filter) (newETag string, err error)
	RemoveFilterDimension(ctx context.Context, filterID, name string, timestamp primitive.Timestamp, eTagSelector string, currentFilter *models.Filter) (newETag string, err error)
	AddFilterDimensionOption(ctx context.Context, filterID, name, option string, timestamp primitive.Timestamp, eTagSelector string, currentFilter *models.Filter) (newETag string, err error)
	AddFilterDimensionOptions(ctx context.Context, filterID, name string, options []string, timestamp primitive.Timestamp, eTagSelector string, currentFilter *models.Filter) (newETag string, err error)
	RemoveFilterDimensionOption(ctx context.Context, filterID string, name string, option string, timestamp primitive.Timestamp, eTagSelector string, currentFilter *models.Filter) (newETag string, err error)
	RemoveFilterDimensionOptions(ctx context.Context, filterID string, name string, options []string, timestamp primitive.Timestamp, eTagSelector string, currentFilter *models.Filter) (newETag string, err error)
	CreateFilterOutput(ctx context.Context, filter *models.Filter) error
	GetFilterOutput(ctx context.Context, filterOutputID string) (*models.Filter, error)
	UpdateFilterOutput(ctx context.Context, filter *models.Filter, timestamp primitive.Timestamp) error
	AddEventToFilterOutput(ctx context.Context, filterOutputID string, event *models.Event) error
	Checker(ctx context.Context, state *healthcheck.CheckState) error
	Close(ctx context.Context) error
	RunTransaction(ctx context.Context, retry bool, fn mongodriver.TransactionFunc) (interface{}, error)
}

MongoDB defines the required methods from MongoDB package

type Service

type Service struct {
	Cfg                           *config.Config
	FilterStore                   MongoDB
	FilterOutputSubmittedProducer kafka.IProducer
	IdentityClient                *identity.Client

	HealthCheck HealthChecker
	Server      HTTPServer
	// contains filtered or unexported fields
}

Service contains all the configs, server and clients to run the Dataset API

func New

func New() *Service

New creates a new empty 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) (err error)

Init initialises all the service dependencies, including healthcheck with checkers, api and middleware

func (*Service) Start

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

Start starts an initialised service