🥮 mooncakes.io

    zmjknn/moon-stream-quality/src/core does not have a README file

    EventMeta

    source
    pub(all) struct EventMeta {
      source_id : String
      partition : Int
      offset : Int64
      ingest_timestamp : Int64
      attributes : Map[String, String]
    } derive(Eq, 
    Debug    )

    Stream event metadata containing source, partition, and offsets.

    EventMeta::new

    source
    fn EventMeta::new(source_id? : String, partition? : Int, offset? : Int64, ingest_timestamp? : Int64) -> EventMeta

    Creates a new default EventMeta.

    EventTimeWindow

    source
    pub(all) struct EventTimeWindow {
      start : Int64
      end : Int64
      events : Array[StreamEvent]
      closed : Bool
    } derive(
    Debug    )

    A half-open event-time window containing events with timestamps in [start, end).

    LateEventPolicy

    source
    pub(all) enum LateEventPolicy {
      Drop
      SideOutput
      Accept
    } derive(Eq, 
    Debug    )

    Determines what to do with an event that arrives behind the watermark.

    StreamContext

    source
    pub(all) struct StreamContext {
      watermark : Int64
      total_events_processed : Int64
      total_errors_encountered : Int64
      current_batch_id : String
      properties : Map[String, String]
    } derive(Eq, 
    Debug    )

    Streaming execution context holding watermark, event counters, and runtime configuration.

    StreamContext::get_property

    source
    fn StreamContext::get_property(self : StreamContext, key : String) -> String?

    Gets a property from context.

    StreamContext::inc_errors

    source
    fn StreamContext::inc_errors(self : StreamContext) -> Unit

    Increments error counter.

    StreamContext::inc_events

    source
    fn StreamContext::inc_events(self : StreamContext) -> Unit

    Increments event counter.

    StreamContext::new

    source
    fn StreamContext::new(batch_id? : String) -> StreamContext

    Creates a new StreamContext.

    StreamContext::reset

    source
    fn StreamContext::reset(self : StreamContext, new_batch_id : String) -> Unit

    Resets the context metrics.

    StreamContext::set_property

    source
    fn StreamContext::set_property(self : StreamContext, key : String, val : String) -> Unit

    Sets a property in context.

    StreamContext::update_watermark

    source
    fn StreamContext::update_watermark(self : StreamContext, timestamp : Int64) -> Unit

    Updates the current watermark timestamp if new timestamp is higher.

    StreamEvent

    source
    pub(all) struct StreamEvent {
      event_id : String
      timestamp : Int64
      event_type : String
      payload : Map[String, Value]
      meta : EventMeta
    } derive(Eq, 
    Debug    )

    Dynamic Stream Event containing ID, event timestamp, payload fields, and metadata.

    StreamEvent::contains_key

    source
    fn StreamEvent::contains_key(self : StreamEvent, key : String) -> Bool

    Checks if event contains a payload key.

    StreamEvent::field_count

    source
    fn StreamEvent::field_count(self : StreamEvent) -> Int

    Counts total fields in payload.

    StreamEvent::get

    source
    fn StreamEvent::get(self : StreamEvent, key : String) -> Value?

    Retrieves a value from event payload by key.

    StreamEvent::get_bool

    source
    fn StreamEvent::get_bool(self : StreamEvent, key : String) -> Bool?

    Gets a boolean value from payload.

    StreamEvent::get_double

    source
    fn StreamEvent::get_double(self : StreamEvent, key : String) -> Double?

    Gets a numeric double value from payload.

    StreamEvent::get_int

    source
    fn StreamEvent::get_int(self : StreamEvent, key : String) -> Int64?

    Gets an integer value from payload.

    StreamEvent::get_string

    source
    fn StreamEvent::get_string(self : StreamEvent, key : String) -> String?

    Gets a string value from payload.

    StreamEvent::has_field

    source
    fn StreamEvent::has_field(self : StreamEvent, key : String) -> Bool

    Checks if a key exists in payload even if value is null.

    StreamEvent::new

    source
    fn StreamEvent::new(event_id : String, timestamp : Int64, event_type? : String, payload? : Map[String, Value], meta? : EventMeta) -> StreamEvent

    Creates a new StreamEvent.

    StreamEvent::set

    source
    fn StreamEvent::set(self : StreamEvent, key : String, val : Value) -> Unit

    Sets a payload key-value pair in event.

    StreamEvent::to_json_string

    source
    fn StreamEvent::to_json_string(self : StreamEvent) -> String

    Formats StreamEvent as JSON-like string.

    Value

    source
    pub(all) enum Value {
      Null
      Bool(Bool)
      Int(Int64)
      Double(Double)
      String(String)
      Bytes(Bytes)
      Timestamp(Int64)
      Array(Array[Value])
      Map(Map[String, Value])
    } derive(Eq, 
    Debug    )

    Dynamic Value representation for stream record attributes.

    Value::as_bool

    source
    fn Value::as_bool(self : Value) -> Bool?

    Converts the value to a boolean if possible.

    Value::as_double

    source
    fn Value::as_double(self : Value) -> Double?

    Converts the value to a Double representation if possible.

    Value::as_int

    source
    fn Value::as_int(self : Value) -> Int64?

    Converts the value to Int64 if possible.

    Value::as_string

    source
    fn Value::as_string(self : Value) -> String?

    Converts the value to String representation.

    Value::compare

    source
    fn Value::compare(self : Value, other : Value) -> Int?

    Compares two values for ordering when supported.

    Value::is_array

    source
    fn Value::is_array(self : Value) -> Bool

    Returns true if the Value is an array.

    Value::is_bool

    source
    fn Value::is_bool(self : Value) -> Bool

    Returns true if the Value is a boolean.

    Value::is_double

    source
    fn Value::is_double(self : Value) -> Bool

    Returns true if the Value is a double/float.

    Value::is_int

    source
    fn Value::is_int(self : Value) -> Bool

    Returns true if the Value is an integer.

    Value::is_map

    source
    fn Value::is_map(self : Value) -> Bool

    Returns true if the Value is a map.

    Value::is_null

    source
    fn Value::is_null(self : Value) -> Bool

    Returns true if the Value is Null.

    Value::is_number

    source
    fn Value::is_number(self : Value) -> Bool

    Returns true if the Value is a numeric type (Int or Double).

    Value::is_string

    source
    fn Value::is_string(self : Value) -> Bool

    Returns true if the Value is a string.

    Value::is_timestamp

    source
    fn Value::is_timestamp(self : Value) -> Bool

    Returns true if the Value is a timestamp.

    Value::to_string

    source
    fn Value::to_string(self : Value) -> String

    Formats the Value into string representation.

    Value::type_name

    source
    fn Value::type_name(self : Value) -> String

    Human-readable type name of the value.

    WatermarkTracker

    source
    pub(all) struct WatermarkTracker {
      max_timestamp : Int64
      allowed_lateness_ms : Int64
      initialized : Bool
    } derive(Eq, 
    Debug    )

    Tracks the highest event timestamp and derives a monotonic watermark.

    WatermarkTracker::classify

    source
    fn WatermarkTracker::classify(self : WatermarkTracker, timestamp : Int64) -> WindowAssignment

    Classifies a timestamp without changing the tracker state.

    WatermarkTracker::new

    source
    fn WatermarkTracker::new(allowed_lateness_ms? : Int64) -> WatermarkTracker

    Creates a watermark tracker with a non-negative lateness allowance.

    WatermarkTracker::observe

    source
    fn WatermarkTracker::observe(self : WatermarkTracker, timestamp : Int64) -> Int64

    Observes a timestamp and returns the new monotonic watermark.

    WindowAssignment

    source
    pub(all) enum WindowAssignment {
      OnTime
      Late
      TooLate
    } derive(Eq, 
    Debug    )

    Describes the relation between an event timestamp and the current watermark.

    WindowRuntime

    source
    pub(all) struct WindowRuntime {
      window_size_ms : Int64
      allowed_lateness_ms : Int64
      max_windows : Int
      policy : LateEventPolicy
      watermark : WatermarkTracker
      windows : Array[EventTimeWindow]
      late_count : Int
      dropped_count : Int
    } derive(
    Debug    )

    Bounded event-time runtime with watermark-driven window closing.

    WindowRuntime::active_window_count

    source
    fn WindowRuntime::active_window_count(self : WindowRuntime) -> Int

    Returns the number of active windows retained by the bounded runtime.

    WindowRuntime::advance_watermark

    source
    fn WindowRuntime::advance_watermark(self : WindowRuntime, timestamp : Int64) -> Array[EventTimeWindow]

    Closes windows whose end is no later than the new watermark.

    WindowRuntime::flush

    source

    Closes and returns every active window. A second flush returns no windows.

    WindowRuntime::ingest

    source

    Inserts an event and returns its event-time classification.

    WindowRuntime::late_event_count

    source
    fn WindowRuntime::late_event_count(self : WindowRuntime) -> Int

    Returns the number of events classified as late or too late.

    WindowRuntime::new

    source
    fn WindowRuntime::new(window_size_ms : Int64, allowed_lateness_ms? : Int64, max_windows? : Int, policy? : LateEventPolicy) -> WindowRuntime

    Creates a bounded runtime. Unsafe numeric configuration is normalized to the smallest useful runtime instead of allowing an unbounded or empty one.

    Source Files

    Powered by MoonBit

    Site sourceReport issueModulesSkillsCLIBuild queueStatistics

    © 2026 mooncakes.io