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

ociqueue

package
v1.9.1 Latest Latest
Warning

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

 Go to latest Published: Nov 1, 2026 License: MIT Imports: 25 Imported by: 0

README

OCI Queue

Parity grade: B · SDK github.com/oracle/oci-go-sdk/v65@v65.126.2 · last audited 2026-10-27 (3a4fe760a)

Coverage

Metric Value
PARITY entries audited 19 (12 ok, 7 partial)
Feature families 11 (10 ok, 1 partial)
Known gaps 9
Deferred items 0
Resource leaks unknown
Known gaps
  • Consumer groups (CreateConsumerGroup, GetConsumerGroup, ListConsumerGroups, UpdateConsumerGroup, DeleteConsumerGroup) are not routed, the consumerGroupId query parameter is ignored by the data plane and the CONSUMER_GROUPS capability is rejected.
  • The LARGE_MESSAGES capability is rejected; only the standard 128 KiB message limit exists.
  • The DLQ cannot be consumed: the SDK has no API to read dead lettered messages, so they are only counted by GetStats and cleared by PurgeQueue.
  • Limits (retention 10 to 604800 s, visibility 0 to 43200 s, poll timeout 0 to 30 s, DLQ count 0 to 20, batch of 20 messages, 128 KiB per message, 512 KiB per request, 64 character channel ids) and the channelConsumptionLimit semantics are recalled from the Oracle documentation and SDK comments, not verified against the live service.
  • channelFilter is an exact match on GetMessages and a substring match on ListChannels; the live matching rules are unverified.
  • Throughput limits and per-queue storage quotas (and therefore channelConsumptionLimit enforcement) are not applied.
  • Messages are held in memory and in snapshots, so large backlogs grow the snapshot.
  • opc-retry-token idempotency, IAM policy checks, resource principals and instance principals are not applied; only API-key signatures are verified.
  • Tag values and defined-tag namespaces are stored verbatim, not validated against a tag namespace; systemTags are never set.

More

Documentation

Index

Constants

View Source 
const DefaultPort = 8156

DefaultPort is the dedicated listener's default TCP port.

View Source 
const (
	// Region is the fixed region baked into emulated OCIDs.
	Region = "us-ashburn-1"
)

Variables

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

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

View Source 
var ErrSnapshotNull = errors.New("ociqueue: null entry in snapshot")

ErrSnapshotNull is returned when a snapshot contains a null entry.

Functions

This section is empty.

Types

type APIError 

type APIError struct {
	Code    string
	Message string
	Status  int
}

APIError is an OCI error response: HTTP status plus the `code` and `message` body fields.

func (*APIError) Error 

func (e *APIError) Error() string

type BatchResultWire 

type BatchResultWire struct {
	Entries        []resultEntryWire `json:"entries"`
	ServerFailures int               `json:"serverFailures"`
	ClientFailures int               `json:"clientFailures"`
}

type ConfigProvider 

type ConfigProvider interface {
	GetOCIQueueSettings() Settings
}

ConfigProvider extracts OCI Queue configuration from the AppContext Config.

type CreateQueueIn 

type CreateQueueIn struct {
	Tags
	Retention    *int              `json:"retentionInSeconds"`
	Visibility   *int              `json:"visibilityInSeconds"`
	Timeout      *int              `json:"timeoutInSeconds"`
	ChannelLimit *int              `json:"channelConsumptionLimit"`
	DLQCount     *int              `json:"deadLetterQueueDeliveryCount"`
	KMSKeyID     *string           `json:"customEncryptionKeyId"`
	Name         string            `json:"displayName"`
	Compartment  string            `json:"compartmentId"`
	Capabilities []json.RawMessage `json:"capabilities"`
}

CreateQueueIn is the CreateQueue request body.

type GetMessageWire 

type GetMessageWire struct {
	VisibleAfter  time.Time `json:"visibleAfter"`
	ExpireAfter   time.Time `json:"expireAfter"`
	CreatedAt     time.Time `json:"createdAt"`
	Metadata      *metaWire `json:"metadata,omitempty"`
	Content       string    `json:"content"`
	Receipt       string    `json:"receipt"`
	ID            int64     `json:"id"`
	DeliveryCount int       `json:"deliveryCount"`
}

type GetParams 

type GetParams struct {
	Visibility    *int
	Timeout       *int
	Limit         *int
	ChannelFilter string
}

GetParams carries the GetMessages query parameters; nil fields fall back to the queue defaults.

type Handler 

type Handler struct {
	Backend *InMemoryBackend

	Port        int
	EnforceAuth bool
	// contains filtered or unexported fields
}

Handler serves OCI Queue: queue administration and the messages API.

func NewHandler 

func NewHandler(backend *InMemoryBackend) *Handler

NewHandler creates a Handler; callers override Port and EnforceAuth from Settings.

func (*Handler) ExtractOperation 

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

ExtractOperation returns the operation name for metrics.

func (*Handler) ExtractResource 

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

ExtractResource returns the addressed resource id for metrics.

func (*Handler) GetSupportedOperations 

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

GetSupportedOperations lists the supported operations.

func (*Handler) Handler 

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

Handler returns the Echo handler for all OCIQueue requests.

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: OCIQueue runs on its own listener.

func (*Handler) Shutdown 

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

Shutdown stops the listener and the sweeper.

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.

func (*Handler) Trusted 

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

Trusted returns a handler that skips request-signature verification, for the in-process dashboard proxy.

type InMemoryBackend 

type InMemoryBackend struct {
	Auth *ociauth.Registry
	// contains filtered or unexported fields
}

InMemoryBackend holds every queue and work request under one lock.

func NewInMemoryBackend 

func NewInMemoryBackend() *InMemoryBackend

NewInMemoryBackend creates an empty backend.

func (*InMemoryBackend) ChangeQueueCompartment 

func (b *InMemoryBackend) ChangeQueueCompartment(id, ifMatch, compartment string) (string, error)

ChangeQueueCompartment moves a queue to another compartment.

func (*InMemoryBackend) Compartments 

func (b *InMemoryBackend) Compartments() []string

Compartments lists the distinct compartments that hold live queues.

func (*InMemoryBackend) CreateQueue 

func (b *InMemoryBackend) CreateQueue(in CreateQueueIn) (string, error)

CreateQueue creates a queue in CREATING state and returns its work request.

func (*InMemoryBackend) DeleteMessage 

func (b *InMemoryBackend) DeleteMessage(queueID, receipt string) error

DeleteMessage deletes one message by receipt.

func (*InMemoryBackend) DeleteMessages 

func (b *InMemoryBackend) DeleteMessages(queueID string, in receiptsIn) (BatchResultWire, error)

DeleteMessages deletes several messages by receipt; each entry reports its own result.

func (*InMemoryBackend) DeleteQueue 

func (b *InMemoryBackend) DeleteQueue(id, ifMatch string) (string, error)

DeleteQueue moves a queue to DELETING.

func (*InMemoryBackend) GetMessages 

func (b *InMemoryBackend) GetMessages(
	ctx context.Context, stop <-chan struct{}, queueID string, p GetParams,
) ([]GetMessageWire, error)

GetMessages returns visible messages, long polling until one arrives, timeout elapses, ctx or stop ends.

func (*InMemoryBackend) GetQueue 

func (b *InMemoryBackend) GetQueue(id, endpoint string) (QueueWire, string, error)

GetQueue returns one queue and its etag.

func (*InMemoryBackend) GetStats 

func (b *InMemoryBackend) GetStats(queueID, channel string) (QueueStatsWire, error)

GetStats counts visible and in-flight messages and their size, for the queue and its dead letter queue.

func (*InMemoryBackend) GetWorkRequest 

func (b *InMemoryBackend) GetWorkRequest(id string) (WorkRequestWire, error)

GetWorkRequest returns one work request.

func (*InMemoryBackend) ListChannels 

func (b *InMemoryBackend) ListChannels(queueID, filter string, o ListOpts) (page.Page[string], error)

ListChannels lists the channels that currently hold messages.

func (*InMemoryBackend) ListQueues 

func (b *InMemoryBackend) ListQueues(
	compartment, id, name, state, endpoint string,
	o ListOpts,
) (page.Page[queueSummaryWire], error)

ListQueues lists the queues of a compartment.

func (*InMemoryBackend) ListWorkRequests 

func (b *InMemoryBackend) ListWorkRequests(
	compartment, resourceID, id string, o ListOpts,
) (page.Page[WorkRequestWire], error)

ListWorkRequests lists the work requests of a compartment, optionally for one resource.

func (*InMemoryBackend) Peek 

func (b *InMemoryBackend) Peek(queueID string, dlq bool, limit int) ([]PeekedMessage, error)

Peek returns up to limit messages of a queue or its dead letter queue without delivering them: no receipt is issued, delivery counts and visibility stay untouched.

func (*InMemoryBackend) PurgeQueue 

func (b *InMemoryBackend) PurgeQueue(id, ifMatch string, in PurgeIn) (string, error)

PurgeQueue deletes the messages of a queue, its dead letter queue, or both, optionally per channel.

func (*InMemoryBackend) PutMessages 

func (b *InMemoryBackend) PutMessages(queueID string, in putMessagesIn) ([]PutMessageWire, error)

PutMessages appends messages to 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.

func (*InMemoryBackend) SetClock 

func (b *InMemoryBackend) SetClock(now func() time.Time)

SetClock injects the time source used for lifecycle, visibility and retention.

func (*InMemoryBackend) SetKeyChecker 

func (b *InMemoryBackend) SetKeyChecker(fn KeyChecker)

SetKeyChecker wires validation of customEncryptionKeyId references against a Vault backend.

func (*InMemoryBackend) SetTimer 

func (b *InMemoryBackend) SetTimer(after func(time.Duration) <-chan time.Time)

SetTimer injects the wait used by long polls; it must return a channel that fires after d.

func (*InMemoryBackend) SetTransition 

func (b *InMemoryBackend) SetTransition(d time.Duration)

SetTransition sets how long CREATING, UPDATING and DELETING states last.

func (*InMemoryBackend) Snapshot 

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

Snapshot serialises the backend state, including registered API keys.

func (*InMemoryBackend) Sweep 

func (b *InMemoryBackend) Sweep()

Sweep applies lifecycle transitions, retention and dead lettering against the backend clock.

func (*InMemoryBackend) UpdateMessage 

func (b *InMemoryBackend) UpdateMessage(queueID, receipt string, vis *int) (UpdatedWire, error)

UpdateMessage changes the visibility of one in-flight message.

func (*InMemoryBackend) UpdateMessages 

func (b *InMemoryBackend) UpdateMessages(queueID string, in receiptsIn) (BatchResultWire, error)

UpdateMessages changes the visibility of several in-flight messages; each entry reports its own result.

func (*InMemoryBackend) UpdateQueue 

func (b *InMemoryBackend) UpdateQueue(id, ifMatch string, in UpdateQueueIn) (string, error)

UpdateQueue changes the mutable settings of a queue; it passes through UPDATING.

func (*InMemoryBackend) WorkRequestExists 

func (b *InMemoryBackend) WorkRequestExists(id string) error

WorkRequestExists reports whether a work request exists (its error list is always empty).

func (*InMemoryBackend) WorkRequestLogs 

func (b *InMemoryBackend) WorkRequestLogs(id string, o ListOpts) (page.Page[wrLogWire], error)

WorkRequestLogs returns the log entries of a work request.

type KeyChecker 

type KeyChecker func(keyID string) (state string, found bool)

KeyChecker reports the lifecycle state of a KMS key OCID and whether it exists.

type ListOpts 

type ListOpts struct {
	SortBy string
	Order  string
	Page   string
	Limit  int
}

ListOpts carries the shared list query parameters.

type Message 

type Message struct {
	Created      time.Time         `json:"created"`
	Expire       time.Time         `json:"expire"`
	VisibleAfter time.Time         `json:"visibleAfter"`
	Props        map[string]string `json:"props,omitempty"`
	Content      string            `json:"content"`
	Channel      string            `json:"channel,omitempty"`
	Receipt      string            `json:"receipt,omitempty"`
	ID           int64             `json:"id"`
	Deliveries   int               `json:"deliveries"`
}

Message is one stored queue message.

type PeekedMessage 

type PeekedMessage struct {
	CreatedAt     time.Time         `json:"createdAt"`
	VisibleAfter  time.Time         `json:"visibleAfter"`
	ExpireAfter   time.Time         `json:"expireAfter"`
	Properties    map[string]string `json:"customProperties,omitempty"`
	Content       string            `json:"content"`
	Channel       string            `json:"channelId,omitempty"`
	ID            int64             `json:"id"`
	DeliveryCount int               `json:"deliveryCount"`
	InFlight      bool              `json:"inFlight"`
}

PeekedMessage is a read-only view of one stored message; it carries no receipt.

type Provider 

type Provider struct{}

Provider implements service.Provider for OCI Queue. It runs 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 PurgeIn 

type PurgeIn struct {
	ConsumerGroupID *string  `json:"consumerGroupId"`
	PurgeType       string   `json:"purgeType"`
	ChannelIDs      []string `json:"channelIds"`
}

PurgeIn is the PurgeQueue request body.

type PutMessageWire 

type PutMessageWire struct {
	ExpireAfter time.Time `json:"expireAfter"`
	ID          int64     `json:"id"`
}

type Queue 

type Queue struct {
	TimeCreated  time.Time  `json:"timeCreated"`
	TimeUpdated  time.Time  `json:"timeUpdated"`
	TransitionAt *time.Time `json:"transitionAt,omitempty"`
	Tags         Tags       `json:"tags"`

	ID            string     `json:"id"`
	CompartmentID string     `json:"compartmentId"`
	Name          string     `json:"name"`
	State         string     `json:"state"`
	KMSKeyID      string     `json:"kmsKeyId,omitempty"`
	Etag          string     `json:"etag"`
	Messages      []*Message `json:"messages,omitempty"`
	DLQ           []*Message `json:"dlq,omitempty"`
	Retention     int        `json:"retentionInSeconds"`
	Visibility    int        `json:"visibilityInSeconds"`
	Timeout       int        `json:"timeoutInSeconds"`
	DLQCount      int        `json:"deadLetterQueueDeliveryCount"`
	ChannelLimit  int        `json:"channelConsumptionLimit"`
	NextMsgID     int64      `json:"nextMsgId"`
	Seq           int64      `json:"seq"`
	// contains filtered or unexported fields
}

Queue is a message queue with its dead letter queue.

type QueueStatsWire 

type QueueStatsWire struct {
	Queue statsWire `json:"queue"`
	DLQ   statsWire `json:"dlq"`
}

type QueueWire 

type QueueWire struct {
	TimeCreated      time.Time                 `json:"timeCreated"`
	TimeUpdated      time.Time                 `json:"timeUpdated"`
	FreeformTags     map[string]string         `json:"freeformTags,omitempty"`
	DefinedTags      map[string]map[string]any `json:"definedTags,omitempty"`
	KMSKeyID         *string                   `json:"customEncryptionKeyId,omitempty"`
	ID               string                    `json:"id"`
	CompartmentID    string                    `json:"compartmentId"`
	Name             string                    `json:"displayName"`
	LifecycleState   string                    `json:"lifecycleState"`
	MessagesEndpoint string                    `json:"messagesEndpoint"`
	Retention        int                       `json:"retentionInSeconds"`
	Visibility       int                       `json:"visibilityInSeconds"`
	Timeout          int                       `json:"timeoutInSeconds"`
	DLQCount         int                       `json:"deadLetterQueueDeliveryCount"`
	ChannelLimit     int                       `json:"channelConsumptionLimit"`
}

type Settings 

type Settings struct {
	Port int `` //nolint:lll // config struct tags are intentionally verbose
	/* 205-byte string literal not displayed */
	EnforceAuth bool `` //nolint:lll // config struct tags are intentionally verbose
	/* 221-byte string literal not displayed */
}

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

func DefaultSettings 

func DefaultSettings() Settings

DefaultSettings returns the settings used when no ConfigProvider is available.

type Tags 

type Tags struct {
	FreeformTags map[string]string         `json:"freeformTags,omitempty"`
	DefinedTags  map[string]map[string]any `json:"definedTags,omitempty"`
}

Tags are the shared freeform and defined tag sets of a resource.

type UpdateQueueIn 

type UpdateQueueIn struct {
	Tags
	Name         *string           `json:"displayName"`
	Visibility   *int              `json:"visibilityInSeconds"`
	Timeout      *int              `json:"timeoutInSeconds"`
	ChannelLimit *int              `json:"channelConsumptionLimit"`
	DLQCount     *int              `json:"deadLetterQueueDeliveryCount"`
	KMSKeyID     *string           `json:"customEncryptionKeyId"`
	Capabilities []json.RawMessage `json:"capabilities"`
}

UpdateQueueIn is the UpdateQueue request body.

type UpdatedWire 

type UpdatedWire struct {
	VisibleAfter time.Time `json:"visibleAfter"`
	ID           int64     `json:"id"`
}

type WorkRequest 

type WorkRequest struct {
	Accepted      time.Time `json:"accepted"`
	FinishAt      time.Time `json:"finishAt"`
	ID            string    `json:"id"`
	Operation     string    `json:"operation"`
	CompartmentID string    `json:"compartmentId"`
	EntityType    string    `json:"entityType"`
	EntityID      string    `json:"entityId"`
	EntityURI     string    `json:"entityUri"`
	Action        string    `json:"action"`
	Seq           int64     `json:"seq"`
}

WorkRequest records an asynchronous admin operation.

type WorkRequestWire 

type WorkRequestWire struct {
	TimeAccepted    time.Time        `json:"timeAccepted"`
	TimeStarted     *time.Time       `json:"timeStarted,omitempty"`
	TimeFinished    *time.Time       `json:"timeFinished,omitempty"`
	OperationType   string           `json:"operationType"`
	Status          string           `json:"status"`
	ID              string           `json:"id"`
	CompartmentID   string           `json:"compartmentId"`
	Resources       []wrResourceWire `json:"resources"`
	PercentComplete float32          `json:"percentComplete"`
}

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.