///|
pub(all) struct EventBatch {
name : String
events : Array[Envelope]
} derive(Eq, @debug.Debug)
///|
pub(all) struct BatchValidationIssue {
index : Int
event_id : String
topic : String
schema : String
path : String
code : String
message : String
} derive(Eq, @debug.Debug)
///|
pub(all) struct BatchValidationReport {
name : String
checked : Int
valid : Int
invalid : Int
issues : Array[BatchValidationIssue]
} derive(Eq, @debug.Debug)
///|
pub(all) struct BatchDecision {
index : Int
event_id : String
topic : String
decision : RuleDecision
} derive(Eq, @debug.Debug)
///|
pub(all) struct BatchRuleReport {
name : String
checked : Int
accepted : Int
rejected : Int
decisions : Array[BatchDecision]
} derive(Eq, @debug.Debug)
///|
pub(all) struct BatchRouteRow {
index : Int
event_id : String
topic : String
routed : Int
selected : Array[String]
} derive(Eq, @debug.Debug)
///|
pub(all) struct BatchRouteReport {
name : String
events : Int
routed_events : Int
unmatched_events : Int
deliveries_planned : Int
rows : Array[BatchRouteRow]
} derive(Eq, @debug.Debug)
///|
pub(all) struct BatchPublishReport {
name : String
events : Int
routed_events : Int
unmatched_events : Int
deliveries : Int
dead_letters_added : Int
lines : Array[String]
} derive(Eq, @debug.Debug)
///|
pub(all) struct BatchTopicCount {
topic : String
count : Int
first_index : Int
last_index : Int
} derive(Eq, @debug.Debug)
///|
pub fn event_batch(
name? : StringView = "batch",
events? : ArrayView[Envelope] = [],
) -> EventBatch {
{ name: name.to_owned(), events: events.to_owned() }
}
///|
pub fn EventBatch::len(self : EventBatch) -> Int {
self.events.length()
}
///|
pub fn EventBatch::is_empty(self : EventBatch) -> Bool {
self.events.length() == 0
}
///|
pub fn EventBatch::append(self : EventBatch, event : Envelope) -> EventBatch {
let events = self.events.copy()
events.push(event)
{ ..self, events, }
}
///|
pub fn EventBatch::append_many(
self : EventBatch,
events : ArrayView[Envelope],
) -> EventBatch {
let appended = self.events.copy()
for event in events {
appended.push(event)
}
{ ..self, events: appended }
}
///|
pub fn EventBatch::with_name(
self : EventBatch,
name : StringView,
) -> EventBatch {
{ ..self, name: name.to_owned() }
}
///|
pub fn EventBatch::filter(
self : EventBatch,
pattern : StringView,
) -> Result[EventBatch, EventRailError] {
match topic_pattern(pattern) {
Err(err) => Err(err)
Ok(parsed) => {
let events : Array[Envelope] = []
for event in self.events {
match parsed.matches_topic(event.topic) {
Err(err) => return Err(err)
Ok(true) => events.push(event)
Ok(false) => ()
}
}
Ok({ ..self, events, })
}
}
}
///|
pub fn EventBatch::to_tape(self : EventBatch) -> EventTape {
event_tape().append_many(self.events)
}
///|
pub fn EventBatch::compact_by_fingerprint(self : EventBatch) -> EventBatch {
let seen : Array[String] = []
let events : Array[Envelope] = []
for event in self.events {
let fingerprint = event.fingerprint()
if !seen.contains(fingerprint) {
seen.push(fingerprint)
events.push(event)
}
}
{ ..self, events, }
}
///|
pub fn EventBatch::sort_by_timestamp(self : EventBatch) -> EventBatch {
let events = self.events.copy()
events.sort_by(compare_event_timestamp)
{ ..self, events, }
}
///|
pub fn EventBatch::topic_counts(self : EventBatch) -> Array[BatchTopicCount] {
let counts : Array[BatchTopicCount] = []
for index, event in self.events {
increment_batch_topic_count(counts, event.topic, index)
}
counts
}
///|
pub fn EventBatch::fingerprints(self : EventBatch) -> Array[String] {
self.events.map(event => event.fingerprint())
}
///|
pub fn EventBatch::manifest_lines(self : EventBatch) -> Array[String] {
let lines : Array[String] = []
for index, event in self.events {
lines.push("idx=\{index};\{event.summary()}")
}
lines
}
///|
pub fn EventBatch::manifest(self : EventBatch) -> String {
self.manifest_lines().join("\n")
}
///|
pub fn EventBatch::validate(
self : EventBatch,
schema : EventSchema,
) -> BatchValidationReport {
let issues : Array[BatchValidationIssue] = []
let mut valid = 0
let mut invalid = 0
for index, event in self.events {
let report = schema.validate(event)
if report.valid {
valid += 1
} else {
invalid += 1
for issue in report.issues {
issues.push({
index,
event_id: event.id,
topic: event.topic,
schema: schema.name,
path: issue.path,
code: issue.code,
message: issue.message,
})
}
}
}
{ name: self.name, checked: self.events.length(), valid, invalid, issues }
}
///|
pub fn EventBatch::evaluate_rules(
self : EventBatch,
rules : RuleSet,
) -> BatchRuleReport {
let decisions : Array[BatchDecision] = []
let mut accepted = 0
let mut rejected = 0
for index, event in self.events {
let decision = rules.decide(event)
match decision {
Accept(_) => accepted += 1
Reject(_) => rejected += 1
}
decisions.push({ index, event_id: event.id, topic: event.topic, decision })
}
{
name: self.name,
checked: self.events.length(),
accepted,
rejected,
decisions,
}
}
///|
pub fn EventBatch::route_preview(
self : EventBatch,
bus : Bus,
mode? : DeliveryMode = Fanout,
) -> Result[BatchRouteReport, EventRailError] {
let rows : Array[BatchRouteRow] = []
let mut routed_events = 0
let mut unmatched_events = 0
let mut deliveries_planned = 0
for index, event in self.events {
match bus.route_with_mode(event, mode) {
Err(err) => return Err(err)
Ok(routes) => {
if routes.length() == 0 {
unmatched_events += 1
} else {
routed_events += 1
}
deliveries_planned += routes.length()
rows.push({
index,
event_id: event.id,
topic: event.topic,
routed: routes.length(),
selected: routes.map(sub => sub.id),
})
}
}
}
Ok({
name: self.name,
events: self.events.length(),
routed_events,
unmatched_events,
deliveries_planned,
rows,
})
}
///|
pub fn EventBatch::publish(
self : EventBatch,
bus : Bus,
handler : (Subscription, Envelope) -> HandlerResult,
mode? : DeliveryMode = Fanout,
) -> Result[(Bus, BatchPublishReport), EventRailError] {
let mut current = bus
let lines : Array[String] = []
let mut routed_events = 0
let mut unmatched_events = 0
let mut deliveries = 0
let mut dead_letters_added = 0
for index, event in self.events {
match current.publish_with_mode(event, mode, handler) {
Err(err) => return Err(err)
Ok((next, report)) => {
current = next
if report.routed == 0 {
unmatched_events += 1
} else {
routed_events += 1
}
deliveries += report.deliveries.length()
dead_letters_added += report.dead_letters_added
lines.push("idx=\{index} \{report.status_line()}")
}
}
}
Ok(
(
current,
{
name: self.name,
events: self.events.length(),
routed_events,
unmatched_events,
deliveries,
dead_letters_added,
lines,
},
),
)
}
///|
pub fn BatchValidationIssue::to_wire(self : BatchValidationIssue) -> String {
"idx=\{self.index};event=\{escape_wire_text(self.event_id)};topic=\{escape_wire_text(self.topic)};schema=\{escape_wire_text(self.schema)};path=\{escape_wire_text(self.path)};code=\{escape_wire_text(self.code)};message=\{escape_wire_text(self.message)}"
}
///|
pub fn BatchValidationReport::summary(self : BatchValidationReport) -> String {
"batch=\{escape_wire_text(self.name)} checked=\{self.checked} valid=\{self.valid} invalid=\{self.invalid} issues=\{self.issues.length()}"
}
///|
pub fn BatchValidationReport::issue_lines(
self : BatchValidationReport,
) -> Array[String] {
self.issues.map(issue => issue.to_wire())
}
///|
pub fn BatchDecision::to_wire(self : BatchDecision) -> String {
"idx=\{self.index};event=\{escape_wire_text(self.event_id)};topic=\{escape_wire_text(self.topic)};decision=\{self.decision.to_wire()}"
}
///|
pub fn BatchRuleReport::summary(self : BatchRuleReport) -> String {
"batch=\{escape_wire_text(self.name)} checked=\{self.checked} accepted=\{self.accepted} rejected=\{self.rejected}"
}
///|
pub fn BatchRuleReport::decision_lines(self : BatchRuleReport) -> Array[String] {
self.decisions.map(decision => decision.to_wire())
}
///|
pub fn BatchRouteRow::to_wire(self : BatchRouteRow) -> String {
"idx=\{self.index};event=\{escape_wire_text(self.event_id)};topic=\{escape_wire_text(self.topic)};routed=\{self.routed};selected=\{self.selected.join(",")}"
}
///|
pub fn BatchRouteReport::summary(self : BatchRouteReport) -> String {
"batch=\{escape_wire_text(self.name)} events=\{self.events} routed=\{self.routed_events} unmatched=\{self.unmatched_events} deliveries=\{self.deliveries_planned}"
}
///|
pub fn BatchRouteReport::route_lines(self : BatchRouteReport) -> Array[String] {
self.rows.map(row => row.to_wire())
}
///|
pub fn BatchPublishReport::summary(self : BatchPublishReport) -> String {
"batch=\{escape_wire_text(self.name)} events=\{self.events} routed=\{self.routed_events} unmatched=\{self.unmatched_events} deliveries=\{self.deliveries} dead_letters=\{self.dead_letters_added}"
}
///|
pub fn BatchPublishReport::manifest(self : BatchPublishReport) -> String {
self.lines.join("\n")
}
///|
pub fn BatchTopicCount::to_wire(self : BatchTopicCount) -> String {
"topic=\{escape_wire_text(self.topic)};count=\{self.count};first=\{self.first_index};last=\{self.last_index}"
}
///|
pub fn RuleDecision::to_wire(self : RuleDecision) -> String {
match self {
Accept(message) => "accept:\{escape_wire_text(message)}"
Reject(reason) => "reject:\{escape_wire_text(reason)}"
}
}
///|
fn increment_batch_topic_count(
counts : Array[BatchTopicCount],
topic : String,
index : Int,
) -> Unit {
for row_index, item in counts {
if item.topic == topic {
counts[row_index] = {
topic: item.topic,
count: item.count + 1,
first_index: item.first_index,
last_index: index,
}
return
}
}
counts.push({ topic, count: 1, first_index: index, last_index: index })
}
///|
fn compare_event_timestamp(left : Envelope, right : Envelope) -> Int {
if left.timestamp_ms != right.timestamp_ms {
left.timestamp_ms - right.timestamp_ms
} else if left.topic != right.topic {
left.topic.compare(right.topic)
} else {
left.id.compare(right.id)
}
}