///|
pub(all) enum TapeState {
Recorded
Replayed
Skipped(String)
} derive(Eq, @debug.Debug)
///|
pub(all) struct TapeEntry {
sequence : Int
event : Envelope
state : TapeState
} derive(Eq, @debug.Debug)
///|
pub(all) struct EventTape {
entries : Array[TapeEntry]
next_sequence : Int
} derive(Eq, @debug.Debug)
///|
pub(all) struct ReplayReport {
events_replayed : Int
deliveries : Int
dead_letters_added : Int
lines : Array[String]
} derive(Eq, @debug.Debug)
///|
pub(all) struct TopicCount {
topic : String
count : Int
first_sequence : Int
last_sequence : Int
} derive(Eq, @debug.Debug)
///|
pub fn event_tape() -> EventTape {
{ entries: [], next_sequence: 0 }
}
///|
pub fn EventTape::append(self : EventTape, event : Envelope) -> EventTape {
let entries = self.entries.copy()
entries.push({ sequence: self.next_sequence, event, state: Recorded })
{ entries, next_sequence: self.next_sequence + 1 }
}
///|
pub fn EventTape::append_many(
self : EventTape,
events : ArrayView[Envelope],
) -> EventTape {
let mut tape = self
for event in events {
tape = tape.append(event)
}
tape
}
///|
pub fn EventTape::len(self : EventTape) -> Int {
self.entries.length()
}
///|
pub fn EventTape::is_empty(self : EventTape) -> Bool {
self.entries.length() == 0
}
///|
pub fn EventTape::select(
self : EventTape,
pattern : StringView,
) -> Result[Array[Envelope], EventRailError] {
match topic_pattern(pattern) {
Err(err) => Err(err)
Ok(parsed) => {
let selected : Array[Envelope] = []
for entry in self.entries {
match parsed.matches_topic(entry.event.topic) {
Err(err) => return Err(err)
Ok(true) => selected.push(entry.event)
Ok(false) => ()
}
}
Ok(selected)
}
}
}
///|
pub fn EventTape::replay(
self : EventTape,
bus : Bus,
handler : (Subscription, Envelope) -> HandlerResult,
) -> Result[(Bus, ReplayReport), EventRailError] {
let mut current_bus = bus
let lines : Array[String] = []
let mut delivery_total = 0
let mut dead_total = 0
for entry in self.entries {
let replay_event = entry.event.add_trace("replay:\{entry.sequence}")
match current_bus.publish(replay_event, handler) {
Err(err) => return Err(err)
Ok((next_bus, report)) => {
current_bus = next_bus
delivery_total += report.deliveries.length()
dead_total += report.dead_letters_added
lines.push("seq=\{entry.sequence} \{report.status_line()}")
}
}
}
Ok(
(
current_bus,
{
events_replayed: self.entries.length(),
deliveries: delivery_total,
dead_letters_added: dead_total,
lines,
},
),
)
}
///|
pub fn EventTape::topic_counts(self : EventTape) -> Array[TopicCount] {
let counts : Array[TopicCount] = []
for entry in self.entries {
increment_topic_count(counts, entry.event.topic, entry.sequence)
}
counts
}
///|
pub fn EventTape::compact_by_fingerprint(self : EventTape) -> EventTape {
let seen : Array[String] = []
let compacted : Array[TapeEntry] = []
for entry in self.entries {
let fingerprint = entry.event.fingerprint()
if !seen.contains(fingerprint) {
seen.push(fingerprint)
compacted.push({ ..entry, sequence: compacted.length() })
}
}
{ entries: compacted, next_sequence: compacted.length() }
}
///|
pub fn EventTape::to_manifest_lines(self : EventTape) -> Array[String] {
let lines : Array[String] = []
for entry in self.entries {
lines.push(entry.to_manifest_line())
}
lines
}
///|
pub fn EventTape::manifest(self : EventTape) -> String {
self.to_manifest_lines().join("\n")
}
///|
pub fn TapeEntry::to_manifest_line(self : TapeEntry) -> String {
"seq=\{self.sequence};state=\{self.state.to_wire()};\{self.event.summary()}"
}
///|
pub fn TapeState::to_wire(self : TapeState) -> String {
match self {
Recorded => "recorded"
Replayed => "replayed"
Skipped(reason) => "skipped:\{escape_wire_text(reason)}"
}
}
///|
pub fn ReplayReport::summary(self : ReplayReport) -> String {
"replayed=\{self.events_replayed} deliveries=\{self.deliveries} dead_letters=\{self.dead_letters_added}"
}
///|
fn increment_topic_count(
counts : Array[TopicCount],
topic : String,
sequence : Int,
) -> Unit {
for idx, item in counts {
if item.topic == topic {
counts[idx] = {
topic: item.topic,
count: item.count + 1,
first_sequence: item.first_sequence,
last_sequence: sequence,
}
return
}
}
counts.push({
topic,
count: 1,
first_sequence: sequence,
last_sequence: sequence,
})
}