queue

package
v2.0.8 Latest Latest
Warning

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

 Go to latest Published: Jul 22, 2026 License: Apache-2.0 Imports: 29 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var (
	// ErrNotRetryable indicates permanent business failure. Maps to HTTP 422.
	// Cloud Tasks still retries non-2xx until maxAttempts — ops must configure DLQ.
	ErrNotRetryable = errors.New("queue: not retryable")

	// ErrDecode indicates malformed inbound payload / protocol decode failure. Maps to 400.
	ErrDecode = errors.New("queue: decode failed")

	// ErrTooLarge indicates the request body exceeded the configured limit. Maps to 413.
	ErrTooLarge = errors.New("queue: body too large")

	// ErrUnauthorized indicates missing or invalid credentials. Maps to 401.
	ErrUnauthorized = errors.New("queue: unauthorized")

	// ErrForbidden indicates valid credentials that fail audience/issuer checks. Maps to 403.
	ErrForbidden = errors.New("queue: forbidden")
)

Sentinel errors used by push HTTP status mapping and handlers.

Functions

func CloudEventsRequestURL added in v2.0.6 

func CloudEventsRequestURL(raw string) (string, url.Values, error)

CloudEventsRequestURL strips ce+http(s) → http(s) and returns Frame config query params. Config query keys (type, source, …) are not forwarded on the wire URL.

Types

type CloudTasksTarget added in v2.0.6 

type CloudTasksTarget struct {
	Project       string
	Location      string
	Queue         string
	ResourceName  string // projects/p/locations/l/queues/q (no leading slash)
	TargetURL     string
	OIDCSA        string
	OIDCAudience  string
	ScheduleDelay time.Duration
	TaskID        string
	// DeprecatedHostForm is true when the URL used host=projects (ambiguous form).
	DeprecatedHostForm bool
}

CloudTasksTarget is a parsed cloudtasks:// publisher URL.

func ParseCloudTasksURL added in v2.0.6 

func ParseCloudTasksURL(raw string) (*CloudTasksTarget, error)

ParseCloudTasksURL parses the canonical and accepted legacy Cloud Tasks URL forms.

Canonical: 

cloudtasks:///projects/{p}/locations/{l}/queues/{q}?url=...

func (*CloudTasksTarget) CreateTaskAPIURL added in v2.0.6 

func (t *CloudTasksTarget) CreateTaskAPIURL() string

CreateTaskAPIURL returns the Cloud Tasks v2 CreateTask REST endpoint.

type DeliveryMode added in v2.0.6 

type DeliveryMode int

DeliveryMode classifies how a subscriber receives messages. 

const (
	// DeliveryModePull opens a gocloud subscription and runs a receive loop.
	DeliveryModePull DeliveryMode = iota
	// DeliveryModePush receives messages via the HTTP multiplexing handler.
	DeliveryModePush
)

func ClassifySubscriberURL added in v2.0.6 

func ClassifySubscriberURL(queueURL string) (DeliveryMode, url.Values, error)

ClassifySubscriberURL parses queueURL and returns DeliveryMode. Uses net/url.Parse — not strings.HasPrefix.

func (DeliveryMode) String added in v2.0.6 

func (m DeliveryMode) String() string

type Inspector 

type Inspector interface {
	ListPublishers() []PublisherInfo
	ListSubscribers() []SubscriberInfo
}

type Manager 

type Manager interface {
	AddPublisher(ctx context.Context, reference string, queueURL string) error
	GetPublisher(reference string) (Publisher, error)
	DiscardPublisher(ctx context.Context, reference string) error

	AddSubscriber(
		ctx context.Context,
		reference string,
		queueURL string,
		handlers ...SubscribeWorker,
	) error
	DiscardSubscriber(ctx context.Context, reference string) error
	GetSubscriber(reference string) (Subscriber, error)

	Publish(ctx context.Context, reference string, payload any, headers ...map[string]string) error
	Init(ctx context.Context) error
	Close(ctx context.Context) error
}

func NewQueueManager 

func NewQueueManager(_ context.Context, workPool workerpool.Manager, opts ...ManagerOption) Manager

type ManagerOption added in v2.0.6 

type ManagerOption func(*queueManager)

ManagerOption configures the queue manager.

func WithHTTPClient added in v2.0.6 

func WithHTTPClient(c *http.Client) ManagerOption

WithHTTPClient sets the shared HTTP client for CE / Cloud Tasks publishers.

func WithServiceName added in v2.0.6 

func WithServiceName(name string) ManagerOption

WithServiceName sets the default CE source host component.

func WithTokenSource added in v2.0.6 

func WithTokenSource(ts oauth2.TokenSource) ManagerOption

WithTokenSource sets the OAuth2 token source for Cloud Tasks CreateTask API calls.

type PublishKind added in v2.0.6 

type PublishKind int

PublishKind classifies how a publisher sends messages. 

const (
	// PublishKindGoCloud uses gocloud OpenTopic (mem://, nats://).
	PublishKindGoCloud PublishKind = iota
	// PublishKindCloudEventsHTTP POSTs CloudEvents binary mode over HTTP.
	PublishKindCloudEventsHTTP
	// PublishKindCloudTasks creates Cloud Tasks via the REST API.
	PublishKindCloudTasks
)

func ClassifyPublisherURL added in v2.0.6 

func ClassifyPublisherURL(queueURL string) (PublishKind, error)

ClassifyPublisherURL parses queueURL and returns PublishKind.

func (PublishKind) String added in v2.0.6 

func (k PublishKind) String() string

type Publisher 

type Publisher interface {
	Initiated() bool
	Ref() string
	Init(ctx context.Context) error

	Publish(ctx context.Context, payload any, headers ...map[string]string) error
	Stop(ctx context.Context) error
	As(i any) bool
}

type PublisherInfo 

type PublisherInfo struct {
	Reference string `json:"reference"`
	URL       string `json:"url"`
	Initiated bool   `json:"initiated"`
}

type PushLookup added in v2.0.6 

type PushLookup interface {
	// HasPushSubscribers reports whether any registered subscriber is push mode.
	HasPushSubscribers() bool
	// LookupPush returns a push target by registration reference.
	LookupPush(ref string) (PushTarget, bool)
	GetSubscriber(reference string) (Subscriber, error)
}

PushLookup is implemented by the concrete queue manager for HTTP push demux. Kept separate from Manager so application test doubles of Manager stay compatible.

type PushTarget added in v2.0.6 

type PushTarget interface {
	Ref() string
	Mode() DeliveryMode
	ProcessPush(ctx context.Context, metadata map[string]string, body []byte) error
}

PushTarget is the narrow surface the HTTP multiplexing handler uses.

type SubscribeWorker 

type SubscribeWorker interface {
	Handle(ctx context.Context, metadata map[string]string, message []byte) error
}

type Subscriber 

type Subscriber interface {
	Ref() string

	URI() string
	Initiated() bool
	State() SubscriberState
	Metrics() SubscriberMetrics
	IsIdle() bool
	// Mode reports pull vs push delivery.
	Mode() DeliveryMode

	Init(ctx context.Context) error
	Receive(ctx context.Context) (*pubsub.Message, error)
	Stop(ctx context.Context) error
	As(i any) bool

	// ProcessPush delivers one push message synchronously (push mode only).
	ProcessPush(ctx context.Context, metadata map[string]string, body []byte) error
}

type SubscriberInfo 

type SubscriberInfo struct {
	Reference string          `json:"reference"`
	URL       string          `json:"url"`
	State     SubscriberState `json:"state"`
	Initiated bool            `json:"initiated"`
	Mode      DeliveryMode    `json:"mode"`
}

type SubscriberMetrics 

type SubscriberMetrics interface {
	IsIdle(state SubscriberState) bool
	IdleTime(state SubscriberState) time.Duration
	AverageProcessingTime() time.Duration
}

type SubscriberState 

type SubscriberState int32
const (
	SubscriberStateWaiting SubscriberState = iota
	SubscriberStateProcessing
	SubscriberStateInError
)

Source Files

View all Source files

Directories

Path Synopsis

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.