deliverymq

Documentation

Index

• func NewMessageHandler(logger *logging.Logger, redisClient *redis.Client, logMQ LogPublisher, ...) consumer.MessageHandler
• func NewRetryScheduler(deliverymq *DeliveryMQ, redisConfig *redis.RedisConfig) scheduler.Scheduler
• func WithQueue(queueConfig *mqs.QueueConfig) func(opts *DeliveryMQOption)
• type AlertMonitor
• type DeliveryError
  • func (e *DeliveryError) Error() string
  • func (e *DeliveryError) Unwrap() error
• type DeliveryInfra
• type DeliveryMQ
  • func New(opts ...func(opts *DeliveryMQOption)) *DeliveryMQ
  • func (q *DeliveryMQ) Init(ctx context.Context) (func(), error)
  • func (q *DeliveryMQ) Publish(ctx context.Context, event models.DeliveryEvent) error
  • func (q *DeliveryMQ) Subscribe(ctx context.Context) (mqs.Subscription, error)
• type DeliveryMQOption
• type DeliveryTracer
• type DestinationGetter
• type EventGetter
• type LogPublisher
• type PostDeliveryError
  • func (e *PostDeliveryError) Error() string
  • func (e *PostDeliveryError) Unwrap() error
• type PreDeliveryError
  • func (e *PreDeliveryError) Error() string
  • func (e *PreDeliveryError) Unwrap() error
• type Publisher
• type RetryMessage
  • func RetryMessageFromDeliveryEvent(deliveryEvent models.DeliveryEvent) RetryMessage
  • func (m *RetryMessage) FromString(str string) error
  • func (m *RetryMessage) ToDeliveryEvent() models.DeliveryEvent
  • func (m *RetryMessage) ToString() (string, error)
• type RetryScheduler

Functions

func NewMessageHandler

func NewMessageHandler(
	logger *logging.Logger,
	redisClient *redis.Client,
	logMQ LogPublisher,
	entityStore DestinationGetter,
	logStore EventGetter,
	publisher Publisher,
	eventTracer DeliveryTracer,
	retryScheduler RetryScheduler,
	retryBackoff backoff.Backoff,
	retryMaxLimit int,
	alertMonitor AlertMonitor,
) consumer.MessageHandler

func NewRetryScheduler

func NewRetryScheduler(deliverymq *DeliveryMQ, redisConfig *redis.RedisConfig) scheduler.Scheduler

func WithQueue

func WithQueue(queueConfig *mqs.QueueConfig) func(opts *DeliveryMQOption)

Types

type AlertMonitor

type AlertMonitor interface {
	HandleAttempt(ctx context.Context, attempt alert.DeliveryAttempt) error
}

type DeliveryError

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

func (*DeliveryError) Error

func (e *DeliveryError) Error() string

func (*DeliveryError) Unwrap

func (e *DeliveryError) Unwrap() error

type DeliveryInfra

type DeliveryInfra interface {
	DeclareInfrastructure(ctx context.Context) error
}

type DeliveryMQ

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

func New

func New(opts ...func(opts *DeliveryMQOption)) *DeliveryMQ

func (*DeliveryMQ) Init

func (q *DeliveryMQ) Init(ctx context.Context) (func(), error)

func (*DeliveryMQ) Publish

func (q *DeliveryMQ) Publish(ctx context.Context, event models.DeliveryEvent) error

func (*DeliveryMQ) Subscribe

func (q *DeliveryMQ) Subscribe(ctx context.Context) (mqs.Subscription, error)

type DeliveryMQOption

type DeliveryMQOption struct {
	QueueConfig *mqs.QueueConfig
}

type DeliveryTracer

type DeliveryTracer interface {
	Deliver(ctx context.Context, deliveryEvent *models.DeliveryEvent, destination *models.Destination) (context.Context, trace.Span)
}

type DestinationGetter

type DestinationGetter interface {
	RetrieveDestination(ctx context.Context, tenantID, destID string) (*models.Destination, error)
}

type EventGetter

type EventGetter interface {
	RetrieveEvent(ctx context.Context, tenantID, eventID string) (*models.Event, error)
}

type LogPublisher

type LogPublisher interface {
	Publish(ctx context.Context, deliveryEvent models.DeliveryEvent) error
}

type PostDeliveryError

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

func (*PostDeliveryError) Error

func (e *PostDeliveryError) Error() string

func (*PostDeliveryError) Unwrap

func (e *PostDeliveryError) Unwrap() error

type PreDeliveryError

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

Error types to distinguish between different stages of delivery

func (*PreDeliveryError) Error

func (e *PreDeliveryError) Error() string

func (*PreDeliveryError) Unwrap

func (e *PreDeliveryError) Unwrap() error

type Publisher

type Publisher interface {
	PublishEvent(ctx context.Context, destination *models.Destination, event *models.Event) (*models.Delivery, error)
}

type RetryMessage

type RetryMessage struct {
	DeliveryEventID string
	EventID         string
	TenantID        string
	DestinationID   string
	Attempt         int
	Telemetry       *models.DeliveryEventTelemetry
}

func RetryMessageFromDeliveryEvent

func RetryMessageFromDeliveryEvent(deliveryEvent models.DeliveryEvent) RetryMessage

func (*RetryMessage) FromString

func (m *RetryMessage) FromString(str string) error

func (*RetryMessage) ToDeliveryEvent

func (m *RetryMessage) ToDeliveryEvent() models.DeliveryEvent

func (*RetryMessage) ToString

func (m *RetryMessage) ToString() (string, error)

type RetryScheduler

type RetryScheduler interface {
	Schedule(ctx context.Context, task string, delay time.Duration, opts ...scheduler.ScheduleOption) error
	Cancel(ctx context.Context, taskID string) error
}