README
¶
streamfleet-mail
Redis-backed resilient email queue for Go.
The library is tightly coupled with Streamfleet, which provides a work queue that is tolerant to network loss and supports Redis clusters.
Instead of reinventing the wheel, the library wraps around go-mail to provide email/SMTP functionality. It simply provides the client-worker queue system for facilitating sending emails.
Features
- Fully-featured email support via go-mail
- Support for Redis Sentinel and Cluster
- Automatic retry and worker crash recovery
- Tolerance for Redis downtime without data loss
Architecture
The library is broken up into two components, coordinated by Redis:
-
The client, which submits (enqueues) emails to the queue.
-
The worker, which accepts emails from the queue and sends them with a configured email client.
In addition to receiving new emails, workers can claim emails that have sat idle too long in other workers' pending lists. Only workers that have crashed or are unresponsive would have their emails reclaimed.
If an email fails to send while the worker is running, the worker will put the email back on the queue and increment its failure count.
Note that workers and clients may exist on the same process; they do not need to be in separate microservices.
If the underlying Redis server is unavailable, the clients will wait until it comes back online. Emails submitted to the client stay queued in-memory until Redis is available.
To read more about how the underlying queue works, see Streamfleet's documentation.
Use It
To use the library, it is recommended to add it and go-mail to your project:
go get https://github.com/termermc/streamfleet-mail
go get https://github.com/wneessen/go-mail
To see some examples, visit the examples directory.
Simple Example
package main
import (
"context"
"github.com/termermc/streamfleet"
sfm "github.com/termermc/streamfleet-mail"
"github.com/wneessen/go-mail"
)
func main() {
var redisOpt = streamfleet.RedisClientOpt{
Addr: "127.0.0.1:6379",
}
const MailQueueKey = "newsletter"
ctx := context.Background()
// Create the email client used by the worker.
mailClient, err := mail.NewClient("smtp.example.com", mail.WithTLSPolicy(mail.TLSMandatory))
if err != nil {
panic(err)
}
// Create worker.
// ServerUniqueId is used to differentiate workers from each other.
// Some other options we could specify:
// - Logger (an implementation of slog.Logger from the standard library to override the default logger)
// - HandlerConcurrency (how many concurrent emails the worker can send, defaults to 1)
// Take a look at the streamfleet.ServerOpt struct for a complete list of options.
worker, err := sfm.NewWorker(ctx, mailClient, streamfleet.ServerOpt{
ServerUniqueId: "node1.example.com",
RedisOpt: redisOpt,
}, MailQueueKey)
if err != nil {
panic(err)
}
defer func() {
_ = worker.Close()
}()
// Launch worker.
// The Run() method blocks until the worker is closed or runs into a fatal error, so it is launched in its own goroutine.
// Note that workers do not need to be created before clients; queued emails will be picked up as soon as a worker is available.
go func() {
err := worker.Run()
if err != nil {
panic(err)
}
}()
// Create a client using the same Redis connection options as the worker.
// Unlike workers, we do not need to manually provide a unique ID.
// Instead, the client generates its own internally.
// Take a look at the streamfleet.ClientOpt struct for a complete list of options.
// Note that providing an email client is not necessary here, as workers manage their own.
client, err := sfm.NewClient(ctx, streamfleet.ClientOpt{
RedisOpt: redisOpt,
}, MailQueueKey)
if err != nil {
panic(err)
}
defer func() {
_ = client.Close()
}()
// Create a message to send.
msg := mail.NewMsg()
_ = msg.From("huangjin@irs.gov")
_ = msg.To("poorman@gmail.com")
msg.Subject("Money Request")
msg.SetBodyString(mail.TypeTextPlain, "We need all your money. Now.")
// Send the email and let a worker handle it.
// We can also choose to wait for a worker to accept and send the message by calling EnqueueAndTrack instead.
err = client.EnqueueAndForget(msg, streamfleet.TaskOpt{})
if err != nil {
panic(err)
}
println("Email sent!")
}
Acknowledgements
I would like to sincerely thank Vladislav Trotsenko for creating go-smtp-mock, which made creating integration tests for this library much easier.
Documentation
¶
Index ¶
Constants ¶
const QueueKeyPrefix = "sfmail:"
QueueKeyPrefix is the prefix used for the underlying Streamfleet queues used for email queues.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Client ¶
type Client struct {
// contains filtered or unexported fields
}
Client is a mail queue client.
func NewClient ¶
func NewClient(ctx context.Context, sfClientOpt streamfleet.ClientOpt, queueKey string) (*Client, error)
NewClient creates a new mail client. Uses the specified queue key and Streamfleet client options. The client will only be able to send mail on the mail queue with the specified key. Make sure all workers this client is meant to queue to have the same queue key.
func (*Client) Close ¶
Close closes the client and releases all associated resources. The client cannot be used after Close has been called.
func (*Client) EnqueueAndForget ¶
EnqueueAndForget adds an email to the queue and immediately returns. Errors in queuing or the status of the email once it is picked up by a server are not tracked. If you need to know when the email has been sent, use EnqueueAndTrack. If the number of locally queued emails would exceed streamfleet.ClientOpt.MaxLocalQueueSize, returns streamfleet.ErrClientQueueFull. This will normally only happen if the client cannot connect to Redis and emails pile up.
func (*Client) EnqueueAndTrack ¶
func (c *Client) EnqueueAndTrack(email *mail.Msg, opt streamfleet.TaskOpt) (*streamfleet.TaskHandle, error)
EnqueueAndTrack adds an email to the queue and returns a handle to it. Errors in queuing or the status of the email once it is picked up by the server can be tracked using the returned handle. If you do not need to know when the email has been sent, use EnqueueAndForget. You should only use this method if you need to know the status of the email, as it is less efficient than EnqueueAndForget.
If the number of locally queued emails would exceed streamfleet.ClientOpt.MaxLocalQueueSize, returns streamfleet.ErrClientQueueFull. This will normally only happen if the client cannot connect to Redis and emails pile up.
type Worker ¶
type Worker struct {
// contains filtered or unexported fields
}
Worker is a worker that receives emails from the queue and sends them.
func NewWorker ¶
func NewWorker(ctx context.Context, mailClient *mail.Client, sfClientOpt streamfleet.ServerOpt, queueKey string) (*Worker, error)
NewWorker creates a new mail worker that takes emails off the queue and sends them. Uses the specified queue key and Streamfleet server options. The worker will only be able to receive mail from the mail queue with the specified key. Make sure all clients this worker is meant to queue from have the same queue key.
Source Files
Directories