notification

Documentation

Index

• Constants
• Variables
• func WithNumWorkers(n int) func(d *Dispatcher)
• type Dispatcher
  • func NewDispatcher(cfg *viper.Viper, svc *Services, opts ...func(d *Dispatcher)) *Dispatcher
  • func (d *Dispatcher) Run(ctx context.Context, wg *sync.WaitGroup)
• type HTTPClient
• type Manager
  • func NewManager() *Manager
  • func (m *Manager) Add(ctx context.Context, tx pgx.Tx, n *hub.Notification) error
  • func (m *Manager) GetPending(ctx context.Context, tx pgx.Tx) (*hub.Notification, error)
  • func (m *Manager) UpdateStatus(ctx context.Context, tx pgx.Tx, notificationID string, processed bool, ...) error
• type ManagerMock
  • func (m *ManagerMock) Add(ctx context.Context, tx pgx.Tx, n *hub.Notification) error
  • func (m *ManagerMock) GetPending(ctx context.Context, tx pgx.Tx) (*hub.Notification, error)
  • func (m *ManagerMock) UpdateStatus(ctx context.Context, tx pgx.Tx, notificationID string, processed bool, ...) error
• type Services
• type Worker
  • func NewWorker(svc *Services, c *cache.Cache, baseURL string, httpClient HTTPClient) *Worker
  • func (w *Worker) Run(ctx context.Context, wg *sync.WaitGroup)

Constants

const (

	// DefaultPayloadContentType represents the default content type used for
	// webhooks notifications.
	DefaultPayloadContentType = "application/cloudevents+json"
)

Variables

var DefaultWebhookPayloadTmpl = template.Must(template.New("").Parse(`
{
	"specversion" : "1.0",
	"id" : "{{ .Event.id }}",
	"source" : "https://artifacthub.io/cloudevents",
	"type" : "io.artifacthub.{{ .Event.kind }}",
	"datacontenttype" : "application/json",
	"data" : {
		"package": {
			"name": "{{ .Package.name }}",
			"version": "{{ .Package.version }}",
			"url": "{{ .Package.url }}",
			"repository": {
				"kind": "{{ .Package.repository.kind }}",
				"name": "{{ .Package.repository.name }}",
				"publisher": "{{ .Package.repository.publisher }}"
			}
		}
	}
}
`))

DefaultWebhookPayloadTmpl is the template used for the webhook payload when the webhook uses the default template.

var (
	// ErrRetryable is meant to be used as a wrapper for other errors to
	// indicate the error is not final and the operation should be retried.
	ErrRetryable = errors.New("retryable error")
)

Functions

func WithNumWorkers

func WithNumWorkers(n int) func(d *Dispatcher)

WithNumWorkers allows providing a specific number of workers for a Dispatcher instance.

Types

type Dispatcher

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

Dispatcher handles a group of workers in charge of delivering notifications.

func NewDispatcher

func NewDispatcher(cfg *viper.Viper, svc *Services, opts ...func(d *Dispatcher)) *Dispatcher

NewDispatcher creates a new Dispatcher instance.

func (*Dispatcher) Run

func (d *Dispatcher) Run(ctx context.Context, wg *sync.WaitGroup)

Run starts the workers and lets them run until the dispatcher is asked to stop via the context provided.

type HTTPClient

type HTTPClient interface {
	Do(req *http.Request) (*http.Response, error)
}

HTTPClient defines the methods an HTTPClient implementation must provide.

type Manager

type Manager struct{}

Manager provides an API to manage notifications.

func NewManager

func NewManager() *Manager

NewManager creates a new Manager instance.

func (*Manager) Add

func (m *Manager) Add(ctx context.Context, tx pgx.Tx, n *hub.Notification) error

Add adds the provided notification to the database.

func (*Manager) GetPending

func (m *Manager) GetPending(ctx context.Context, tx pgx.Tx) (*hub.Notification, error)

GetPending returns a pending notification to be delivered if available.

func (*Manager) UpdateStatus

func (m *Manager) UpdateStatus(
	ctx context.Context,
	tx pgx.Tx,
	notificationID string,
	processed bool,
	processedErr error,
) error

UpdateStatus the provided notification status in the database.

type ManagerMock

type ManagerMock struct {
	mock.Mock
}

ManagerMock is a mock implementation of the NotificationManager interface.

func (*ManagerMock) Add

func (m *ManagerMock) Add(ctx context.Context, tx pgx.Tx, n *hub.Notification) error

Add implements the NotificationManager interface.

func (*ManagerMock) GetPending

func (m *ManagerMock) GetPending(ctx context.Context, tx pgx.Tx) (*hub.Notification, error)

GetPending implements the NotificationManager interface.

func (*ManagerMock) UpdateStatus

func (m *ManagerMock) UpdateStatus(
	ctx context.Context,
	tx pgx.Tx,
	notificationID string,
	processed bool,
	processedErr error,
) error

UpdateStatus implements the NotificationManager interface.

type Services

type Services struct {
	DB                  hub.DB
	ES                  hub.EmailSender
	NotificationManager hub.NotificationManager
	SubscriptionManager hub.SubscriptionManager
	RepositoryManager   hub.RepositoryManager
	PackageManager      hub.PackageManager
}

Services is a wrapper around several internal services used to handle notifications deliveries.

type Worker

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

Worker is in charge of delivering notifications to their intended recipients.

func NewWorker

func NewWorker(
	svc *Services,
	c *cache.Cache,
	baseURL string,
	httpClient HTTPClient,
) *Worker

NewWorker creates a new Worker instance.

func (*Worker) Run

func (w *Worker) Run(ctx context.Context, wg *sync.WaitGroup)

Run is the main loop of the worker. It calls processNotification periodically until it's asked to stop via the context provided.