message

Documentation

Index

• Constants
• Variables
• type DB
  • func NewDB(ctx context.Context, db *sql.DB, a *alertlog.Store, ...) (*DB, error)
  • func (db *DB) SendMessages(ctx context.Context, send SendFunc, status StatusFunc) error
  • func (db *DB) UpdateMessageStatus(ctx context.Context, status *notification.SendResult) error
• type Message
  • func (m Message) Base() nfymsg.Base
• type SendFunc
• type Status
  • func (s Status) IsSent() bool
• type StatusFunc
• type Throttle
  • func NewThrottle(cfg ThrottleConfig, now time.Time, byTypeOnly bool) *Throttle
  • func (tr *Throttle) InCooldown(msg Message) bool
  • func (tr *Throttle) Record(msg Message)
• type ThrottleConfig
• type ThrottleConfigBuilder
  • func (b *ThrottleConfigBuilder) AddRules(rules []ThrottleRule)
  • func (b *ThrottleConfigBuilder) Config() ThrottleConfig
  • func (b *ThrottleConfigBuilder) WithDestTypes(destTypes ...string) *ThrottleConfigBuilder
  • func (b *ThrottleConfigBuilder) WithMsgTypes(msgTypes ...gadb.EnumOutgoingMessagesType) *ThrottleConfigBuilder
• type ThrottleItem
• type ThrottleRule
• type ThrottleRules
  • func (rs ThrottleRules) MaxDuration() time.Duration
  • func (rs ThrottleRules) Rules(Message) []ThrottleRule

Constants

const (
	// StatusPending means the message is waiting to be sent.
	StatusPending = Status("pending")

	// StatusSending means the message is in the process of being sent upstream
	StatusSending = Status("sending")

	// StatusQueuedRemotely means the message has been sent upstream, but is in a remote queue.
	StatusQueuedRemotely = Status("queued_remotely")

	// StatusSent means the message has been sent upstream, and has left the remote queue (if one exists).
	StatusSent = Status("sent")

	// StatusDelivered will be set on delivery if the upstream supports delivery confirmation.
	StatusDelivered = Status("delivered")

	// StatusFailed means the message failed to send.
	StatusFailed = Status("failed")

	// StatusStale is used if the message expired before being sent.
	StatusStale = Status("stale")
)

Defined status values

Variables

var ErrAbort = errors.New("aborted due to pause")

ErrAbort is returned when an early-abort is returned due to pause.

Types

type DB

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

DB implements a priority message sender using Postgres.

func NewDB

func NewDB(ctx context.Context, db *sql.DB, a *alertlog.Store, pausable lifecycle.Pausable) (*DB, error)

NewDB creates a new DB.

func (*DB) SendMessages

func (db *DB) SendMessages(ctx context.Context, send SendFunc, status StatusFunc) error

SendMessages will send notifications using SendFunc.

func (*DB) UpdateMessageStatus

func (db *DB) UpdateMessageStatus(ctx context.Context, status *notification.SendResult) error

UpdateMessageStatus will update the state of a message.

type Message

type Message struct {
	ID         string
	Type       gadb.EnumOutgoingMessagesType
	DestID     notification.DestID
	Dest       gadb.DestV1
	AlertID    int
	AlertLogID int
	VerifyID   string

	UserID     string
	ServiceID  string
	ScheduleID string
	CreatedAt  time.Time
	SentAt     time.Time

	StatusAlertIDs []int64
}

Message represents the data for an outgoing message.

func (Message) Base

func (m Message) Base() nfymsg.Base

type SendFunc

type SendFunc func(context.Context, *Message) (*notification.SendResult, error)

SendFunc defines a function that sends messages.

type Status

type Status string

Status represents the current state of an outgoing message.

func (Status) IsSent

func (s Status) IsSent() bool

IsSent returns true if the message has been successfully sent to the downstream server.

type StatusFunc

type StatusFunc func(ctx context.Context, providerID notification.ProviderMessageID) (*notification.Status, error)

StatusFunc is used to fetch the latest status of a message.

type Throttle

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

Throttle represents the throttled messages for a queue.

func NewThrottle

func NewThrottle(cfg ThrottleConfig, now time.Time, byTypeOnly bool) *Throttle

NewThrottle creates a new Throttle used to manage outgoing messages in a queue.

func (*Throttle) InCooldown

func (tr *Throttle) InCooldown(msg Message) bool

InCooldown returns true or false depending on the cooldown state of a throttled message.

func (*Throttle) Record

func (tr *Throttle) Record(msg Message)

Record keeps track of the outgoing messages being throttled in a queue.

type ThrottleConfig

type ThrottleConfig interface {
	Rules(Message) []ThrottleRule
	MaxDuration() time.Duration
}

ThrottleConfig provides ThrottleRules for a given message.

var GlobalCMThrottle ThrottleConfig = ThrottleRules{{Count: 5, Per: 5 * time.Second}}

GlobalCMThrottle represents the rate limits for each notification type.

var PerCMThrottle ThrottleConfig

PerCMThrottle configures rate limits for individual contact methods.

type ThrottleConfigBuilder

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

ThrottleConfigBuilder can be used to build advanced throttle configurations.

func (*ThrottleConfigBuilder) AddRules

func (b *ThrottleConfigBuilder) AddRules(rules []ThrottleRule)

AddRules will append a set of rules for the current filter (if any).

func (*ThrottleConfigBuilder) Config

func (b *ThrottleConfigBuilder) Config() ThrottleConfig

Config will return a ThrottleConfig for the current top-level configuration.

func (*ThrottleConfigBuilder) WithDestTypes

func (b *ThrottleConfigBuilder) WithDestTypes(destTypes ...string) *ThrottleConfigBuilder

WithDestTypes allows adding rules for messages matching at least one DestType.

func (*ThrottleConfigBuilder) WithMsgTypes

func (b *ThrottleConfigBuilder) WithMsgTypes(msgTypes ...gadb.EnumOutgoingMessagesType) *ThrottleConfigBuilder

WithMsgTypes allows adding rules for messages matching at least one MessageType.

type ThrottleItem

type ThrottleItem struct {
	DestHash  gadb.DestHashV1
	BucketDur time.Duration
}

ThrottleItem represents the messages being throttled.

type ThrottleRule

type ThrottleRule struct {
	Count int
	Per   time.Duration

	// Smooth indicates that the rule should impose a rate limit over the entire duration.
	Smooth bool
}

ThrottleRule sets the number of messages allowed to be sent per set duration.

type ThrottleRules

type ThrottleRules []ThrottleRule

ThrottleRules is a collection of ThrottleRule that implements the ThrottleConfig interface.

func (ThrottleRules) MaxDuration

func (rs ThrottleRules) MaxDuration() time.Duration

MaxDuration returns the longest `Per` value for the set of rules.

func (ThrottleRules) Rules

func (rs ThrottleRules) Rules(Message) []ThrottleRule

Rules always returns the set of configured rules for all messages.