///|
/// Event envelope routed through the bus.
pub(all) struct Envelope {
id : String
topic : String
payload : EventValue
headers : Array[Pair]
timestamp_ms : Int
attempt : Int
trace : Array[String]
} derive(Eq, @debug.Debug)
///|
pub fn envelope(
id? : StringView = "",
topic : StringView,
payload : EventValue,
headers? : ArrayView[Pair] = [],
timestamp_ms? : Int = 0,
attempt? : Int = 0,
trace? : ArrayView[String] = [],
) -> Envelope {
let normalized_id = if id == "" {
stable_event_id(topic, payload, timestamp_ms)
} else {
id.to_owned()
}
{
id: normalized_id,
topic: topic.to_owned(),
payload,
headers: headers.to_owned(),
timestamp_ms,
attempt,
trace: trace.to_owned(),
}
}
///|
pub fn stable_event_id(
topic : StringView,
payload : EventValue,
timestamp_ms : Int,
) -> String {
"evt-\{topic.to_owned().length()}-\{payload.to_wire().length()}-\{timestamp_ms}"
}
///|
pub fn Envelope::with_header(
self : Envelope,
key : StringView,
value : StringView,
) -> Envelope {
let headers = self.headers.copy()
headers.push(pair(key, value))
{ ..self, headers, }
}
///|
pub fn Envelope::header(self : Envelope, key : StringView) -> String? {
let wanted = key.to_owned()
for h in self.headers {
if h.key == wanted {
return Some(h.value)
}
}
None
}
///|
pub fn Envelope::next_attempt(self : Envelope) -> Envelope {
{ ..self, attempt: self.attempt + 1 }
}
///|
pub fn Envelope::add_trace(self : Envelope, step : StringView) -> Envelope {
let trace = self.trace.copy()
trace.push(step.to_owned())
{ ..self, trace, }
}
///|
pub fn Envelope::payload_path(
self : Envelope,
path : StringView,
) -> EventValue? {
self.payload.get_path(path)
}
///|
pub fn Envelope::summary(self : Envelope) -> String {
let headers = pairs_to_wire(self.headers)
"Envelope(id=\{self.id},topic=\{self.topic},attempt=\{self.attempt},headers=\{headers},payload=\{self.payload.to_wire()})"
}
///|
pub fn Envelope::fingerprint(self : Envelope) -> String {
"\{self.topic}|\{self.payload.to_wire()}|\{pairs_to_wire(self.headers)}"
}