///|
pub(all) struct EventDigest {
index : Int
event_id : String
topic : String
timestamp_ms : Int
attempt : Int
fingerprint : String
header_count : Int
trace_count : Int
payload_fields : Int
} derive(Eq, @debug.Debug)
///|
pub(all) struct SnapshotDeliveryRow {
subscription_id : String
stats : DeliveryStats
} derive(Eq, @debug.Debug)
///|
pub(all) struct EventSnapshot {
name : String
source : String
timestamp_ms : Int
event_count : Int
delivery_count : Int
dead_letter_count : Int
digests : Array[EventDigest]
topic_counts : Array[TopicCount]
delivery_stats : DeliveryStats
subscription_stats : Array[SnapshotDeliveryRow]
} derive(Eq, @debug.Debug)
///|
pub(all) enum SnapshotChangeKind {
SnapshotAdded
SnapshotRemoved
SnapshotChanged
} derive(Eq, @debug.Debug)
///|
pub(all) struct SnapshotChange {
kind : SnapshotChangeKind
event_id : String
before_fingerprint : String
after_fingerprint : String
} derive(Eq, @debug.Debug)
///|
pub(all) struct SnapshotDiff {
left : String
right : String
added : Int
removed : Int
changed : Int
changes : Array[SnapshotChange]
} derive(Eq, @debug.Debug)
///|
pub fn event_snapshot(
name : StringView,
source : StringView,
events : ArrayView[Envelope],
bus : Bus,
timestamp_ms? : Int = 0,
) -> EventSnapshot {
let tape = event_tape().append_many(events)
{
name: name.to_owned(),
source: source.to_owned(),
timestamp_ms,
event_count: events.length(),
delivery_count: bus.delivery_count(),
dead_letter_count: bus.dead_letter_count(),
digests: build_event_digests(events),
topic_counts: tape.topic_counts(),
delivery_stats: bus.delivery_stats(),
subscription_stats: snapshot_subscription_stats(bus),
}
}
///|
pub fn EventBatch::snapshot(
self : EventBatch,
bus : Bus,
name? : StringView = "batch-snapshot",
timestamp_ms? : Int = 0,
) -> EventSnapshot {
event_snapshot(name, self.name, self.events, bus, timestamp_ms~)
}
///|
pub fn EventTape::snapshot(
self : EventTape,
bus : Bus,
name? : StringView = "tape-snapshot",
timestamp_ms? : Int = 0,
) -> EventSnapshot {
event_snapshot(
name,
"event-tape",
self.entries.map(entry => entry.event),
bus,
timestamp_ms~,
)
}
///|
pub fn EventSnapshot::summary(self : EventSnapshot) -> String {
"snapshot=\{escape_wire_text(self.name)} source=\{escape_wire_text(self.source)} events=\{self.event_count} deliveries=\{self.delivery_count} dead_letters=\{self.dead_letter_count} topics=\{self.topic_counts.length()}"
}
///|
pub fn EventSnapshot::event_ids(self : EventSnapshot) -> Array[String] {
self.digests.map(digest => digest.event_id)
}
///|
pub fn EventSnapshot::find_event(
self : EventSnapshot,
event_id : StringView,
) -> EventDigest? {
let wanted = event_id.to_owned()
for digest in self.digests {
if digest.event_id == wanted {
return Some(digest)
}
}
None
}
///|
pub fn EventSnapshot::topic_count(
self : EventSnapshot,
topic : StringView,
) -> Int {
let wanted = topic.to_owned()
for row in self.topic_counts {
if row.topic == wanted {
return row.count
}
}
0
}
///|
pub fn EventSnapshot::subscription_stats_for(
self : EventSnapshot,
subscription_id : StringView,
) -> DeliveryStats? {
let wanted = subscription_id.to_owned()
for row in self.subscription_stats {
if row.subscription_id == wanted {
return Some(row.stats)
}
}
None
}
///|
pub fn EventSnapshot::manifest_lines(self : EventSnapshot) -> Array[String] {
let lines : Array[String] = []
lines.push(self.summary())
lines.push("delivery=\{self.delivery_stats.to_wire()}")
for topic in self.topic_counts {
lines.push(
"topic=\{escape_wire_text(topic.topic)};count=\{topic.count};first=\{topic.first_sequence};last=\{topic.last_sequence}",
)
}
for row in self.subscription_stats {
lines.push(row.to_wire())
}
for digest in self.digests {
lines.push(digest.to_wire())
}
lines
}
///|
pub fn EventSnapshot::manifest(self : EventSnapshot) -> String {
self.manifest_lines().join("\n")
}
///|
pub fn EventSnapshot::diff(
self : EventSnapshot,
other : EventSnapshot,
) -> SnapshotDiff {
let changes : Array[SnapshotChange] = []
let mut added = 0
let mut removed = 0
let mut changed = 0
for digest in self.digests {
match other.find_event(digest.event_id) {
None => {
removed += 1
changes.push({
kind: SnapshotRemoved,
event_id: digest.event_id,
before_fingerprint: digest.fingerprint,
after_fingerprint: "",
})
}
Some(next) =>
if next.fingerprint != digest.fingerprint {
changed += 1
changes.push({
kind: SnapshotChanged,
event_id: digest.event_id,
before_fingerprint: digest.fingerprint,
after_fingerprint: next.fingerprint,
})
}
}
}
for digest in other.digests {
match self.find_event(digest.event_id) {
None => {
added += 1
changes.push({
kind: SnapshotAdded,
event_id: digest.event_id,
before_fingerprint: "",
after_fingerprint: digest.fingerprint,
})
}
Some(_) => ()
}
}
{ left: self.name, right: other.name, added, removed, changed, changes }
}
///|
pub fn EventDigest::to_wire(self : EventDigest) -> String {
"idx=\{self.index};event=\{escape_wire_text(self.event_id)};topic=\{escape_wire_text(self.topic)};timestamp=\{self.timestamp_ms};attempt=\{self.attempt};headers=\{self.header_count};trace=\{self.trace_count};fields=\{self.payload_fields};fingerprint=\{escape_wire_text(self.fingerprint)}"
}
///|
pub fn SnapshotDeliveryRow::to_wire(self : SnapshotDeliveryRow) -> String {
"subscription=\{escape_wire_text(self.subscription_id)};\{self.stats.to_wire()}"
}
///|
pub fn SnapshotChangeKind::to_wire(self : SnapshotChangeKind) -> String {
match self {
SnapshotAdded => "added"
SnapshotRemoved => "removed"
SnapshotChanged => "changed"
}
}
///|
pub fn SnapshotChange::to_wire(self : SnapshotChange) -> String {
"kind=\{self.kind.to_wire()};event=\{escape_wire_text(self.event_id)};before=\{escape_wire_text(self.before_fingerprint)};after=\{escape_wire_text(self.after_fingerprint)}"
}
///|
pub fn SnapshotDiff::summary(self : SnapshotDiff) -> String {
"diff=\{escape_wire_text(self.left)}..\{escape_wire_text(self.right)} added=\{self.added} removed=\{self.removed} changed=\{self.changed}"
}
///|
pub fn SnapshotDiff::change_lines(self : SnapshotDiff) -> Array[String] {
self.changes.map(change => change.to_wire())
}
///|
fn build_event_digests(events : ArrayView[Envelope]) -> Array[EventDigest] {
let digests : Array[EventDigest] = []
for index, event in events {
digests.push({
index,
event_id: event.id,
topic: event.topic,
timestamp_ms: event.timestamp_ms,
attempt: event.attempt,
fingerprint: event.fingerprint(),
header_count: event.headers.length(),
trace_count: event.trace.length(),
payload_fields: event.payload.field_count(),
})
}
digests
}
///|
fn snapshot_subscription_stats(bus : Bus) -> Array[SnapshotDeliveryRow] {
let rows : Array[SnapshotDeliveryRow] = []
for row in bus.subscription_delivery_stats() {
rows.push({ subscription_id: row.subscription_id, stats: row.stats })
}
rows.sort_by(compare_snapshot_delivery_row)
rows
}
///|
fn compare_snapshot_delivery_row(
left : SnapshotDeliveryRow,
right : SnapshotDeliveryRow,
) -> Int {
left.subscription_id.compare(right.subscription_id)
}