Documentation
¶
Index ¶
- Constants
- func CalculateDelayAccuracy(payload *[]byte, delayMs int64) (time.Duration, bool)
- func CalculateDelayAccuracyFromDeliveryTime(deliveryTimeMs int64) (time.Duration, bool)
- func CalculateEndToEndLatency(payload *[]byte) (time.Time, time.Duration)
- func ExecuteTemplate(tmpl *template.Template, id int, seq ...uint64) string
- func FormatTimestamp(timestamp uint64) time.Time
- func GetEndpoints(serviceName string) ([]string, error)
- func InjectId(s string, id int) string
- func MessageBody(staticSize int, sizeTemplate *template.Template, id int) []byte
- func NextURI(uris []string, index *int) string
- func ParseHeaders(headerStrings []string) map[string]any
- func ParseHeadersWithTemplates(headerStrings []string) (map[string]any, map[string]*template.Template, error)
- func ParseSize(sizeStr string) (int, error)
- func ParseURI(rawURI string, defaultScheme string, defaultPort string) uri
- func PastTense(outcome string) string
- func Rate(rate float32) string
- func RateLimiter(publishRate float32) *rate.Limiter
- func ReorderUrls(urls []*url.URL, spreadConnections bool, clientId int) []*url.URL
- func ResolveTerminus(tmpl *template.Template, id int) string
- func Retry(ctx context.Context, delay time.Duration, f func() bool) bool
- func UpdatePayload(useMillis bool, payload *[]byte) *[]byte
- func WrappedSequence(len int, start int) []int
- type OutOfOrderTracker
- type SequenceCheckResult
- type SequenceStatus
Constants ¶
const ( HeaderPublisherID = "x-omq-publisher-id" HeaderSequence = "x-omq-seq" )
Variables ¶
This section is empty.
Functions ¶
func CalculateDelayAccuracy ¶ added in v0.29.0
CalculateDelayAccuracy calculates the accuracy of delayed message delivery. It compares the expected delivery time (timeSent + delay) with the actual delivery time. Returns the delay accuracy (negative means early, positive means late) and whether the message was delayed.
func CalculateDelayAccuracyFromDeliveryTime ¶ added in v0.47.0
CalculateDelayAccuracyFromDeliveryTime calculates the accuracy of delayed message delivery using an absolute delivery time (e.g. from x-opt-delivery-time annotation). The deliveryTimeMs is the expected delivery time in milliseconds since Unix epoch. Returns the delay accuracy (negative means early, positive means late) and whether the value was valid.
func ExecuteTemplate ¶ added in v0.32.0
ExecuteTemplate executes a Go template with the given publisher id. An optional seq argument provides the message sequence number for cycling through comma-separated values; when omitted, the global MessagesPublished counter is used instead.
func FormatTimestamp ¶ added in v0.7.0
func GetEndpoints ¶ added in v0.21.0
K8sGet performs a GET request on the Kubernetes API with the given URL; We only need 1 GET on a simple REST endpoint, and this way we avoid importing go-client with lots of dependencies
func NextURI ¶ added in v0.42.0
NextURI returns the next URI from the list and updates the index. The index wraps around when it reaches the end of the list.
func ParseHeaders ¶ added in v0.30.0
ParseHeaders parses header strings in the format "key1=value1,key2=value2" and converts numeric values to appropriate types (int64 or float64)
func ParseHeadersWithTemplates ¶ added in v0.31.0
func ParseHeadersWithTemplates(headerStrings []string) (map[string]any, map[string]*template.Template, error)
ParseHeadersWithTemplates parses header strings as templates. Each entry is a single key=value pair where the value may contain comma-separated items that will be cycled through per message.
func ParseSize ¶ added in v0.33.0
ParseSize parses a size string that may include units (e.g., "10mb", "1kb", "100") Supported units: b, kb, mb (case-insensitive) If no unit is specified, assumes bytes
func PastTense ¶ added in v0.42.0
PastTense converts message outcome verbs to their past tense form for logging.
func RateLimiter ¶ added in v0.24.0
func ReorderUrls ¶ added in v0.35.0
ReorderUrls reorders URLs based on SpreadConnections setting and client ID.
func ResolveTerminus ¶ added in v0.32.0
func Retry ¶ added in v0.50.0
Retry calls f repeatedly until f returns true or ctx is cancelled. A fixed delay is inserted between each failed attempt. Returns true if f succeeded, false if ctx was cancelled first.
func UpdatePayload ¶
func WrappedSequence ¶ added in v0.7.1
generate a sequence of `len` integres starting with `start` and wrapped, so that it contains all values from 0 to len-1
Types ¶
type OutOfOrderTracker ¶ added in v0.44.0
type OutOfOrderTracker struct {
// contains filtered or unexported fields
}
OutOfOrderTracker tracks per-publisher sequence numbers to detect out-of-order messages.
func NewOutOfOrderTracker ¶ added in v0.44.0
func NewOutOfOrderTracker() *OutOfOrderTracker
func (*OutOfOrderTracker) Check ¶ added in v0.44.0
func (t *OutOfOrderTracker) Check(publisherID int, seq uint64) SequenceCheckResult
Check returns the sequence status for a message from the given publisher. The first message from each publisher always returns SequenceOK.
type SequenceCheckResult ¶ added in v0.44.0
type SequenceCheckResult struct {
Status SequenceStatus
LastSeq uint64
}
SequenceCheckResult holds the outcome of a sequence check along with context about the previous message from the same publisher.
type SequenceStatus ¶ added in v0.44.0
type SequenceStatus int
const ( SequenceOK SequenceStatus = iota SequenceOutOfOrder // seq < last seen (reordered) SequenceGap // seq > last seen + 1 (missing messages) )
Source Files