///|
pub(all) struct ProjectionSpec {
name : String
topic : TopicPattern
key_path : String
include_paths : Array[String]
require_key : Bool
} derive(Eq, @debug.Debug)
///|
pub(all) struct ProjectionIssue {
event_id : String
topic : String
code : String
message : String
} derive(Eq, @debug.Debug)
///|
pub(all) struct ProjectionRow {
key : String
topic : String
count : Int
first_timestamp_ms : Int
last_timestamp_ms : Int
last_event_id : String
fields : Array[Pair]
} derive(Eq, @debug.Debug)
///|
pub(all) struct EventProjection {
spec : ProjectionSpec
rows : Array[ProjectionRow]
skipped : Int
issues : Array[ProjectionIssue]
} derive(Eq, @debug.Debug)
///|
pub fn projection_spec(
name : StringView,
key_path : StringView,
topic? : StringView = "**",
include_paths? : ArrayView[String] = [],
require_key? : Bool = true,
) -> Result[ProjectionSpec, EventRailError] {
match topic_pattern(topic) {
Err(err) => Err(err)
Ok(parsed) =>
Ok({
name: name.to_owned(),
topic: parsed,
key_path: key_path.to_owned(),
include_paths: include_paths.to_owned(),
require_key,
})
}
}
///|
pub fn event_projection(spec : ProjectionSpec) -> EventProjection {
{ spec, rows: [], skipped: 0, issues: [] }
}
///|
pub fn ProjectionSpec::empty(self : ProjectionSpec) -> EventProjection {
event_projection(self)
}
///|
pub fn ProjectionSpec::project_events(
self : ProjectionSpec,
events : ArrayView[Envelope],
) -> Result[EventProjection, EventRailError] {
let mut projection = event_projection(self)
for event in events {
match projection.apply(event) {
Err(err) => return Err(err)
Ok(next) => projection = next
}
}
Ok(projection)
}
///|
pub fn EventTape::project(
self : EventTape,
spec : ProjectionSpec,
) -> Result[EventProjection, EventRailError] {
let mut projection = event_projection(spec)
for entry in self.entries {
match projection.apply(entry.event) {
Err(err) => return Err(err)
Ok(next) => projection = next
}
}
Ok(projection)
}
///|
pub fn EventBatch::project(
self : EventBatch,
spec : ProjectionSpec,
) -> Result[EventProjection, EventRailError] {
spec.project_events(self.events)
}
///|
pub fn EventProjection::apply(
self : EventProjection,
event : Envelope,
) -> Result[EventProjection, EventRailError] {
match self.spec.topic.matches_topic(event.topic) {
Err(err) => Err(err)
Ok(false) => Ok({ ..self, skipped: self.skipped + 1 })
Ok(true) =>
match projection_key(event, self.spec.key_path, self.spec.require_key) {
None => {
let issues = self.issues.copy()
issues.push({
event_id: event.id,
topic: event.topic,
code: "key.missing",
message: "projection key \{self.spec.key_path} is missing",
})
Ok({ ..self, skipped: self.skipped + 1, issues })
}
Some(key) => {
let rows = self.rows.copy()
upsert_projection_row(rows, key, event, self.spec.include_paths)
Ok({ ..self, rows, })
}
}
}
}
///|
pub fn EventProjection::len(self : EventProjection) -> Int {
self.rows.length()
}
///|
pub fn EventProjection::is_empty(self : EventProjection) -> Bool {
self.rows.length() == 0
}
///|
pub fn EventProjection::find(
self : EventProjection,
key : StringView,
) -> ProjectionRow? {
let wanted = key.to_owned()
for row in self.rows {
if row.key == wanted {
return Some(row)
}
}
None
}
///|
pub fn EventProjection::summary(self : EventProjection) -> String {
"projection=\{escape_wire_text(self.spec.name)} rows=\{self.rows.length()} skipped=\{self.skipped} issues=\{self.issues.length()}"
}
///|
pub fn EventProjection::manifest_lines(self : EventProjection) -> Array[String] {
self.rows.map(row => row.to_wire())
}
///|
pub fn EventProjection::manifest(self : EventProjection) -> String {
self.manifest_lines().join("\n")
}
///|
pub fn EventProjection::issue_lines(self : EventProjection) -> Array[String] {
self.issues.map(issue => issue.to_wire())
}
///|
pub fn ProjectionRow::field(self : ProjectionRow, key : StringView) -> String? {
let wanted = key.to_owned()
for item in self.fields {
if item.key == wanted {
return Some(item.value)
}
}
None
}
///|
pub fn ProjectionRow::to_wire(self : ProjectionRow) -> String {
"key=\{escape_wire_text(self.key)};topic=\{escape_wire_text(self.topic)};count=\{self.count};first=\{self.first_timestamp_ms};last=\{self.last_timestamp_ms};event=\{escape_wire_text(self.last_event_id)};fields=\{pairs_to_wire(self.fields)}"
}
///|
pub fn ProjectionIssue::to_wire(self : ProjectionIssue) -> String {
"event=\{escape_wire_text(self.event_id)};topic=\{escape_wire_text(self.topic)};code=\{escape_wire_text(self.code)};message=\{escape_wire_text(self.message)}"
}
///|
fn upsert_projection_row(
rows : Array[ProjectionRow],
key : String,
event : Envelope,
include_paths : Array[String],
) -> Unit {
let fields = projection_fields(event, include_paths)
for index, row in rows {
if row.key == key {
rows[index] = {
key: row.key,
topic: event.topic,
count: row.count + 1,
first_timestamp_ms: row.first_timestamp_ms,
last_timestamp_ms: event.timestamp_ms,
last_event_id: event.id,
fields,
}
return
}
}
rows.push({
key,
topic: event.topic,
count: 1,
first_timestamp_ms: event.timestamp_ms,
last_timestamp_ms: event.timestamp_ms,
last_event_id: event.id,
fields,
})
}
///|
fn projection_fields(
event : Envelope,
include_paths : Array[String],
) -> Array[Pair] {
let fields : Array[Pair] = []
for path in include_paths {
match projection_value(event, path) {
Some(value) => fields.push(pair(path, value))
None => ()
}
}
fields
}
///|
fn projection_key(
event : Envelope,
path : String,
require_key : Bool,
) -> String? {
match projection_value(event, path) {
Some(value) => Some(value)
None => if require_key { None } else { Some(event.id) }
}
}
///|
fn projection_value(event : Envelope, path : String) -> String? {
if path == "$id" {
return Some(event.id)
}
if path == "$topic" {
return Some(event.topic)
}
if path == "$timestamp_ms" {
return Some(event.timestamp_ms.to_string())
}
if path == "$attempt" {
return Some(event.attempt.to_string())
}
match event.payload_path(path) {
Some(value) => Some(projection_value_to_text(value))
None => None
}
}
///|
fn projection_value_to_text(value : EventValue) -> String {
match value {
VNull => "null"
VBool(actual) => if actual { "true" } else { "false" }
VInt(actual) => actual.to_string()
VText(actual) => actual
VList(_) | VRecord(_) => value.to_wire()
}
}