Documentation
¶
Index ¶
- Constants
- func OverReleaseCount() int64
- func ReleaseMessage(m hermod.Message)
- func ResetOverReleaseCount()
- func SanitizeMap(m map[string]any) map[string]any
- func SanitizeValue(v any) any
- func TryFixJSON(data []byte) []byte
- type DefaultMessage
- func (m *DefaultMessage) After() []byte
- func (m *DefaultMessage) Before() []byte
- func (m *DefaultMessage) ClearCachedPayload()
- func (m *DefaultMessage) ClearPayloads()
- func (m *DefaultMessage) Clone() hermod.Message
- func (m *DefaultMessage) Data() map[string]any
- func (m *DefaultMessage) DataRef() map[string]any
- func (m *DefaultMessage) ID() string
- func (m *DefaultMessage) MarshalJSON() ([]byte, error)
- func (m *DefaultMessage) Metadata() map[string]string
- func (m *DefaultMessage) MetadataRef() map[string]string
- func (m *DefaultMessage) MetadataValue(key string) (string, bool)
- func (m *DefaultMessage) Operation() hermod.Operation
- func (m *DefaultMessage) Payload() []byte
- func (m *DefaultMessage) PayloadLen() int
- func (m *DefaultMessage) RefCount() int32
- func (m *DefaultMessage) Release()
- func (m *DefaultMessage) Reset()
- func (m *DefaultMessage) Retain()
- func (m *DefaultMessage) Schema() string
- func (m *DefaultMessage) SetAfter(after []byte)
- func (m *DefaultMessage) SetBefore(before []byte)
- func (m *DefaultMessage) SetData(key string, value any)
- func (m *DefaultMessage) SetID(id string)
- func (m *DefaultMessage) SetMetadata(key, value string)
- func (m *DefaultMessage) SetOperation(op hermod.Operation)
- func (m *DefaultMessage) SetPayload(payload []byte)
- func (m *DefaultMessage) SetSchema(schema string)
- func (m *DefaultMessage) SetTable(table string)
- func (m *DefaultMessage) Table() string
- func (m *DefaultMessage) ToMap() map[string]any
Constants ¶
const NonObjectPayloadKey = "payload"
NonObjectPayloadKey is where a payload that is not a JSON object is exposed.
Sources are free to emit a bare string, a scalar, an array, or bytes that are not JSON at all. Such a payload has no field names to merge into the message root, so it is preserved under this single key instead. That keeps it addressable by transformations and, crucially, keeps it in the serialised output: previously only objects survived and everything else was silently dropped.
The name is deliberately "payload": array payloads have always been exposed this way, and it is the template variable the UI advertises for notification sinks ({{.payload}}). Widening the existing key to cover strings and scalars keeps every working template working, where a new name would not.
Variables ¶
This section is empty.
Functions ¶
func OverReleaseCount ¶
func OverReleaseCount() int64
OverReleaseCount reports how many times a message was released after its refcount already reached zero. Any non-zero value means some owner is releasing a reference it does not hold; treat it as a correctness bug, not a tuning signal.
func ReleaseMessage ¶
ReleaseMessage returns a message to the pool.
func ResetOverReleaseCount ¶
func ResetOverReleaseCount()
ResetOverReleaseCount zeroes the counter. Intended for tests that assert a pipeline runs with balanced reference counting.
func SanitizeMap ¶
SanitizeMap sanitizes all values in a map.
func SanitizeValue ¶
SanitizeValue converts special types (like UUIDs) to JSON-friendly strings.
func TryFixJSON ¶
TryFixJSON attempts to fix common JSON issues like trailing commas to make unmarshaling more lenient.
It works on the bytes rather than on a string copy of them. The first thing this does is decide whether the payload even looks like JSON, and it used to copy the whole body into a string to find out — which for a text, CSV or binary body is the answer "no" at the cost of a full copy, on every message.
Types ¶
type DefaultMessage ¶
type DefaultMessage struct {
// contains filtered or unexported fields
}
DefaultMessage is a concrete implementation of the hermod.Message interface. It uses a sync.Pool to minimize allocations.
func AcquireMessage ¶
func AcquireMessage() *DefaultMessage
AcquireMessage gets a message from the pool.
func (*DefaultMessage) After ¶
func (m *DefaultMessage) After() []byte
func (*DefaultMessage) Before ¶
func (m *DefaultMessage) Before() []byte
func (*DefaultMessage) ClearCachedPayload ¶
func (m *DefaultMessage) ClearCachedPayload()
ClearCachedPayload clears only the marshaled payload bytes.
func (*DefaultMessage) ClearPayloads ¶
func (m *DefaultMessage) ClearPayloads()
ClearPayloads clears the data content of the message but keeps metadata/system fields.
func (*DefaultMessage) Clone ¶
func (m *DefaultMessage) Clone() hermod.Message
func (*DefaultMessage) Data ¶
func (m *DefaultMessage) Data() map[string]any
func (*DefaultMessage) DataRef ¶
func (m *DefaultMessage) DataRef() map[string]any
func (*DefaultMessage) ID ¶
func (m *DefaultMessage) ID() string
func (*DefaultMessage) MarshalJSON ¶
func (m *DefaultMessage) MarshalJSON() ([]byte, error)
func (*DefaultMessage) Metadata ¶
func (m *DefaultMessage) Metadata() map[string]string
func (*DefaultMessage) MetadataRef ¶
func (m *DefaultMessage) MetadataRef() map[string]string
func (*DefaultMessage) MetadataValue ¶ added in v1.8.1
func (m *DefaultMessage) MetadataValue(key string) (string, bool)
MetadataValue reads one metadata entry without copying the map.
Metadata() has to clone — handing out the live map would race with a concurrent SetMetadata — so reading a single key through it copies every other entry to throw them away. The trace propagator does exactly that twice per sink write (traceparent, then tracestate), and it is not alone. This does the lookup under the same read lock, so it is as safe and costs nothing.
func (*DefaultMessage) Operation ¶
func (m *DefaultMessage) Operation() hermod.Operation
func (*DefaultMessage) Payload ¶
func (m *DefaultMessage) Payload() []byte
func (*DefaultMessage) PayloadLen ¶ added in v1.8.1
func (m *DefaultMessage) PayloadLen() int
PayloadLen reports len(Payload()) without copying the payload.
The engine asks for a payload's size far more often than for the payload: batch-by-bytes accounting sums it for every message and the per-write debug log reports it. Going through Payload() for that cloned the bytes and threw the copy away — together 2.8 GB of the 4.9 GB a 150k-message benchmark allocated, to produce two integers.
It materialises the data map exactly as Payload() does, so the two never disagree about a message whose payload has not been marshalled yet; TestPayloadLenMatchesPayload holds them together.
func (*DefaultMessage) RefCount ¶
func (m *DefaultMessage) RefCount() int32
RefCount reports the message's current reference count.
It exists so the ownership contract can actually be asserted rather than reasoned about. Every node executor must return messages the caller owns one reference to; that invariant is invisible without being able to read the count, which is why a violation in the traversal's source branch went unnoticed until it was corrupting data. Use it in tests and diagnostics, not to make control-flow decisions: the value can change under you at any moment.
func (*DefaultMessage) Release ¶
func (m *DefaultMessage) Release()
func (*DefaultMessage) Reset ¶
func (m *DefaultMessage) Reset()
Reset clears the message state so it can be reused.
func (*DefaultMessage) Retain ¶
func (m *DefaultMessage) Retain()
func (*DefaultMessage) Schema ¶
func (m *DefaultMessage) Schema() string
func (*DefaultMessage) SetAfter ¶
func (m *DefaultMessage) SetAfter(after []byte)
func (*DefaultMessage) SetBefore ¶
func (m *DefaultMessage) SetBefore(before []byte)
func (*DefaultMessage) SetData ¶
func (m *DefaultMessage) SetData(key string, value any)
func (*DefaultMessage) SetMetadata ¶
func (m *DefaultMessage) SetMetadata(key, value string)
func (*DefaultMessage) SetOperation ¶
func (m *DefaultMessage) SetOperation(op hermod.Operation)
func (*DefaultMessage) SetPayload ¶
func (m *DefaultMessage) SetPayload(payload []byte)
func (*DefaultMessage) SetSchema ¶
func (m *DefaultMessage) SetSchema(schema string)
func (*DefaultMessage) SetTable ¶
func (m *DefaultMessage) SetTable(table string)
func (*DefaultMessage) Table ¶
func (m *DefaultMessage) Table() string
func (*DefaultMessage) ToMap ¶
func (m *DefaultMessage) ToMap() map[string]any
Source Files