use tokio::sync::mpsc::{self, Sender};
use tokio::sync::mpsc::error::TrySendError;
use crate::email::{EmailJob, EmailMessage};
use crate::worker::EmailWorker;
#[derive(Debug)]
pub enum QueueError {
Full,
Closed,
}
#[derive(Clone)]
pub struct EmailQueue {
tx: Sender<EmailJob>,
}
impl EmailQueue {
pub fn start(capacity: usize) -> Self {
let (tx, rx) = mpsc::channel(capacity);
let worker = EmailWorker::new(rx, tx.clone());
tokio::spawn(worker.run());
EmailQueue { tx }
}
pub fn enqueue(&self, message: EmailMessage) -> Result<(), QueueError> {
let job = EmailJob::new(message);
self.tx.try_send(job).map_err(|e| match e {
TrySendError::Full(_) => QueueError::Full,
TrySendError::Closed(_) => QueueError::Closed,
})
}
}
use std::time::Duration;
#[derive(Debug, Clone)]
pub struct EmailMessage {
pub to: String,
pub subject: String,
pub body: String,
}
#[derive(Debug, Clone)]
pub struct EmailJob {
pub message: EmailMessage,
pub attempt: u32,
pub max_attempts: u32,
}
impl EmailJob {
pub fn new(message: EmailMessage) -> Self {
EmailJob { message, attempt: 0, max_attempts: 5 }
}
pub fn can_retry(&self) -> bool {
self.attempt + 1 < self.max_attempts
}
pub fn next_attempt(&mut self) {
self.attempt += 1;
}
pub fn backoff(&self) -> Duration {
let base = 200u64;
let millis = base.saturating_mul(1u64 << self.attempt.min(8));
Duration::from_millis(millis.min(30_000))
}
}
use tokio::sync::mpsc::{Receiver, Sender};
use tokio::time::sleep;
use crate::email::EmailJob;
pub struct EmailWorker {
rx: Receiver<EmailJob>,
tx: Sender<EmailJob>,
}
impl EmailWorker {
pub fn new(rx: Receiver<EmailJob>, tx: Sender<EmailJob>) -> Self {
EmailWorker { rx, tx }
}
pub async fn run(mut self) {
while let Some(job) = self.rx.recv().await {
match Self::deliver(&job).await {
Ok(()) => {
println!("delivered to {}", job.message.to);
}
Err(err) if job.can_retry() => {
let mut retry = job.clone();
retry.next_attempt();
let delay = retry.backoff();
eprintln!("delivery failed ({err}), retry #{} in {:?}", retry.attempt, delay);
let tx = self.tx.clone();
tokio::spawn(async move {
sleep(delay).await;
let _ = tx.send(retry).await;
});
}
Err(err) => {
eprintln!("dropping job for {} after {} attempts: {err}", job.message.to, job.attempt + 1);
}
}
}
println!("email worker shutting down");
}
async fn deliver(job: &EmailJob) -> Result<(), String> {
// Placeholder for an SMTP client call.
if job.message.to.contains('@') {
Ok(())
} else {
Err(format!("invalid recipient {}", job.message.to))
}
}
}
This snippet shows a small but realistic asynchronous email queue built on tokio, split into the message type and job wrapper, the background worker that drains a channel, and the client-facing queue handle. The design separates the producer side (anything that wants to send mail) from the consumer side (a single worker task) using an mpsc channel, which is the idiomatic way to fan work into a dedicated processor without sharing mutable state.
In email.rs, EmailMessage is the plain payload, while EmailJob wraps it with retry bookkeeping: an attempt counter and a max_attempts ceiling. The backoff method computes an exponential delay (base * 2^attempt) capped at 30 seconds, which spreads out retries so a flaky SMTP endpoint isn't hammered. Keeping the retry policy on the job itself means the worker stays generic and the message carries its own fate.
In worker.rs, EmailWorker::run loops with while let Some(job) = self.rx.recv().await, so it processes jobs one at a time and exits cleanly when every sender is dropped. On failure it inspects job.can_retry(); if attempts remain it calls job.next_attempt(), sleeps for backoff() via tokio::time::sleep, and re-enqueues through a cloned Sender. Re-queuing rather than blocking the worker on a sleep in-line is a deliberate trade-off — here the code sleeps before re-sending so ordering is simple, but the Sender clone shows how a job can be pushed back for later. Exhausted jobs are logged and dropped, which acts as a crude dead-letter path.
In queue.rs, EmailQueue owns the Sender and spawns the worker with tokio::spawn, returning a cheap, clonable handle. enqueue wraps a message in a fresh EmailJob and uses try_send so a full bounded channel surfaces backpressure as QueueError::Full instead of silently blocking the caller.
The key ideas worth internalizing: bounded channels give backpressure, dropping all senders is the natural shutdown signal, and attaching retry state to each job keeps the worker loop simple. A pitfall to note is that re-enqueueing after a sleep can reorder jobs and, with a bounded channel, a retry send can itself fail — production code would route those to a persistent store rather than a log line.
Related snips
export interface RetryOptions {
retries: number;
baseMs: number;
maxMs: number;
signal?: AbortSignal;
onRetry?: (attempt: number, delay: number, err: unknown) => void;Exponential backoff with jitter for retries
struct Config<'a> {
name: &'a str,
value: &'a str,
}
fn parse_config(line: &str) -> Config {Lifetime annotations for flexible borrowing in structs
package dbutil
import (
"context"
"github.com/jackc/pgconn"Retry Postgres serialization failures with bounded attempts
use crossbeam::channel::unbounded;
use std::thread;
fn main() {
let (tx, rx) = unbounded();Crossbeam for advanced concurrent data structures
// Creating a Promise
const myPromise = new Promise((resolve, reject) => {
const success = true;
setTimeout(() => {
if (success) {Promises and async/await patterns for asynchronous JavaScript
use tracing::{info, instrument};
#[instrument]
fn process_request(user_id: u64) {
info!(user_id, "Processing request");
// Work happens heretracing for structured logging and distributed tracing
Share this code
Here's the card — post it anywhere.
