Help us shape the future of Go! Take the 2026 Go Developer Survey

taskqueue

package
v1.20.3 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Sep 19, 2026 License: GPL-3.0 Imports: 12 Imported by: 0

Documentation

Overview

Package taskqueue carries OpenTelemetry trace context across the asynq boundary. 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

View Source
const (
	EmailTypeVerify        = "verify"
	EmailTypeMaintenance   = "maintenance"
	EmailTypeExpiration    = "expiration"
	EmailTypeTrafficExceed = "traffic_exceed"
	EmailTypeCustom        = "custom"
)
View Source
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"
)
View Source
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"
	SchedulerTotalServerData      = "scheduler:total:server"
	SchedulerResetTraffic         = "scheduler:reset:traffic"
	SchedulerTrafficStat          = "scheduler:traffic:stat"
	SchedulerLogCleanup           = "scheduler:log:cleanup"
	SchedulerFlushTraffic         = "scheduler:flush:traffic"
)
View Source
const (
	// ScheduledBatchSendEmail scheduled batch send email
	ScheduledBatchSendEmail = "scheduled:email:batch"

	// ForthwithQuotaTask create quota task immediately
	ForthwithQuotaTask = "forthwith:quota:task"

	// SchedulerExchangeRate fetch exchange rate task
	SchedulerExchangeRate = "scheduler:exchange:rate"
)
View Source
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"
)
View Source
const (
	// ForthwithSendEmail forthwith send email
	ForthwithSendEmail = "forthwith:email:send"
)
View Source
const (
	// ForthwithSendSms forthwith send email
	ForthwithSendSms = "forthwith:sms:send"
)
View Source
const ForthwithTrafficStatistics = "forthwith:traffic:statistics"

Variables

This section is empty.

Functions

func ActivationTaskID

func ActivationTaskID(orderNo string) string

func EventTaskID

func EventTaskID(id int64) string

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.

func Wrap

func Wrap(ctx context.Context, task *asynq.Task) *asynq.Task

Wrap returns a task whose payload carries ctx's trace context; without an active span (or with a payload the envelope cannot hold) it returns the task unchanged.

Types

type Client

type Client struct {
	*asynq.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).

func NewClient

func NewClient(client *asynq.Client) *Client

func (*Client) EnqueueContext

func (c *Client) EnqueueContext(ctx context.Context, task *asynq.Task, opts ...asynq.Option) (*asynq.TaskInfo, error)

type DeferCloseOrderPayload

type DeferCloseOrderPayload struct {
	OrderNo string `json:"order_no"`
}

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"`
}

type NodeStatus

type NodeStatus struct {
	OnlineUsers []OnlineUser `json:"online_users"`
	Status      ServerStatus `json:"status"`
	LastAt      int64        `json:"last_at"`
}

type OnlineUser

type OnlineUser struct {
	UID int64  `json:"uid"`
	IP  string `json:"ip"`
}

type SendEmailPayload

type SendEmailPayload struct {
	requestmeta.Metadata
	Type    string                 `json:"type"`
	Email   string                 `json:"to"`
	Subject string                 `json:"subject"`
	Content map[string]interface{} `json:"content"`
}

type SendSmsPayload

type SendSmsPayload struct {
	requestmeta.Metadata
	Type          uint8  `json:"type"`
	Telephone     string `json:"telephone"`
	TelephoneArea string `json:"area"`
	Content       string `json:"content"`
}

type ServerStatus

type ServerStatus struct {
	Cpu       float64 `json:"cpu"`
	Mem       float64 `json:"mem"`
	Disk      float64 `json:"disk"`
	UpdatedAt int64   `json:"updated_at"`
}

type ServerTrafficCount

type ServerTrafficCount struct {
	ServerId  int64  `json:"server_id"`
	Name      string `json:"name"`
	Today     int64  `json:"today"`
	Yesterday int64  `json:"yesterday"`
}

type TrafficStatistics

type TrafficStatistics struct {
	ServerId int64         `json:"server_id"`
	Protocol string        `json:"protocol"`
	Logs     []UserTraffic `json:"logs"`
}

type UserTraffic

type UserTraffic struct {
	SID      int64 `json:"uid"`
	Upload   int64 `json:"upload"`
	Download int64 `json:"download"`
}

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL