Documentation
¶
Overview ¶
Package publish sends a message to the queue behind a topic key. It owns the lookup-and-send plumbing every pipeline stage otherwise repeats — resolve the key to a queue and a topic name, wrap the payload in a message, publish — and the message-ID convention that controls deduplication (see IntentID).
Every producer publishes through this package. Building a message anywhere else would put the ID choice back at each call site, which is the mistake the convention exists to prevent, so a linter restricts message construction to here and to the queue backends.
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func IntentID ¶
IntentID names the occasion to publish rather than the entity published about: entityID says what the message concerns, and cause says why this particular message exists.
Passing no cause asks for at-most-once delivery per entity — every later publish about that entity is dropped while an earlier row survives. That is right only for a hand-off that happens once in an entity's life, such as announcing that it was created. Anything re-sent by design — a wake-up, a poll, a re-dispatch, a dead-letter reconciliation — must name its cause, or it collides with that one-shot publish and is lost.
Each cause segment must be stable across redeliveries of one occurrence and different between occurrences, so derive it from whatever provoked the publish: the dependency that reached a terminal state, the build and the status observed, the dead letter being reconciled. A wall-clock reading or a random value satisfies "different" while destroying "stable", leaving every redelivery to publish again. An empty segment carries no information and makes two different occasions share an ID, so callers pass none.
func Message ¶
func Message(ctx context.Context, registry consumer.TopicRegistry, key consumer.TopicKey, params MessageParams) error
Message publishes params.Payload to the topic registered for key. Params.Tenant selects the shard and is propagated as queue-name metadata.
Params.ID selects the dedup behavior, so the caller must choose it deliberately. The queue deduplicates on (tenant, topic, partition key, message ID) against every row it has not garbage-collected yet, consumed ones included — a window with no upper bound on a busy partition. A publish that collides is reported as a success and writes nothing, and nothing retries it.
Build params.ID with IntentID: name the entity the message is about and the cause this particular message exists for. A retry of the same cause then dedups, which is what makes redelivery safe, while a new cause about the same entity can never be swallowed by an older row.
func UniqueID ¶
UniqueID returns a message ID no earlier publish has used, so the publish cannot be deduplicated away.
This is the fallback for a cause with nothing stable to name it by, and it costs the idempotency IntentID preserves: a redelivery mints a fresh ID and publishes a second time, so the consumer has to absorb the duplicate. Prefer IntentID wherever the cause can be identified.
Types ¶
type MessageParams ¶
type MessageParams struct {
// Tenant selects the queue shard.
Tenant string
// ID identifies the message for deduplication.
ID string
// Payload is the serialized message body.
Payload []byte
// PartitionKey selects the ordered partition.
PartitionKey string
// Metadata contains side-band delivery attributes.
Metadata map[string]string
}
MessageParams describes a message to publish.
Source Files