README
¶
Arkoren Queue Cloudflare
arkoren.dev/driver/cloudflare/cloudflarequeue provides a Cloudflare Queues
driver that works from any host: it sends jobs with the Queues REST API and
consumes them as an HTTP pull consumer, using the official
github.com/cloudflare/cloudflare-go/v7 SDK. It sends and receives complete
queue.Message envelopes as text messages.
Install with go get arkoren.dev/driver/cloudflare and import
"arkoren.dev/driver/cloudflare/cloudflarequeue". Provision the queues and
configure HTTP pull consumers before consuming. The
Cloudflare queue guide
covers token permissions, worker wiring, and lease limits.
New(apiToken, accountID, queues, configure...) needs an API token with Queues read
and write permission, the account ID, and a map from application queue names to
Cloudflare queue IDs. NewFrom(client, accountID, queues, configure...) wraps a
caller-owned client. Both take trailing func(*cloudflarequeue.Builder) callbacks:
driver, err := cloudflarequeue.New(token, accountID, queues, func(consumer *cloudflarequeue.Builder) {
consumer.BatchSize(5)
})
Callbacks run directly once, synchronously in order on one fresh draft before
validation. Nil entries panic; expanded nil/empty slices omit edits. Settings
return the same draft. Both constructors copy the required queue map before
callbacks, so a callback's edits to your original map cannot change routing.
Do not mutate it concurrently with capture. Published settings are detached;
retained drafts cannot change drivers. Native clients and loggers remain shared.
NewFrom ignores Endpoint and applies all other settings; Client() returns
that same pointer without transferring ownership or closing its transport.
Cloudflare answers a pull immediately, so Consume polls and backs off while a
queue is empty. It cannot extend a lease, so a handler must finish within
VisibilityTimeout. Reject leaves the message to its lease; configure a
dead-letter queue on the pull consumer to keep rejected and exhausted messages.
Defaults are one message per pull, a 30-second lease, and empty-poll waits doubling
from one second to ten seconds. A larger batch is sequential: earlier handlers
consume later messages' lease time too. Cancellation is checked before every
handler, including remaining messages in a pulled batch; their unused leases
expire for redelivery. The active handler must return cooperatively.
Positive retry delays are rounded up to seconds. Delays beyond 24 hours fail
without a vendor request or settlement; the driver never shortens the requested
lower bound. Acknowledge and Retry require exactly one matching native success
count, not merely a successful HTTP response. A warning accompanying that count
can succeed; an absent or zero matching count fails and leaves local settlement
open. Repeated or concurrent settlement returns ErrDeliverySettled.
Construction makes no requests and needs no Start/Stop hook. You own the consumer
context and native transport lifecycle. A nil logger selects slog.Default() at
construction. Handler-returned errors (including settlement failures), Fail causes
and confirmed warnings are logged as supplied; the driver does not recursively
sanitize native objects or arbitrary caller errors. Keep credentials and payloads
out of them.
See the API reference.
Documentation
¶
Overview ¶
Package cloudflarequeue provides a Cloudflare Queues-backed queue driver.
The driver sends jobs with the Queues REST API and consumes them as an HTTP pull consumer, so it works from any host, not only from Cloudflare Workers. New builds a Cloudflare API client from an API token and NewFrom wraps a caller-owned one. Both map application queue names to Cloudflare queue IDs.
Cloudflare returns pull requests immediately instead of holding them open, so Driver.Consume polls and backs off while a queue is empty. It cannot extend a lease: a handler must finish within Builder.VisibilityTimeout.
The queue behind each name needs an HTTP pull consumer. Create it with `wrangler queues consumer http add <queue>` or in the dashboard; add a dead-letter queue there to keep rejected and exhausted messages.
Both constructors invoke ordered Builder callbacks directly once on one fresh draft. Nil entries panic; expanded nil/empty slices omit edits. Publication captures routing before callbacks and copies completed settings, not clients or loggers. NewFrom ignores Endpoint and never closes the supplied client. Retained drafts cannot change a constructed driver. Both constructors perform no network I/O and need no Start/Stop hook. The caller owns Consume's context; cancellation stops remaining handlers in an already-pulled batch and leaves their leases to expire for redelivery. Defaults are one message per pull, a 30-second lease, and empty-poll waits doubling from one second to ten seconds. Lease renewal is unsupported. Retry rounds positive delays up to seconds and rejects delays above 24 hours without shortening them or settling the delivery. Acknowledge and Retry need exactly one matching native success count; warnings with that count may still succeed. Unconfirmed settlements remain open for an explicit later attempt. Nil loggers resolve slog.Default at construction. Handler-returned errors (including settlement failures), Fail causes and confirmed warnings are logged as supplied, not recursively sanitized; keep credentials and payloads out of those diagnostic values.
Example:
driver, err := cloudflarequeue.New(token, accountID, queues, func(consumer *cloudflarequeue.Builder) {
consumer.BatchSize(5)
})
if err != nil {
return err
}
_ = driver
Index ¶
- Variables
- type Builder
- func (consumer *Builder) BatchSize(count int) *Builder
- func (consumer *Builder) Endpoint(endpoint string) *Builder
- func (consumer *Builder) Logger(logger *slog.Logger) *Builder
- func (consumer *Builder) MaxPollInterval(interval time.Duration) *Builder
- func (consumer *Builder) PollInterval(interval time.Duration) *Builder
- func (consumer *Builder) VisibilityTimeout(timeout time.Duration) *Builder
- type BuilderFunc
- type Driver
Constants ¶
This section is empty.
Variables ¶
var ( // ErrUnknownQueue is returned when a queue name has no configured // Cloudflare queue ID. // Routing is fixed at construction; unknown names cause no vendor request. // // Example: // // err := driver.Dispatch(ctx, queue.Message{Queue: "not-configured", Name: "send-email"}) // if errors.Is(err, cloudflarequeue.ErrUnknownQueue) { // // Correct the application's queue routing before retrying. // } ErrUnknownQueue = errors.New("unknown job queue") // ErrInvalidConfig is returned when the driver configuration is invalid. // Constructors wrap it with the rejected setting and perform no requests. // // Example: // // _, err := cloudflarequeue.New("", "account-id", map[string]string{"emails": "queue-id"}) // if errors.Is(err, cloudflarequeue.ErrInvalidConfig) { // // Load the required API token before starting the worker. // } ErrInvalidConfig = errors.New("invalid Cloudflare Queues job driver configuration") // ErrInvalidHandler is returned when Consume is given a nil handler. // Validation happens before polling; an already canceled context takes priority. // // Example: // // err := driver.Consume(ctx, "emails", nil) // if errors.Is(err, cloudflarequeue.ErrInvalidHandler) { // // Supply a delivery handler; no message was pulled. // } ErrInvalidHandler = errors.New("invalid job delivery handler") // ErrDeliverySettled is returned when a delivery is settled more than once. // A concurrent settlement in progress also returns this error without issuing // another request. Failed native settlements do not mark the delivery settled. // // Example: // // if err := delivery.Acknowledge(ctx); err != nil { // return err // } // if err := delivery.Acknowledge(ctx); errors.Is(err, cloudflarequeue.ErrDeliverySettled) { // // This delivery was already completed; do not settle it again. // } ErrDeliverySettled = errors.New("job delivery already settled") )
Functions ¶
This section is empty.
Types ¶
type Builder ¶
type Builder struct {
// contains filtered or unexported fields
}
Builder holds optional Cloudflare polling settings during construction. Callbacks share one fresh draft; completed settings are copied. Native clients remain caller-owned and NewFrom ignores Endpoint.
Example:
configure := func(consumer *Builder) {
consumer.BatchSize(10)
}
driver, err := New("api-token", "account-id", map[string]string{"emails": "queue-id"}, configure)
if err != nil {
return err
}
_ = driver
func (*Builder) BatchSize ¶
BatchSize replaces the sequential pull batch size. Zero selects one; construction rejects values outside 1–100. It returns the same draft.
Example:
configure := func(consumer *Builder) {
consumer.BatchSize(10)
}
driver, err := New("api-token", "account-id", map[string]string{"emails": "queue-id"}, configure)
if err != nil {
return err
}
_ = driver
func (*Builder) Endpoint ¶
Endpoint replaces New's API base URL. Empty uses Cloudflare's default; NewFrom ignores it. It returns the same draft without making requests.
Example:
configure := func(consumer *Builder) {
consumer.Endpoint("http://localhost:8080/")
}
driver, err := New("api-token", "account-id", map[string]string{"emails": "queue-id"}, configure)
if err != nil {
return err
}
_ = driver
func (*Builder) Logger ¶
Logger replaces the shared delivery logger. Nil selects slog.Default at construction. It returns the same draft without mutating the logger. Handler-returned errors (including settlement failures), Fail causes and confirmed-settlement warnings are logged as supplied; contents are not recursively sanitized. Keep credentials and payloads out of application errors and apply redaction in the application's logger if its handlers require it.
Example:
configure := func(consumer *Builder) {
consumer.Logger(slog.New(slog.DiscardHandler))
}
driver, err := New("api-token", "account-id", map[string]string{"emails": "queue-id"}, configure)
if err != nil {
return err
}
_ = driver
func (*Builder) MaxPollInterval ¶
MaxPollInterval replaces the backoff cap. Zero selects ten seconds; a positive cap below PollInterval is raised to it. It returns the same draft.
Example:
configure := func(consumer *Builder) {
consumer.MaxPollInterval(5 * time.Second)
}
driver, err := New("api-token", "account-id", map[string]string{"emails": "queue-id"}, configure)
if err != nil {
return err
}
_ = driver
func (*Builder) PollInterval ¶
PollInterval replaces the initial empty-poll delay. Zero selects one second; negative values fail construction. It returns the same draft.
Example:
configure := func(consumer *Builder) {
consumer.PollInterval(time.Second)
}
driver, err := New("api-token", "account-id", map[string]string{"emails": "queue-id"}, configure)
if err != nil {
return err
}
_ = driver
func (*Builder) VisibilityTimeout ¶
VisibilityTimeout replaces the nonrenewable lease. Zero selects 30 seconds; construction accepts 1ms through 12 hours. It returns the same draft.
Example:
configure := func(consumer *Builder) {
consumer.VisibilityTimeout(time.Minute)
}
driver, err := New("api-token", "account-id", map[string]string{"emails": "queue-id"}, configure)
if err != nil {
return err
}
_ = driver
type BuilderFunc ¶
type BuilderFunc = func(*Builder)
BuilderFunc edits one fresh polling draft synchronously, once in order, before validation. Nil entries panic; expanded nil or empty slices omit edits. Routing is captured before callbacks, and retained drafts cannot change the driver. Native clients and loggers remain shared, not recursively copied or sanitized.
Example:
configure := func(consumer *cloudflarequeue.Builder) {
consumer.BatchSize(5)
}
policies := []cloudflarequeue.BuilderFunc{configure}
driver, err := cloudflarequeue.New(token, accountID, routes, policies...)
if err != nil {
return err
}
_ = driver
type Driver ¶
type Driver struct {
// contains filtered or unexported fields
}
Driver implements queue.DispatcherDriver and queue.ConsumerDriver using Cloudflare Queues over its REST API: messages are pushed with the send endpoint and consumed as an HTTP pull consumer.
Delivery is at least once. A message that is not settled before its lease ends is delivered again; see Builder.VisibilityTimeout. Construction captures routing before callbacks. Native clients and loggers are shared dependencies: configure them before use. The driver opens no background worker, owns no closeable connection, and has no Start or Stop method; callers own Consume's context and native transport lifecycle. Do not copy Driver values.
Example:
driver, err := cloudflarequeue.New(token, accountID, map[string]string{"emails": "queue-id"})
if err != nil {
return err
}
consumer := queue.NewConsumer(driver)
_ = consumer // Register jobs and run it with the application's worker context.
func New ¶
func New(apiToken, accountID string, queueIDs map[string]string, configure ...BuilderFunc) (*Driver, error)
New creates a driver that authenticates with apiToken, an API token with Queues read and write permissions.
accountID is the Cloudflare account ID. queues maps application queue names to Cloudflare queue IDs; every queue to be consumed needs an HTTP pull consumer. All three are required: an empty token, account ID, or queue map, or an entry with a blank name or ID, returns an error wrapping ErrInvalidConfig. The map is copied before callbacks, so caller edits made during a callback or afterwards do not affect routing. Callers must not mutate it concurrently with input capture. configure callbacks run directly once, synchronously in order on one fresh draft. Nil entries panic; expanded nil/empty slices supply no callbacks. Retained drafts cannot change the driver. New makes no request.
Example:
driver, err := cloudflarequeue.New(os.Getenv("CLOUDFLARE_API_TOKEN"),
os.Getenv("CLOUDFLARE_ACCOUNT_ID"),
map[string]string{"emails": "6b8f3a0c2d1e4f5a8b7c9d0e1f2a3b4c"},
func(consumer *cloudflarequeue.Builder) {
consumer.BatchSize(10)
},
)
if err != nil {
return err
}
_ = driver
func NewFrom ¶
func NewFrom(client *cloudflare.Client, accountID string, queueIDs map[string]string, configure ...BuilderFunc) (*Driver, error)
NewFrom creates a driver that calls the Cloudflare API through client. The caller retains ownership of client and its credentials.
accountID and queues are required and validated and copied exactly as in New; a nil client also returns an error wrapping ErrInvalidConfig. All Builder settings apply except Builder.Endpoint. Callbacks follow New's ordered once-only policy, including nil-entry panics before validation. Routing is captured before those callbacks. Loggers and the native client stay shared; the driver neither closes nor recursively sanitizes their contents.
Example:
client := cloudflare.NewClient(option.WithAPIToken(token))
driver, err := cloudflarequeue.NewFrom(client, accountID, map[string]string{
"emails": "6b8f3a0c2d1e4f5a8b7c9d0e1f2a3b4c",
})
if err != nil {
return err
}
_ = driver
func (*Driver) Client ¶
func (driver *Driver) Client() *cloudflare.Client
Client returns the underlying Cloudflare API client. It is the same pointer supplied to NewFrom; neither method transfers ownership or closes its transports. Configure it before concurrent use. Native calls bypass the driver's routing and settlement checks, and native diagnostics are not sanitized by this package.
Example:
native := cloudflare.NewClient(option.WithAPIToken(token))
driver, err := cloudflarequeue.NewFrom(native, accountID, routes)
if err != nil {
return err
}
_ = driver.Client() == native // Identity and ownership are preserved.
func (*Driver) Consume ¶
func (driver *Driver) Consume(ctx context.Context, queueName string, handler queue.DeliveryHandler) error
Consume pulls messages from the named queue and passes each to handler until ctx is canceled or a pull fails. Cloudflare answers a pull at once even when the queue is empty, so Consume waits Builder.PollInterval, doubling up to Builder.MaxPollInterval, between empty pulls.
Handlers run one after another within a batch. A handler error is logged and leaves the delivery unsettled: Cloudflare delivers it again when its lease ends. Cancellation returns the context error. A failed pull returns an error after the Cloudflare client has exhausted its own retries; the caller (for example queue.Consumer.Run) decides whether to start Consume again. Cancellation is checked before every handler, including remaining messages in an already-pulled batch; unused leases expire normally. No background handler goroutines are started: a running synchronous handler must return cooperatively. Native pull errors and context identity remain inspectable.
Example:
return driver.Consume(ctx, "emails", func(ctx context.Context, delivery queue.Delivery) error {
message := delivery.Message()
_ = message // Process it before acknowledging.
return delivery.Acknowledge(ctx)
})
func (*Driver) Dispatch ¶
Dispatch JSON-encodes message and pushes it as a text message to the Cloudflare queue configured for message.Queue. The envelope carries Metadata, so it needs no Cloudflare-specific header. It returns an error wrapping ErrUnknownQueue when the queue has no configured ID and the context error when ctx is already canceled.
Cloudflare limits a message to 128 KB including the envelope, whose Payload is base64 encoded. Native request errors retain their original causes; this is not a delivery receipt or proof that a consumer processed the message.
Example:
message := queue.Message{Queue: "emails", Name: "send-email", Payload: []byte(`{"user_id":42}`)}
if err := driver.Dispatch(ctx, message); err != nil {
return err
}
func (*Driver) MessagingSystem ¶
MessagingSystem returns "cloudflare_queues", the messaging.system telemetry value that the queue facades report for this driver.
Example:
var driver cloudflarequeue.Driver name := driver.MessagingSystem() _ = name
Source Files
Directories