Documentation
¶
Overview ¶
Package mq is queue and stream admin for your org: create them, watch them drain, ack what you pulled.
It is Hanzo MQ, the managed message-queue product: org-scoped administration of durable JetStream queues on the platform message plane — streams, their messages, pull consumers and delivery — served at /v1/mq over the broker apps/pubsub embeds.
THE SPLIT — mq vs pubsub ¶
One broker, two ORTHOGONAL surfaces. pubsub is the messaging DATA plane (publish, subscribe, request/reply — the subject side). mq is the queue and stream ADMIN plane plus pull delivery (create/inspect/purge/delete streams, manage consumers, pull the next batch). No operation appears on both: the authored MQ spec's publish/subscribe/request/subjects operations are deliberately NOT served here — see typed_wire_test.go, where every refused operation of the authored spec (openapi d86248f^:mq/openapi.yaml) is pinned with its reason.
TENANCY ¶
The broker is the ONE in-cluster plane; it also carries platform-internal streams (the analytics event plane, the Kafka facade's topics). Isolation is therefore enforced HERE, from the validated principal and never from a request field:
- stream NAMES are namespaced per org on the wire ("MQ_<org>_<name>") and presented bare; a caller can name only streams inside its own namespace.
- stream SUBJECTS are confined to the org's subject space "mq.<org>.>": callers state subjects RELATIVE to it ("orders.*"), the prefix is added on the way in and stripped on the way out. Two orgs can never bind overlapping subjects, and no tenant stream can capture a platform subject (event.>, commerce.>, …).
CONNECTION ¶
Mount dials pubsub.URL() — the ONE bus knob every app in this process reads — with unlimited reconnect, so this app mounts (and can describe itself) with no broker running; every op answers 503 until the plane is reachable and the health op reports degraded rather than lying. Shutdown drains the client.
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
Types ¶
type Config ¶
type Config struct {
// Name is the stream name, unique within the org (alphanumeric, hyphens, underscores).
Name string `json:"name"`
// Subjects are the org-relative subjects bound to this stream (wildcards supported). Default: the stream name.
Subjects []string `json:"subjects"`
// Retention is the retention policy: limits (default), interest, or workqueue.
Retention string `json:"retention"`
// MaxMsgs caps the number of stored messages; -1 (default) is unlimited.
MaxMsgs int64 `json:"max_msgs"`
// MaxBytes caps the stream's total stored bytes; -1 (default) is unlimited.
MaxBytes int64 `json:"max_bytes"`
// MaxAge caps message age, e.g. "24h" or "7d"; "0" (default) is unlimited.
MaxAge string `json:"max_age"`
// MaxMsgSize caps one message's size in bytes; -1 (default) is the broker's limit.
MaxMsgSize int32 `json:"max_msg_size"`
// Storage is the storage backend: file (default) or memory.
Storage string `json:"storage"`
// Replicas is the number of stream replicas (1–5); this plane runs 1.
Replicas int `json:"num_replicas"`
}
Config is a stream's configuration, spec-shaped: subjects are org-relative and durations are strings, exactly as the caller states them.
type Consumer ¶
type Consumer struct {
// Name is the consumer name.
Name string `json:"name"`
// Stream is the stream this consumer reads.
Stream string `json:"stream_name"`
// Config is the consumer's configuration.
Config Durable `json:"config"`
// Delivered is the highest delivered sequence pair.
Delivered Sequences `json:"delivered"`
// AckFloor is the highest contiguously acknowledged sequence pair.
AckFloor Sequences `json:"ack_floor"`
// Pending is the number of messages yet to be delivered.
Pending uint64 `json:"num_pending"`
// Redelivered is the number of messages currently being redelivered.
Redelivered int `json:"num_redelivered"`
// Waiting is the number of pull requests waiting for messages.
Waiting int `json:"num_waiting"`
// AckPending is the number of delivered, not yet acknowledged messages.
AckPending int `json:"num_ack_pending"`
// Created is when the consumer was created.
Created time.Time `json:"created"`
}
Consumer is one durable consumer: its configuration and delivery state.
type Durable ¶
type Durable struct {
// Name is the durable consumer name (alphanumeric, hyphens, underscores).
Name string `json:"durable_name"`
// Filter delivers only messages on this org-relative subject (wildcards supported).
Filter string `json:"filter_subject"`
// Ack is the acknowledgment policy: explicit (default), all, or none.
Ack string `json:"ack_policy"`
// Deliver is where delivery starts: all (default), last, new, by_start_sequence, by_start_time, or last_per_subject.
Deliver string `json:"deliver_policy"`
// StartSeq is the starting sequence for deliver_policy by_start_sequence.
StartSeq uint64 `json:"opt_start_seq"`
// StartTime is the starting instant for deliver_policy by_start_time.
StartTime *time.Time `json:"opt_start_time"`
// MaxDeliver caps delivery attempts per message; -1 (default) is unlimited.
MaxDeliver int `json:"max_deliver"`
// AckWait is how long the broker waits for an ack before redelivering, e.g. "30s" (default).
AckWait string `json:"ack_wait"`
// Replay is the replay pacing: instant (default) or original.
Replay string `json:"replay_policy"`
// MaxAckPending caps unacknowledged messages in flight (default 1000).
MaxAckPending int `json:"max_ack_pending"`
// Description says what this consumer is for.
Description string `json:"description"`
}
Durable is a consumer's configuration, spec-shaped.
type Health ¶
type Health struct {
// Status is ok when the message plane answers, degraded otherwise.
Status string `json:"status"`
// Version is the connected broker's server version; empty while degraded.
Version string `json:"version"`
// Uptime is how long this surface has been mounted.
Uptime string `json:"uptime"`
}
Health is the surface's liveness answer.
type Message ¶
type Message struct {
// Subject is the org-relative subject the message was stored under.
Subject string `json:"subject"`
// Data is the payload, base64-encoded.
Data string `json:"data"`
// Headers are the message headers, when any were published.
Headers map[string][]string `json:"headers,omitempty"`
// Sequence is the message's stream sequence.
Sequence uint64 `json:"sequence"`
// Timestamp is when the broker stored the message.
Timestamp time.Time `json:"timestamp"`
// Delivered is how many times a consumer has been handed this message (pulls only).
Delivered int `json:"num_delivered,omitempty"`
// Remaining is how many messages follow this one for the consumer (pulls only).
Remaining uint64 `json:"num_pending,omitempty"`
}
Message is one stored message, payload base64-encoded.
type Purge ¶
type Purge struct {
// Name is the stream name, from the path.
Name string `json:"name"`
// Filter purges only messages on this org-relative subject (wildcards supported).
Filter string `json:"filter"`
// Keep retains that many newest messages.
Keep uint64 `json:"keep"`
}
Purge narrows a purge: by org-relative subject filter, or keeping the newest messages.
type Sequences ¶
type Sequences struct {
// Consumer is the consumer's own sequence.
Consumer uint64 `json:"consumer_seq"`
// Stream is the corresponding stream sequence.
Stream uint64 `json:"stream_seq"`
}
Sequences is a consumer/stream sequence pair.
type State ¶
type State struct {
// Messages is the number of messages currently stored.
Messages uint64 `json:"messages"`
// Bytes is the total stored size.
Bytes uint64 `json:"bytes"`
// FirstSeq is the sequence of the first stored message.
FirstSeq uint64 `json:"first_seq"`
// FirstTS is the timestamp of the first stored message.
FirstTS time.Time `json:"first_ts"`
// LastSeq is the sequence of the last stored message.
LastSeq uint64 `json:"last_seq"`
// LastTS is the timestamp of the last stored message.
LastTS time.Time `json:"last_ts"`
// Consumers is the number of consumers attached to this stream.
Consumers int `json:"consumer_count"`
// Subjects is the number of distinct subjects stored.
Subjects uint64 `json:"num_subjects"`
// Deleted is the number of deleted messages (sequence gaps).
Deleted int `json:"num_deleted"`
}
State is a stream's current state on the broker.
type Stream ¶
type Stream struct {
// Name is the stream name within the org.
Name string `json:"name"`
// Config is the stream's configuration.
Config Config `json:"config"`
// State is the stream's current state.
State State `json:"state"`
// Created is when the stream was created.
Created time.Time `json:"created"`
}
Stream is one stream: its configuration and live state.