Documentation
¶
Overview ¶
Package taskqueue is what the producers and the consumers of the asynq task queue share: the task type names, the payloads, the task IDs that deduplicate enqueues, and the client and worker middleware that carry the OpenTelemetry trace context across the queue. Producers depend on these definitions, never on the consumers in internal/transport/task.
asynq has no message headers, so the producer wraps the task payload in an envelope holding the W3C trace context, and the worker-side middleware unwraps it, resumes the producer's trace and opens a consumer span around the handler. Payloads without an envelope (older in-flight tasks, traceless producers, scheduler ticks) pass through untouched and get a root span, so every task execution logs a trace id either way.
Index ¶
- Constants
- func ActivationTaskID(orderNo string) string
- func EventTaskID(id int64) string
- func Middleware() asynq.MiddlewareFunc
- func Wrap(ctx context.Context, task *asynq.Task) *asynq.Task
- type Client
- type DeferCloseOrderPayload
- type EventDeliverPayload
- type ForthwithActivateOrderPayload
- type SendEmailPayload
- type SendSmsPayload
Constants ¶
const ( EmailTypeVerify = "verify" EmailTypeMaintenance = "maintenance" EmailTypeExpiration = "expiration" EmailTypeTrafficExceed = "traffic_exceed" EmailTypeCustom = "custom" )
The email types of SendEmailPayload: each selects the configured subject and template, except EmailTypeCustom, whose payload carries the text.
const ( DeferCloseOrder = "defer:order:close" ForthwithActivateOrder = "forthwith:order:activate" SchedulerReconcilePaidOrders = "scheduler:order:reconcile-paid" SchedulerReconcilePendingOrders = "scheduler:order:reconcile-pending" SchedulerPublishOrderEvents = "scheduler:order:publish-events" SchedulerCleanupOrderEvents = "scheduler:order:cleanup-events" // SchedulerDailyOrderReport pushes the previous day's settlement summary // to the administrators bound on Telegram. SchedulerDailyOrderReport = "scheduler:order:daily-report" )
The order task types.
const ( SchedulerCheckSubscription = "scheduler:check:subscription" // SchedulerRemindExpiringSubscriptions warns owners whose subscription // expires soon. Daily, at a civil hour. SchedulerRemindExpiringSubscriptions = "scheduler:subscription:remind-expiring" // SchedulerDispatchDomainEvents pumps the generic domain-event outbox // onto the asynq queue as events:deliver tasks. SchedulerDispatchDomainEvents = "scheduler:events:dispatch" SchedulerResetTraffic = "scheduler:reset:traffic" SchedulerTrafficStat = "scheduler:traffic:stat" SchedulerLogCleanup = "scheduler:log:cleanup" SchedulerFlushTraffic = "scheduler:flush:traffic" )
The task types of the scheduled jobs.
const ( // ScheduledBatchSendEmail runs a batch email task. ScheduledBatchSendEmail = "scheduled:email:batch" // ForthwithQuotaTask runs a quota task right away. ForthwithQuotaTask = "forthwith:quota:task" // SchedulerExchangeRate refreshes the currency exchange rates. SchedulerExchangeRate = "scheduler:exchange:rate" )
const BatchEmailTaskTimeout = 6 * time.Hour
BatchEmailTaskTimeout bounds one run of a batch email task, as the asynq.Timeout of every batch email enqueue (the initial one and each continuation). asynq's default is 30 minutes, which a paced campaign to thousands of recipients overruns; the worker stops itself before the deadline and the campaign continues from a follow-up task, so the timeout only has to be long enough for a chunk worth running.
const ( // EventDeliver carries one domain event from the outbox to its topic // subscribers; asynq is the broker (retry, backoff, dead-letter archive). EventDeliver = "events:deliver" )
const (
// ForthwithSendEmail sends one email (SendEmailPayload).
ForthwithSendEmail = "forthwith:email:send"
)
const (
// ForthwithSendSms sends one text message (SendSmsPayload).
ForthwithSendSms = "forthwith:sms:send"
)
Variables ¶
This section is empty.
Functions ¶
func ActivationTaskID ¶
ActivationTaskID is the task ID of an order's activation: every producer (payment callback, reconciliation) enqueues under the same ID, so a paid order has one activation task queued at a time, however often its payment is reported.
func EventTaskID ¶
EventTaskID deduplicates enqueues of the same outbox event: the publish pump may replay an enqueue when marking the event published fails, and the task-id conflict turns that replay into a no-op.
func Middleware ¶
func Middleware() asynq.MiddlewareFunc
Middleware resumes the producer's trace from the payload envelope (when present), hands the handler the unwrapped payload, and records the task execution as a consumer span.
Types ¶
type Client ¶
Client wraps *asynq.Client so EnqueueContext stamps the caller's trace context onto the task. Wrapping rebuilds the task, which drops options embedded at NewTask time — pass options to EnqueueContext instead. The plain Enqueue promotes from the inner client untraced (it has no context to read).
type DeferCloseOrderPayload ¶
type DeferCloseOrderPayload struct {
OrderNo string `json:"order_no"`
}
DeferCloseOrderPayload names the unpaid order DeferCloseOrder closes.
type EventDeliverPayload ¶
type EventDeliverPayload struct {
ID int64 `json:"id"`
Topic string `json:"topic"`
Key string `json:"key"`
Payload string `json:"payload"`
}
EventDeliverPayload is the broker representation of one outbox event.
type ForthwithActivateOrderPayload ¶
type ForthwithActivateOrderPayload struct {
OrderNo string `json:"order_no"`
}
ForthwithActivateOrderPayload names the paid order to activate.
type SendEmailPayload ¶
type SendEmailPayload struct {
requestmeta.Metadata
Type string `json:"type"`
Email string `json:"to"`
Subject string `json:"subject"`
Content map[string]any `json:"content"`
}
SendEmailPayload is one email to send; Content fills the template of its Type.
type SendSmsPayload ¶
type SendSmsPayload struct {
requestmeta.Metadata
Type uint8 `json:"type"`
Telephone string `json:"telephone"`
TelephoneArea string `json:"area"`
Content string `json:"content"`
}
SendSmsPayload is one text message to send: Content is the verification code the message carries.
Source Files