gcpcloudtasks

package
v1.8.1 Latest Latest
Warning

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

 Go to latest Published: Oct 20, 2026 License: MIT Imports: 41 Imported by: 0

README

Cloud Tasks

Parity grade: B · SDK cloud.google.com/go/cloudtasks@v1.20.0 · last audited 2026-10-18 (1c5323168)

Coverage

Metric Value
PARITY entries audited 20 (20 ok)
Feature families 10 (5 ok, 5 partial)
Known gaps 3
Deferred items 0
Resource leaks clean
Known gaps
  • ListQueues filter supports only state=, name= and name: terms.
  • Queue max_burst_size is an emulator-chosen derivation; real Cloud Tasks burst sizing is not documented.
  • Persisted in-flight attempts are not resumed mid-flight: after a restore the task is simply due again and its attempt count keeps the interrupted attempt.

More

Documentation

Index

Constants

View Source 
const (
	// DefaultDispatchInterval is how often due tasks are picked up.
	DefaultDispatchInterval = 100 * time.Millisecond
)
View Source 
const DefaultPort = 8123

DefaultPort is the dedicated gRPC listener's default TCP port.

Variables

View Source 
var ErrNilAppContext = errors.New("gcpcloudtasks: nil app context")

ErrNilAppContext is returned when Init is called with a nil AppContext.

Functions

This section is empty.

Types

type ConfigProvider 

type ConfigProvider interface {
	GetGCPCloudTasksSettings() Settings
}

ConfigProvider extracts Cloud Tasks configuration from the AppContext Config.

type Handler 

type Handler struct {
	Backend *InMemoryBackend

	// DispatchInterval is the dispatcher tick; zero disables the background dispatcher.
	DispatchInterval time.Duration
	Port             int
	// contains filtered or unexported fields
}

Handler serves the Cloud Tasks gRPC API on a dedicated listener.

func NewHandler 

func NewHandler(backend *InMemoryBackend) *Handler

NewHandler creates a Handler; callers override Port from Settings.

func (*Handler) API 

func (h *Handler) API() cloudtaskspb.CloudTasksServer

API exposes the CloudTasks service for in-process callers (dashboard proxy).

func (*Handler) Addr 

func (h *Handler) Addr() string

Addr returns the bound listener address, or "" before StartWorker.

func (*Handler) ExtractOperation 

func (h *Handler) ExtractOperation(*echo.Context) string

ExtractOperation is unused: requests never reach the shared router.

func (*Handler) ExtractResource 

func (h *Handler) ExtractResource(*echo.Context) string

ExtractResource is unused: requests never reach the shared router.

func (*Handler) GetSupportedOperations 

func (h *Handler) GetSupportedOperations() []string

GetSupportedOperations lists the supported gRPC methods.

func (*Handler) Handler 

func (h *Handler) Handler() echo.HandlerFunc

Handler returns a 404 handler; the real API is gRPC on the dedicated port.

func (*Handler) MatchPriority 

func (h *Handler) MatchPriority() int

MatchPriority is irrelevant because RouteMatcher never matches.

func (*Handler) Name 

func (h *Handler) Name() string

Name returns the service name.

func (*Handler) Reset 

func (h *Handler) Reset()

Reset clears all in-memory state.

func (*Handler) Restore 

func (h *Handler) Restore(ctx context.Context, data []byte) error

Restore implements persistence.Persistable by delegating to the backend.

func (*Handler) RouteMatcher 

func (h *Handler) RouteMatcher() service.Matcher

RouteMatcher never matches: Cloud Tasks runs on its own gRPC listener.

func (*Handler) Shutdown 

func (h *Handler) Shutdown(ctx context.Context)

Shutdown stops the gRPC server and dispatcher, aborting in-flight attempts.

func (*Handler) Snapshot 

func (h *Handler) Snapshot(ctx context.Context) []byte

Snapshot implements persistence.Persistable by delegating to the backend.

func (*Handler) StartWorker 

func (h *Handler) StartWorker(ctx context.Context) error

StartWorker binds the dedicated listener synchronously so a port conflict is returned to the caller.

type InMemoryBackend 

type InMemoryBackend struct {
	// contains filtered or unexported fields
}

InMemoryBackend holds queues, tasks, tombstones, IAM policies and CMEK configs under one lock.

func NewInMemoryBackend 

func NewInMemoryBackend() *InMemoryBackend

NewInMemoryBackend creates an empty backend.

func (*InMemoryBackend) BatchCreateTasks 

func (b *InMemoryBackend) BatchCreateTasks(
	req *cloudtaskspb.BatchCreateTasksRequest,
) (*longrunningpb.Operation, error)

BatchCreateTasks creates up to 100 tasks and records a finished operation listing per-task failures.

func (*InMemoryBackend) BatchDeleteTasks 

func (b *InMemoryBackend) BatchDeleteTasks(
	req *cloudtaskspb.BatchDeleteTasksRequest,
) (*longrunningpb.Operation, error)

BatchDeleteTasks deletes up to 1000 tasks and records a finished operation listing per-task failures.

func (*InMemoryBackend) CreateQueue 

func (b *InMemoryBackend) CreateQueue(req *cloudtaskspb.CreateQueueRequest) (*cloudtaskspb.Queue, error)

CreateQueue creates a queue under a location.

func (*InMemoryBackend) CreateTask 

func (b *InMemoryBackend) CreateTask(req *cloudtaskspb.CreateTaskRequest) (*cloudtaskspb.Task, error)

CreateTask adds a task to a queue.

func (*InMemoryBackend) DeleteQueue 

func (b *InMemoryBackend) DeleteQueue(name string) error

DeleteQueue removes a queue and its tasks; the name cannot be reused for 7 days.

func (*InMemoryBackend) DeleteTask 

func (b *InMemoryBackend) DeleteTask(name string) error

DeleteTask removes a task; its name stays reserved for an hour.

func (*InMemoryBackend) DispatchDue 

func (b *InMemoryBackend) DispatchDue()

DispatchDue starts every attempt that is due and allowed by queue state, rate and concurrency limits.

func (*InMemoryBackend) Drain 

func (b *InMemoryBackend) Drain()

Drain blocks until every in-flight dispatch has finished.

func (*InMemoryBackend) GetCmekConfig 

func (b *InMemoryBackend) GetCmekConfig(name string) (*cloudtaskspb.CmekConfig, error)

GetCmekConfig returns a location's CMEK config; an unconfigured location reports an empty kms_key.

func (*InMemoryBackend) GetIamPolicy 

func (b *InMemoryBackend) GetIamPolicy(resource string) (*iampb.Policy, error)

GetIamPolicy returns a queue's IAM policy, or an empty policy when none is set.

func (*InMemoryBackend) GetQueue 

func (b *InMemoryBackend) GetQueue(name string) (*cloudtaskspb.Queue, error)

GetQueue returns a queue.

func (*InMemoryBackend) GetTask 

func (b *InMemoryBackend) GetTask(req *cloudtaskspb.GetTaskRequest) (*cloudtaskspb.Task, error)

GetTask returns a task.

func (*InMemoryBackend) ListQueues 

func (b *InMemoryBackend) ListQueues(req *cloudtaskspb.ListQueuesRequest) ([]*cloudtaskspb.Queue, string, error)

ListQueues lists a location's queues sorted by name.

func (*InMemoryBackend) ListTasks 

func (b *InMemoryBackend) ListTasks(req *cloudtaskspb.ListTasksRequest) ([]*cloudtaskspb.Task, string, error)

ListTasks lists a queue's tasks ordered by schedule time, then name.

func (*InMemoryBackend) PauseQueue 

func (b *InMemoryBackend) PauseQueue(name string) (*cloudtaskspb.Queue, error)

PauseQueue stops dispatching; tasks can still be added.

func (*InMemoryBackend) PurgeQueue 

func (b *InMemoryBackend) PurgeQueue(name string) (*cloudtaskspb.Queue, error)

PurgeQueue deletes every task in a queue.

func (*InMemoryBackend) Reset 

func (b *InMemoryBackend) Reset()

Reset clears all state.

func (*InMemoryBackend) Restore 

func (b *InMemoryBackend) Restore(ctx context.Context, data []byte) error

Restore loads backend state from a snapshot; in-flight attempts at snapshot time are simply due again.

func (*InMemoryBackend) ResumeQueue 

func (b *InMemoryBackend) ResumeQueue(name string) (*cloudtaskspb.Queue, error)

ResumeQueue restarts dispatching of a paused queue.

func (*InMemoryBackend) RunDispatcher 

func (b *InMemoryBackend) RunDispatcher(ctx context.Context, every time.Duration)

RunDispatcher dispatches due tasks every interval until ctx ends.

func (*InMemoryBackend) RunTask 

func (b *InMemoryBackend) RunTask(req *cloudtaskspb.RunTaskRequest) (*cloudtaskspb.Task, error)

RunTask forces an immediate attempt regardless of queue state, schedule or rate limits.

func (*InMemoryBackend) SetAppEngineBaseURL 

func (b *InMemoryBackend) SetAppEngineBaseURL(u string)

SetAppEngineBaseURL sets where AppEngineHttpRequest tasks are dispatched; empty disables them.

func (*InMemoryBackend) SetIamPolicy 

func (b *InMemoryBackend) SetIamPolicy(req *iampb.SetIamPolicyRequest) (*iampb.Policy, error)

SetIamPolicy replaces a queue's IAM policy; a non-empty etag must match the stored policy.

func (*InMemoryBackend) SetRunContext 

func (b *InMemoryBackend) SetRunContext(ctx context.Context)

SetRunContext sets the context in-flight dispatches run under; cancel it to abort them.

func (*InMemoryBackend) Snapshot 

func (b *InMemoryBackend) Snapshot(ctx context.Context) []byte

Snapshot serialises queues with their tasks, name tombstones, IAM policies and CMEK configs.

func (*InMemoryBackend) TestIamPermissions 

func (b *InMemoryBackend) TestIamPermissions(
	req *iampb.TestIamPermissionsRequest,
) (*iampb.TestIamPermissionsResponse, error)

TestIamPermissions reports every requested permission as granted: the emulator performs no authorization.

func (*InMemoryBackend) UpdateCmekConfig 

func (b *InMemoryBackend) UpdateCmekConfig(
	req *cloudtaskspb.UpdateCmekConfigRequest,
) (*cloudtaskspb.CmekConfig, error)

UpdateCmekConfig sets a location's customer-managed encryption key; the key is recorded, not used.

func (*InMemoryBackend) UpdateQueue 

func (b *InMemoryBackend) UpdateQueue(req *cloudtaskspb.UpdateQueueRequest) (*cloudtaskspb.Queue, error)

UpdateQueue updates the masked fields of a queue, creating it when it does not exist.

type Provider 

type Provider struct{}

Provider implements service.Provider for Google Cloud Tasks. It serves gRPC on its own listener (see Handler.StartWorker) and never matches on the shared AWS router.

func (*Provider) Init 

func (p *Provider) Init(ctx *service.AppContext) (service.Registerable, error)

Init creates the backend and handler; the port is bound in StartWorker.

func (*Provider) Name 

func (p *Provider) Name() string

Name returns the service provider name.

type Settings 

type Settings struct {
	AppEngineBaseURL string `` //nolint:lll // config struct tags are intentionally verbose
	/* 208-byte string literal not displayed */
	Port int `` //nolint:lll // config struct tags are intentionally verbose
	/* 204-byte string literal not displayed */
}

Settings holds service-level configuration, embedded in the root CLI command as CLI.GCPCloudTasks.

func DefaultSettings 

func DefaultSettings() Settings

DefaultSettings returns the settings used when no ConfigProvider is available.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.