///|
pub(all) struct DeliveryStats {
total : Int
delivered : Int
dropped : Int
retry_scheduled : Int
dead_lettered : Int
} derive(Eq, @debug.Debug)
///|
pub(all) struct TopicDeliveryStats {
topic : String
stats : DeliveryStats
} derive(Eq, @debug.Debug)
///|
pub(all) struct SubscriptionDeliveryStats {
subscription_id : String
stats : DeliveryStats
} derive(Eq, @debug.Debug)
///|
pub fn delivery_stats() -> DeliveryStats {
{ total: 0, delivered: 0, dropped: 0, retry_scheduled: 0, dead_lettered: 0 }
}
///|
pub fn DeliveryStats::record(
self : DeliveryStats,
status : DeliveryStatus,
) -> DeliveryStats {
match status {
Delivered(_) =>
{ ..self, total: self.total + 1, delivered: self.delivered + 1 }
Dropped(_) => { ..self, total: self.total + 1, dropped: self.dropped + 1 }
RetryScheduled(_) =>
{
..self,
total: self.total + 1,
retry_scheduled: self.retry_scheduled + 1,
}
DeadLettered(_) =>
{ ..self, total: self.total + 1, dead_lettered: self.dead_lettered + 1 }
}
}
///|
pub fn DeliveryStats::is_clean(self : DeliveryStats) -> Bool {
self.dropped == 0 && self.retry_scheduled == 0 && self.dead_lettered == 0
}
///|
pub fn DeliveryStats::failure_count(self : DeliveryStats) -> Int {
self.dropped + self.retry_scheduled + self.dead_lettered
}
///|
pub fn DeliveryStats::to_wire(self : DeliveryStats) -> String {
"total=\{self.total};delivered=\{self.delivered};dropped=\{self.dropped};retry=\{self.retry_scheduled};dead=\{self.dead_lettered}"
}
///|
pub fn PublishReport::delivery_stats(self : PublishReport) -> DeliveryStats {
collect_delivery_stats(self.deliveries)
}
///|
pub fn Bus::delivery_stats(self : Bus) -> DeliveryStats {
collect_delivery_stats(self.deliveries)
}
///|
pub fn Bus::deliveries_for_topic(
self : Bus,
topic : StringView,
) -> Array[Delivery] {
let wanted = topic.to_owned()
self.deliveries.filter(delivery => delivery.topic == wanted)
}
///|
pub fn Bus::deliveries_for_subscription(
self : Bus,
subscription_id : StringView,
) -> Array[Delivery] {
let wanted = subscription_id.to_owned()
self.deliveries.filter(delivery => delivery.subscription_id == wanted)
}
///|
pub fn Bus::dead_letters_for_subscription(
self : Bus,
subscription_id : StringView,
) -> Array[DeadLetter] {
let wanted = subscription_id.to_owned()
self.dead_letters.filter(letter => letter.subscription_id == wanted)
}
///|
pub fn Bus::last_delivery_for_subscription(
self : Bus,
subscription_id : StringView,
) -> Delivery? {
let deliveries = self.deliveries_for_subscription(subscription_id)
deliveries.last()
}
///|
pub fn Bus::topic_delivery_stats(self : Bus) -> Array[TopicDeliveryStats] {
let rows : Array[TopicDeliveryStats] = []
for delivery in self.deliveries {
record_topic_delivery(rows, delivery)
}
rows
}
///|
pub fn Bus::subscription_delivery_stats(
self : Bus,
) -> Array[SubscriptionDeliveryStats] {
let rows : Array[SubscriptionDeliveryStats] = []
for delivery in self.deliveries {
record_subscription_delivery(rows, delivery)
}
rows
}
///|
pub fn TopicDeliveryStats::to_wire(self : TopicDeliveryStats) -> String {
"topic=\{escape_wire_text(self.topic)};\{self.stats.to_wire()}"
}
///|
pub fn SubscriptionDeliveryStats::to_wire(
self : SubscriptionDeliveryStats,
) -> String {
"subscription=\{escape_wire_text(self.subscription_id)};\{self.stats.to_wire()}"
}
///|
fn collect_delivery_stats(deliveries : Array[Delivery]) -> DeliveryStats {
for delivery in deliveries; stats = delivery_stats() {
continue stats.record(delivery.status)
} nobreak {
stats
}
}
///|
fn record_topic_delivery(
rows : Array[TopicDeliveryStats],
delivery : Delivery,
) -> Unit {
for idx, row in rows {
if row.topic == delivery.topic {
rows[idx] = { ..row, stats: row.stats.record(delivery.status) }
return
}
}
rows.push({
topic: delivery.topic,
stats: delivery_stats().record(delivery.status),
})
}
///|
fn record_subscription_delivery(
rows : Array[SubscriptionDeliveryStats],
delivery : Delivery,
) -> Unit {
for idx, row in rows {
if row.subscription_id == delivery.subscription_id {
rows[idx] = { ..row, stats: row.stats.record(delivery.status) }
return
}
}
rows.push({
subscription_id: delivery.subscription_id,
stats: delivery_stats().record(delivery.status),
})
}