///|
pub(all) enum PlanSeverity {
PlanInfo
PlanWarning
PlanError
} derive(Eq, @debug.Debug)
///|
pub(all) struct PlanIssue {
severity : PlanSeverity
code : String
message : String
} derive(Eq, @debug.Debug)
///|
pub(all) struct SubscriptionBlueprint {
id : String
pattern : String
priority : Int
group : String
retry : RetryPolicy
enabled : Bool
guards : Array[EventPredicate]
} derive(Eq, @debug.Debug)
///|
pub(all) struct BusPlan {
name : String
subscriptions : Array[SubscriptionBlueprint]
} derive(Eq, @debug.Debug)
///|
pub(all) struct PlanBuildReport {
name : String
subscriptions : Int
enabled : Int
groups : Int
guarded : Int
issues : Array[PlanIssue]
} derive(Eq, @debug.Debug)
///|
pub(all) struct PlanCoverageRow {
index : Int
event_id : String
topic : String
selected : Array[String]
} derive(Eq, @debug.Debug)
///|
pub(all) struct PlanCoverageReport {
name : String
events : Int
matched : Int
unmatched : Int
deliveries_planned : Int
rows : Array[PlanCoverageRow]
issues : Array[PlanIssue]
} derive(Eq, @debug.Debug)
///|
pub fn subscription_blueprint(
id : StringView,
pattern : StringView,
priority? : Int = 0,
group? : StringView = "",
retry? : RetryPolicy = retry_policy(),
enabled? : Bool = true,
guards? : ArrayView[EventPredicate] = [],
) -> SubscriptionBlueprint {
{
id: id.to_owned(),
pattern: pattern.to_owned(),
priority,
group: group.to_owned(),
retry,
enabled,
guards: guards.to_owned(),
}
}
///|
pub fn bus_plan(
name? : StringView = "bus-plan",
subscriptions? : ArrayView[SubscriptionBlueprint] = [],
) -> BusPlan {
{ name: name.to_owned(), subscriptions: subscriptions.to_owned() }
}
///|
pub fn BusPlan::len(self : BusPlan) -> Int {
self.subscriptions.length()
}
///|
pub fn BusPlan::is_empty(self : BusPlan) -> Bool {
self.subscriptions.length() == 0
}
///|
pub fn BusPlan::add(
self : BusPlan,
blueprint : SubscriptionBlueprint,
) -> BusPlan {
let subscriptions = self.subscriptions.copy()
subscriptions.push(blueprint)
{ ..self, subscriptions, }
}
///|
pub fn BusPlan::validate(self : BusPlan) -> Array[PlanIssue] {
let issues : Array[PlanIssue] = []
if self.subscriptions.length() == 0 {
issues.push(
plan_issue(PlanError, "plan.empty", "plan has no subscriptions"),
)
return issues
}
let mut enabled = 0
let ids : Array[String] = []
for blueprint in self.subscriptions {
if blueprint.id == "" {
issues.push(
plan_issue(PlanError, "subscription.id", "subscription id is empty"),
)
} else if ids.contains(blueprint.id) {
issues.push(
plan_issue(
PlanError,
"subscription.duplicate",
"duplicate subscription id \{blueprint.id}",
),
)
} else {
ids.push(blueprint.id)
}
if blueprint.enabled {
enabled += 1
}
match topic_pattern(blueprint.pattern) {
Err(err) =>
issues.push(
plan_issue(
PlanError,
"pattern.invalid",
"subscription \{blueprint.id}: \{err.message()}",
),
)
Ok(parsed) =>
if parsed.is_catch_all() && blueprint.priority > 0 {
issues.push(
plan_issue(
PlanWarning,
"catch_all.priority",
"catch-all subscription \{blueprint.id} has positive priority",
),
)
}
}
if blueprint.retry.max_attempts <= 0 {
issues.push(
plan_issue(
PlanError,
"retry.max_attempts",
"subscription \{blueprint.id} retry max_attempts must be positive",
),
)
}
if blueprint.retry.base_delay_ms < 0 {
issues.push(
plan_issue(
PlanError,
"retry.base_delay",
"subscription \{blueprint.id} retry base delay must not be negative",
),
)
}
if blueprint.retry.backoff_factor <= 0 {
issues.push(
plan_issue(
PlanError,
"retry.backoff",
"subscription \{blueprint.id} retry backoff must be positive",
),
)
}
}
if enabled == 0 {
issues.push(
plan_issue(
PlanError,
"plan.disabled",
"plan has no enabled subscriptions",
),
)
}
issues
}
///|
pub fn BusPlan::build(
self : BusPlan,
) -> Result[(Bus, PlanBuildReport), EventRailError] {
let issues = self.validate()
let mut built = bus()
for blueprint in self.subscriptions {
match
built.subscribe(
blueprint.id,
blueprint.pattern,
priority=blueprint.priority,
group=blueprint.group,
retry=blueprint.retry,
enabled=blueprint.enabled,
guards=blueprint.guards,
) {
Err(err) => return Err(err)
Ok(next) => built = next
}
}
let report = plan_build_report(self.name, self.subscriptions, issues)
Ok((built, report))
}
///|
pub fn BusPlan::coverage(
self : BusPlan,
events : ArrayView[Envelope],
mode? : DeliveryMode = Fanout,
) -> Result[PlanCoverageReport, EventRailError] {
match self.build() {
Err(err) => Err(err)
Ok((built, build_report)) => {
let rows : Array[PlanCoverageRow] = []
let mut matched = 0
let mut unmatched = 0
let mut deliveries_planned = 0
for index, event in events {
match built.route_with_mode(event, mode) {
Err(err) => return Err(err)
Ok(routes) => {
if routes.length() == 0 {
unmatched += 1
} else {
matched += 1
}
deliveries_planned += routes.length()
rows.push({
index,
event_id: event.id,
topic: event.topic,
selected: routes.map(sub => sub.id),
})
}
}
}
Ok({
name: self.name,
events: events.length(),
matched,
unmatched,
deliveries_planned,
rows,
issues: build_report.issues,
})
}
}
}
///|
pub fn BusPlan::manifest(self : BusPlan) -> String {
let lines : Array[String] = []
lines.push(
"plan=\{escape_wire_text(self.name)} subscriptions=\{self.subscriptions.length()}",
)
for blueprint in self.subscriptions {
lines.push(blueprint.to_wire())
}
lines.join("\n")
}
///|
pub fn SubscriptionBlueprint::to_wire(self : SubscriptionBlueprint) -> String {
"sub=\{escape_wire_text(self.id)};pattern=\{escape_wire_text(self.pattern)};priority=\{self.priority};group=\{escape_wire_text(self.group)};enabled=\{self.enabled};retry=\{self.retry.max_attempts}/\{self.retry.base_delay_ms}/\{self.retry.backoff_factor};guards=\{self.guards.length()}"
}
///|
pub fn PlanBuildReport::summary(self : PlanBuildReport) -> String {
"plan=\{escape_wire_text(self.name)} subscriptions=\{self.subscriptions} enabled=\{self.enabled} groups=\{self.groups} guarded=\{self.guarded} issues=\{self.issues.length()}"
}
///|
pub fn PlanBuildReport::issue_lines(self : PlanBuildReport) -> Array[String] {
self.issues.map(issue => issue.to_wire())
}
///|
pub fn PlanCoverageRow::to_wire(self : PlanCoverageRow) -> String {
"idx=\{self.index};event=\{escape_wire_text(self.event_id)};topic=\{escape_wire_text(self.topic)};selected=\{self.selected.join(",")}"
}
///|
pub fn PlanCoverageReport::summary(self : PlanCoverageReport) -> String {
"plan=\{escape_wire_text(self.name)} events=\{self.events} matched=\{self.matched} unmatched=\{self.unmatched} deliveries=\{self.deliveries_planned} issues=\{self.issues.length()}"
}
///|
pub fn PlanCoverageReport::coverage_lines(
self : PlanCoverageReport,
) -> Array[String] {
self.rows.map(row => row.to_wire())
}
///|
pub fn PlanSeverity::to_wire(self : PlanSeverity) -> String {
match self {
PlanInfo => "info"
PlanWarning => "warning"
PlanError => "error"
}
}
///|
pub fn PlanIssue::to_wire(self : PlanIssue) -> String {
"severity=\{self.severity.to_wire()};code=\{escape_wire_text(self.code)};message=\{escape_wire_text(self.message)}"
}
///|
fn plan_build_report(
name : String,
subscriptions : Array[SubscriptionBlueprint],
issues : Array[PlanIssue],
) -> PlanBuildReport {
let groups : Array[String] = []
let mut enabled = 0
let mut guarded = 0
for blueprint in subscriptions {
if blueprint.enabled {
enabled += 1
}
if blueprint.group != "" && !groups.contains(blueprint.group) {
groups.push(blueprint.group)
}
if blueprint.guards.length() > 0 {
guarded += 1
}
}
{
name,
subscriptions: subscriptions.length(),
enabled,
groups: groups.length(),
guarded,
issues,
}
}
///|
fn plan_issue(
severity : PlanSeverity,
code : StringView,
message : StringView,
) -> PlanIssue {
{ severity, code: code.to_owned(), message: message.to_owned() }
}