Documentation
¶
Index ¶
- Constants
- Variables
- type APIError
- type ConfigProvider
- type ConfirmationWire
- type CreateSubscriptionIn
- type CreateTopicIn
- type Delivery
- type DeliveryView
- type FunctionInvoker
- type Handler
- func (h *Handler) ExtractOperation(c *echo.Context) string
- func (h *Handler) ExtractResource(c *echo.Context) string
- func (h *Handler) GetSupportedOperations() []string
- func (h *Handler) Handler() echo.HandlerFunc
- func (h *Handler) MatchPriority() int
- func (h *Handler) Name() string
- func (h *Handler) Reset()
- func (h *Handler) Restore(ctx context.Context, data []byte) error
- func (h *Handler) RouteMatcher() service.Matcher
- func (h *Handler) Shutdown(ctx context.Context)
- func (h *Handler) Snapshot(ctx context.Context) []byte
- func (h *Handler) StartWorker(ctx context.Context) error
- func (h *Handler) Trusted() echo.HandlerFunc
- type InMemoryBackend
- func (b *InMemoryBackend) ChangeSubscriptionCompartment(id, ifMatch, compartment string) error
- func (b *InMemoryBackend) ChangeTopicCompartment(id, ifMatch, compartment string) error
- func (b *InMemoryBackend) Compartments() []string
- func (b *InMemoryBackend) ConfirmSubscription(id, token, protocol string) (ConfirmationWire, error)
- func (b *InMemoryBackend) CreateSubscription(in CreateSubscriptionIn, baseURL string) (SubWire, error)
- func (b *InMemoryBackend) CreateTopic(in CreateTopicIn, endpoint string) (TopicWire, error)
- func (b *InMemoryBackend) DeleteSubscription(id, ifMatch string) error
- func (b *InMemoryBackend) DeleteTopic(id, ifMatch string) error
- func (b *InMemoryBackend) DeliverDue(ctx context.Context, spawn func(func()))
- func (b *InMemoryBackend) Deliveries(topicID, subscriptionID string) []DeliveryView
- func (b *InMemoryBackend) GetSubscription(id string) (SubWire, error)
- func (b *InMemoryBackend) GetTopic(id, endpoint string) (TopicWire, error)
- func (b *InMemoryBackend) Kick()
- func (b *InMemoryBackend) ListSubscriptions(compartment, topicID string, o ListOpts) (page.Page[subSummaryWire], error)
- func (b *InMemoryBackend) ListTopics(compartment, id, name, state, endpoint string, o ListOpts) (page.Page[TopicWire], error)
- func (b *InMemoryBackend) NextDue() (time.Duration, func(time.Duration) <-chan time.Time)
- func (b *InMemoryBackend) PublishMessage(topicID, messageType string, in PublishIn) (PublishWire, error)
- func (b *InMemoryBackend) ResendConfirmation(id string) (SubWire, string, error)
- func (b *InMemoryBackend) Reset()
- func (b *InMemoryBackend) Restore(ctx context.Context, data []byte) error
- func (b *InMemoryBackend) SetClock(now func() time.Time)
- func (b *InMemoryBackend) SetFunctionInvoker(fn FunctionInvoker)
- func (b *InMemoryBackend) SetTimer(after func(time.Duration) <-chan time.Time)
- func (b *InMemoryBackend) SetTransition(d time.Duration)
- func (b *InMemoryBackend) Snapshot(ctx context.Context) []byte
- func (b *InMemoryBackend) Sweep()
- func (b *InMemoryBackend) TopicActive(id string) bool
- func (b *InMemoryBackend) Unsubscribe(id, token, protocol string) (string, error)
- func (b *InMemoryBackend) UpdateSubscription(id, ifMatch string, in UpdateSubscriptionIn) (UpdateSubscriptionIn, string, error)
- func (b *InMemoryBackend) UpdateTopic(id, ifMatch, endpoint string, in UpdateTopicIn) (TopicWire, error)
- type ListOpts
- type Provider
- type PublishIn
- type PublishWire
- type Settings
- type SubWire
- type Subscription
- type Tags
- type Topic
- type TopicWire
- type UpdateSubscriptionIn
- type UpdateTopicIn
Constants ¶
const DefaultPort = 8157
DefaultPort is the dedicated listener's default TCP port.
const (
// Region is the fixed region baked into emulated OCIDs.
Region = "us-ashburn-1"
)
Variables ¶
var ErrEndpointStatus = errors.New("endpoint answered")
ErrEndpointStatus marks a delivery attempt that the endpoint answered with a non-2xx status.
var ErrNilAppContext = errors.New("ocinotifications: nil app context")
ErrNilAppContext is returned when Init is called with a nil AppContext.
var ErrSnapshotNull = errors.New("ocinotifications: null entry in snapshot")
ErrSnapshotNull is returned when a snapshot contains a null entry.
Functions ¶
This section is empty.
Types ¶
type APIError ¶
APIError is an OCI error response: HTTP status plus the `code` and `message` body fields.
type ConfigProvider ¶
type ConfigProvider interface {
GetOCINotificationsSettings() Settings
}
ConfigProvider extracts OCI Notifications configuration from the AppContext Config.
type ConfirmationWire ¶
type CreateSubscriptionIn ¶
type CreateSubscriptionIn struct {
Tags
Metadata *string `json:"metadata"`
TopicID string `json:"topicId"`
Compartment string `json:"compartmentId"`
Protocol string `json:"protocol"`
Endpoint string `json:"endpoint"`
}
CreateSubscriptionIn is the CreateSubscription request body.
type CreateTopicIn ¶
type CreateTopicIn struct {
Tags
Description *string `json:"description"`
Name string `json:"name"`
Compartment string `json:"compartmentId"`
}
CreateTopicIn is the CreateTopic request body.
type Delivery ¶
type Delivery struct {
Published time.Time `json:"published"`
NextAttempt time.Time `json:"nextAttempt"`
Deadline time.Time `json:"deadline"`
Completed time.Time `json:"completed"`
SubscriptionID string `json:"subscriptionId"`
Title string `json:"title,omitempty"`
Kind string `json:"kind"`
MessageID string `json:"messageId"`
TopicID string `json:"topicId"`
TopicName string `json:"topicName"`
LastError string `json:"lastError,omitempty"`
Protocol string `json:"protocol"`
Endpoint string `json:"endpoint"`
ID string `json:"id"`
Body string `json:"body"`
MessageType string `json:"messageType"`
ConfirmationURL string `json:"confirmationUrl,omitempty"`
UnsubscribeURL string `json:"unsubscribeUrl,omitempty"`
Status string `json:"status"`
Attempts int `json:"attempts"`
LastStatus int `json:"lastStatus,omitempty"`
Seq int64 `json:"seq"`
// contains filtered or unexported fields
}
Delivery records one message or confirmation handed to a subscription endpoint.
type DeliveryView ¶
type DeliveryView = Delivery
DeliveryView is a delivery record as reported by the admin endpoint.
type FunctionInvoker ¶
type FunctionInvoker func(ctx context.Context, functionID string, header http.Header, body []byte) error
FunctionInvoker invokes an OCI Function by OCID with the message as payload.
type Handler ¶
type Handler struct {
Backend *InMemoryBackend
Port int
EnforceAuth bool
// contains filtered or unexported fields
}
Handler serves OCI Notifications: topics, subscriptions and message publishing.
func NewHandler ¶
func NewHandler(backend *InMemoryBackend) *Handler
NewHandler creates a Handler; callers override Port and EnforceAuth from Settings.
func (*Handler) ExtractOperation ¶
ExtractOperation returns the operation name for metrics.
func (*Handler) ExtractResource ¶
ExtractResource returns the addressed resource id for metrics.
func (*Handler) GetSupportedOperations ¶
GetSupportedOperations lists the supported operations.
func (*Handler) Handler ¶
func (h *Handler) Handler() echo.HandlerFunc
Handler returns the Echo handler for all OCINotifications requests.
func (*Handler) MatchPriority ¶
MatchPriority is irrelevant because RouteMatcher never matches.
func (*Handler) RouteMatcher ¶
RouteMatcher never matches: OCINotifications runs on its own listener.
func (*Handler) Snapshot ¶
Snapshot implements persistence.Persistable by delegating to the backend.
func (*Handler) StartWorker ¶
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 ¶
InMemoryBackend holds every topic, subscription and delivery under one lock.
func NewInMemoryBackend ¶
func NewInMemoryBackend() *InMemoryBackend
NewInMemoryBackend creates an empty backend.
func (*InMemoryBackend) ChangeSubscriptionCompartment ¶
func (b *InMemoryBackend) ChangeSubscriptionCompartment(id, ifMatch, compartment string) error
ChangeSubscriptionCompartment moves a subscription to another compartment.
func (*InMemoryBackend) ChangeTopicCompartment ¶
func (b *InMemoryBackend) ChangeTopicCompartment(id, ifMatch, compartment string) error
ChangeTopicCompartment moves a topic to another compartment.
func (*InMemoryBackend) Compartments ¶
func (b *InMemoryBackend) Compartments() []string
Compartments lists the distinct compartments that hold live topics or subscriptions.
func (*InMemoryBackend) ConfirmSubscription ¶
func (b *InMemoryBackend) ConfirmSubscription(id, token, protocol string) (ConfirmationWire, error)
ConfirmSubscription activates a PENDING subscription from the token sent to its endpoint.
func (*InMemoryBackend) CreateSubscription ¶
func (b *InMemoryBackend) CreateSubscription(in CreateSubscriptionIn, baseURL string) (SubWire, error)
CreateSubscription subscribes an endpoint to a topic; protocols that need it start PENDING and are sent a confirmation.
func (*InMemoryBackend) CreateTopic ¶
func (b *InMemoryBackend) CreateTopic(in CreateTopicIn, endpoint string) (TopicWire, error)
CreateTopic creates a topic in CREATING state.
func (*InMemoryBackend) DeleteSubscription ¶
func (b *InMemoryBackend) DeleteSubscription(id, ifMatch string) error
DeleteSubscription removes a subscription and its deliveries.
func (*InMemoryBackend) DeleteTopic ¶
func (b *InMemoryBackend) DeleteTopic(id, ifMatch string) error
DeleteTopic moves a topic to DELETING; its subscriptions go with it once the transition completes.
func (*InMemoryBackend) DeliverDue ¶
func (b *InMemoryBackend) DeliverDue(ctx context.Context, spawn func(func()))
DeliverDue starts an attempt for every pending delivery whose time has come, using spawn to run each one.
func (*InMemoryBackend) Deliveries ¶
func (b *InMemoryBackend) Deliveries(topicID, subscriptionID string) []DeliveryView
Deliveries lists delivery records, optionally filtered by topic and subscription, oldest first.
func (*InMemoryBackend) GetSubscription ¶
func (b *InMemoryBackend) GetSubscription(id string) (SubWire, error)
GetSubscription returns one subscription.
func (*InMemoryBackend) GetTopic ¶
func (b *InMemoryBackend) GetTopic(id, endpoint string) (TopicWire, error)
GetTopic returns one topic.
func (*InMemoryBackend) Kick ¶
func (b *InMemoryBackend) Kick()
Kick wakes the dispatcher so it re-evaluates which deliveries are due.
func (*InMemoryBackend) ListSubscriptions ¶
func (b *InMemoryBackend) ListSubscriptions( compartment, topicID string, o ListOpts, ) (page.Page[subSummaryWire], error)
ListSubscriptions lists the subscriptions of a compartment, optionally of one topic.
func (*InMemoryBackend) ListTopics ¶
func (b *InMemoryBackend) ListTopics( compartment, id, name, state, endpoint string, o ListOpts, ) (page.Page[TopicWire], error)
ListTopics lists the topics of a compartment.
func (*InMemoryBackend) NextDue ¶
NextDue reports how long until the earliest pending delivery is due; negative when none is waiting.
func (*InMemoryBackend) PublishMessage ¶
func (b *InMemoryBackend) PublishMessage(topicID, messageType string, in PublishIn) (PublishWire, error)
PublishMessage fans a message out to every ACTIVE subscription of a topic.
func (*InMemoryBackend) ResendConfirmation ¶
func (b *InMemoryBackend) ResendConfirmation(id string) (SubWire, string, error)
ResendConfirmation sends a PENDING subscription's confirmation again.
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 transitions and delivery retries.
func (*InMemoryBackend) SetFunctionInvoker ¶
func (b *InMemoryBackend) SetFunctionInvoker(fn FunctionInvoker)
SetFunctionInvoker wires ORACLE_FUNCTIONS subscriptions to the OCI Functions backend.
func (*InMemoryBackend) SetTimer ¶
func (b *InMemoryBackend) SetTimer(after func(time.Duration) <-chan time.Time)
SetTimer injects the wait the dispatcher uses until the next retry; it must return a channel firing after d.
func (*InMemoryBackend) SetTransition ¶
func (b *InMemoryBackend) SetTransition(d time.Duration)
SetTransition sets how long CREATING and DELETING topic 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 topic lifecycle transitions against the backend clock; deleted topics take their subscriptions along.
func (*InMemoryBackend) TopicActive ¶
func (b *InMemoryBackend) TopicActive(id string) bool
TopicActive reports whether id is an ACTIVE topic.
func (*InMemoryBackend) Unsubscribe ¶
func (b *InMemoryBackend) Unsubscribe(id, token, protocol string) (string, error)
Unsubscribe removes a subscription from the token embedded in its unsubscribe link.
func (*InMemoryBackend) UpdateSubscription ¶
func (b *InMemoryBackend) UpdateSubscription( id, ifMatch string, in UpdateSubscriptionIn, ) (UpdateSubscriptionIn, string, error)
UpdateSubscription changes a subscription's delivery policy and tags.
func (*InMemoryBackend) UpdateTopic ¶
func (b *InMemoryBackend) UpdateTopic(id, ifMatch, endpoint string, in UpdateTopicIn) (TopicWire, error)
UpdateTopic changes a topic's description and tags.
type Provider ¶
type Provider struct{}
Provider implements service.Provider for OCI Notifications. 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.
type PublishWire ¶
type Settings ¶
type Settings struct {
Port int `` //nolint:lll // config struct tags are intentionally verbose
/* 230-byte string literal not displayed */
EnforceAuth bool `` //nolint:lll // config struct tags are intentionally verbose
/* 229-byte string literal not displayed */
}
Settings holds service-level configuration, embedded in the root CLI as CLI.OCINotifications.
func DefaultSettings ¶
func DefaultSettings() Settings
DefaultSettings returns the settings used when no ConfigProvider is available.
type SubWire ¶
type SubWire struct {
FreeformTags map[string]string `json:"freeformTags,omitempty"`
DefinedTags map[string]map[string]any `json:"definedTags,omitempty"`
ID string `json:"id"`
TopicID string `json:"topicId"`
Protocol string `json:"protocol"`
Endpoint string `json:"endpoint"`
LifecycleState string `json:"lifecycleState"`
CompartmentID string `json:"compartmentId"`
DeliverPolicy string `json:"deliverPolicy"`
Etag string `json:"etag"`
CreatedTime int64 `json:"createdTime"`
}
type Subscription ¶
type Subscription struct {
TimeCreated time.Time `json:"timeCreated"`
Tags Tags `json:"tags"`
ID string `json:"id"`
TopicID string `json:"topicId"`
CompartmentID string `json:"compartmentId"`
Protocol string `json:"protocol"`
Endpoint string `json:"endpoint"`
State string `json:"state"`
Metadata string `json:"metadata,omitempty"`
BaseURL string `json:"baseUrl"`
ConfirmToken string `json:"confirmToken"`
UnsubToken string `json:"unsubToken"`
Etag string `json:"etag"`
MaxRetryMs int `json:"maxRetryMs"`
Seq int64 `json:"seq"`
}
Subscription is one endpoint subscribed to a topic.
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 Topic ¶
type Topic struct {
TimeCreated time.Time `json:"timeCreated"`
TransitionAt *time.Time `json:"transitionAt,omitempty"`
Tags Tags `json:"tags"`
ID string `json:"id"`
ShortID string `json:"shortId"`
CompartmentID string `json:"compartmentId"`
Name string `json:"name"`
Description string `json:"description"`
State string `json:"state"`
Etag string `json:"etag"`
Seq int64 `json:"seq"`
}
Topic is a notification topic.
type TopicWire ¶
type TopicWire struct {
TimeCreated time.Time `json:"timeCreated"`
FreeformTags map[string]string `json:"freeformTags,omitempty"`
DefinedTags map[string]map[string]any `json:"definedTags,omitempty"`
Name string `json:"name"`
TopicID string `json:"topicId"`
CompartmentID string `json:"compartmentId"`
LifecycleState string `json:"lifecycleState"`
APIEndpoint string `json:"apiEndpoint"`
ShortTopicID string `json:"shortTopicId"`
Description string `json:"description,omitempty"`
Etag string `json:"etag"`
}
type UpdateSubscriptionIn ¶
type UpdateSubscriptionIn struct {
Tags
DeliveryPolicy *deliveryPolicyWire `json:"deliveryPolicy"`
}
UpdateSubscriptionIn is the UpdateSubscription request body.
type UpdateTopicIn ¶
UpdateTopicIn is the UpdateTopic request body.