queue

Documentation

Overview

Package queue is in charge of outgoing messages, queueing them when submitted, attempting a first delivery over SMTP, retrying with backoff and sending DSNs for delayed or failed deliveries.

Index

• Variables
• func Add(ctx context.Context, log *mlog.Log, senderAccount string, ...) (int64, error)
• func Count(ctx context.Context) (int, error)
• func Drop(ctx context.Context, ID int64, toDomain string, recipient string) (int, error)
• func Init() error
• func Kick(ctx context.Context, ID int64, toDomain, recipient string, transport *string) (int, error)
• func Shutdown()
• func Start(resolver dns.Resolver, done chan struct{}) error
• type Msg
  • func List(ctx context.Context) ([]Msg, error)
  • func (m Msg) MessagePath() string
  • func (m Msg) Recipient() smtp.Path
  • func (m Msg) Sender() smtp.Path
• type ReadReaderAtCloser
  • func OpenMessage(ctx context.Context, id int64) (ReadReaderAtCloser, error)

Variables

var DB *bstore.DB // Exported for making backups.

var DBTypes = []any{Msg{}} // Types stored in DB.

var Localserve bool

Set for mox localserve, to prevent queueing.

Functions

func Add

func Add(ctx context.Context, log *mlog.Log, senderAccount string, mailFrom, rcptTo smtp.Path, has8bit, smtputf8 bool, size int64, messageID string, msgPrefix []byte, msgFile *os.File, dsnutf8Opt []byte, consumeFile bool) (int64, error)

Add a new message to the queue. The queue is kicked immediately to start a first delivery attempt.

If consumeFile is true, it is removed as part of delivery (by rename or copy and remove). msgFile is never closed by Add.

dnsutf8Opt is a utf8-version of the message, to be used only for DNSs. If set, this data is used as the message when delivering the DSN and the remote SMTP server supports SMTPUTF8. If the remote SMTP server does not support SMTPUTF8, the regular non-utf8 message is delivered.

func Count

func Count(ctx context.Context) (int, error)

Count returns the number of messages in the delivery queue.

func Drop

func Drop(ctx context.Context, ID int64, toDomain string, recipient string) (int, error)

Drop removes messages from the queue that match all nonzero parameters. If all parameters are zero, all messages are removed. Returns number of messages removed.

func Init

func Init() error

Init opens the queue database without starting delivery.

func Kick

func Kick(ctx context.Context, ID int64, toDomain, recipient string, transport *string) (int, error)

Kick sets the NextAttempt for messages matching all filter parameters (ID, toDomain, recipient) that are nonzero, and kicks the queue, attempting delivery of those messages. If all parameters are zero, all messages are kicked. If transport is set, the delivery attempts for the matching messages will use the transport. An empty string is the default transport, i.e. direct delivery. Returns number of messages queued for immediate delivery.

func Shutdown

func Shutdown()

Shutdown closes the queue database. The delivery process isn't stopped. For tests only.

func Start

func Start(resolver dns.Resolver, done chan struct{}) error

Start opens the database by calling Init, then starts the delivery process.

Types

type Msg

type Msg struct {
	ID                 int64
	Queued             time.Time      `bstore:"default now"`
	SenderAccount      string         // Failures are delivered back to this local account. Also used for routing.
	SenderLocalpart    smtp.Localpart // Should be a local user and domain.
	SenderDomain       dns.IPDomain
	RecipientLocalpart smtp.Localpart // Typically a remote user and domain.
	RecipientDomain    dns.IPDomain
	RecipientDomainStr string              // For filtering.
	Attempts           int                 // Next attempt is based on last attempt and exponential back off based on attempts.
	DialedIPs          map[string][]net.IP // For each host, the IPs that were dialed. Used for IP selection for later attempts.
	NextAttempt        time.Time           // For scheduling.
	LastAttempt        *time.Time
	LastError          string
	Has8bit            bool   // Whether message contains bytes with high bit set, determines whether 8BITMIME SMTP extension is needed.
	SMTPUTF8           bool   // Whether message requires use of SMTPUTF8.
	Size               int64  // Full size of message, combined MsgPrefix with contents of message file.
	MessageID          string // Used when composing a DSN, in its References header.
	MsgPrefix          []byte

	// If set, this message is a DSN and this is a version using utf-8, for the case
	// the remote MTA supports smtputf8. In this case, Size and MsgPrefix are not
	// relevant.
	DSNUTF8 []byte

	// If non-empty, the transport to use for this message. Can be set through cli or
	// admin interface. If empty (the default for a submitted message), regular routing
	// rules apply.
	Transport string
}

Msg is a message in the queue.

func List

func List(ctx context.Context) ([]Msg, error)

List returns all messages in the delivery queue. Ordered by earliest delivery attempt first.

func (Msg) MessagePath

func (m Msg) MessagePath() string

MessagePath returns the path where the message is stored.

func (Msg) Recipient

func (m Msg) Recipient() smtp.Path

Recipient of message as used in RCPT TO.

func (Msg) Sender

func (m Msg) Sender() smtp.Path

Sender of message as used in MAIL FROM.

type ReadReaderAtCloser

type ReadReaderAtCloser interface {
	io.ReadCloser
	io.ReaderAt
}

func OpenMessage

func OpenMessage(ctx context.Context, id int64) (ReadReaderAtCloser, error)

OpenMessage opens a message present in the queue.