🥮 mooncakes.io

    moonwatermarkkit

    Event-time watermark, lateness, and window assignment primitives for MoonBit.

    watermark
    event-time
    window
    lateness
    streaming
    Download zip
    Author
    Version
    0.2.0
    License
    Apache-2.0
    Repository
    Last updated
    3 months ago
    Downloads
    30

    No dependencies

    #MoonWatermarkKit

    MoonWatermarkKit is a MoonBit foundation library for event-time watermarking, late-event classification, window assignment, and stream audit summaries.

    It is not a message broker, storage engine, or distributed stream runtime. It provides the small decision core that stream processors, log analyzers, metrics pipelines, and IoT event systems can embed.

    let state = WatermarkState::new(default_policy())
      .observe(StreamEvent::new("sensor-a", 120_000, 1, 10))
    let assignment = assign_event(state, minute_window(), StreamEvent::new("sensor-a", 122_000, 2, 11))
    println(assignment.status.name())

    #Verification

    moon fmt --check
moon info && git diff --exit-code -- '*.mbti'
moon check --target all
moon build --target all
moon test --target all
moon run cmd/main

    #Installation

    moon add Oyc996/moonwatermarkkit

    The package has no third-party runtime dependencies. WatermarkState is an immutable value, so a caller can keep one state per source or partition and persist it using the caller's own checkpoint mechanism.

    EventAssignment

    source
    pub(all) struct EventAssignment {
      event : StreamEvent
      status : EventStatus
      watermark_ms : Int
      windows : Array[TimeWindow]
    } derive(Eq, 
    Debug    )

    EventAssignment::accepted

    source
    fn EventAssignment::accepted(self : EventAssignment) -> Bool

    EventAssignment::to_json

    source
    fn EventAssignment::to_json(self : EventAssignment) -> String

    EventStatus

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

    EventStatus::name

    source
    fn EventStatus::name(self : EventStatus) -> String

    PartitionWatermark

    source
    pub(all) struct PartitionWatermark {
      partition : String
      state : WatermarkState
      idle : Bool
    } derive(Eq, 
    Debug    )

    StreamAudit

    source
    pub(all) struct StreamAudit {
      total : Int
      on_time : Int
      late : Int
      too_late : Int
      accepted : Int
    } derive(Eq, 
    Debug    )

    StreamAudit::healthy

    source
    fn StreamAudit::healthy(self : StreamAudit) -> Bool

    StreamAudit::to_json

    source
    fn StreamAudit::to_json(self : StreamAudit) -> String

    StreamEvent

    source
    pub(all) struct StreamEvent {
      key : String
      event_time_ms : Int
      arrival_time_ms : Int
      value : Int
    } derive(Eq, 
    Debug    )

    StreamEvent::new

    source
    fn StreamEvent::new(key : String, event_time_ms : Int, arrival_time_ms : Int, value : Int) -> StreamEvent

    TimeWindow

    source
    pub(all) struct TimeWindow {
      start_ms : Int
      end_ms : Int
    } derive(Eq, 
    Debug    )

    TimeWindow::contains

    source
    fn TimeWindow::contains(self : TimeWindow, event_time_ms : Int) -> Bool

    TimeWindow::is_finalized

    source
    fn TimeWindow::is_finalized(self : TimeWindow, watermark_ms : Int, allowed_lateness_ms : Int) -> Bool

    TimeWindow::new

    source
    fn TimeWindow::new(start_ms : Int, end_ms : Int) -> TimeWindow

    TimeWindow::to_json

    source
    fn TimeWindow::to_json(self : TimeWindow) -> String

    WatermarkCoordinator

    source
    pub(all) struct WatermarkCoordinator {
      policy : WatermarkPolicy
      partitions : Array[PartitionWatermark]
    } derive(Eq, 
    Debug    )

    WatermarkCoordinator::active_partition_count

    source
    fn WatermarkCoordinator::active_partition_count(self : WatermarkCoordinator) -> Int

    WatermarkCoordinator::global_watermark

    source
    fn WatermarkCoordinator::global_watermark(self : WatermarkCoordinator) -> Int?

    WatermarkCoordinator::new

    source

    WatermarkCoordinator::observe

    source
    fn WatermarkCoordinator::observe(self : WatermarkCoordinator, partition : String, event : StreamEvent) -> WatermarkCoordinator

    WatermarkCoordinator::refresh_idle

    source
    fn WatermarkCoordinator::refresh_idle(self : WatermarkCoordinator, now_ms : Int) -> WatermarkCoordinator

    WatermarkPolicy

    source
    pub(all) struct WatermarkPolicy {
      max_out_of_order_ms : Int
      allowed_lateness_ms : Int
      idle_timeout_ms : Int
    } derive(Eq, 
    Debug    )

    WatermarkPolicy::new

    source
    fn WatermarkPolicy::new(max_out_of_order_ms : Int, allowed_lateness_ms : Int, idle_timeout_ms : Int) -> WatermarkPolicy

    WatermarkState

    source
    pub(all) struct WatermarkState {
      policy : WatermarkPolicy
      max_event_time_ms : Int
      current_watermark_ms : Int
      last_arrival_ms : Int
      initialized : Bool
    } derive(Eq, 
    Debug    )

    WatermarkState::classify

    source
    fn WatermarkState::classify(self : WatermarkState, event : StreamEvent) -> EventStatus

    WatermarkState::deadline_for

    source
    fn WatermarkState::deadline_for(self : WatermarkState, event_time_ms : Int) -> Int

    WatermarkState::mark_idle

    source
    fn WatermarkState::mark_idle(self : WatermarkState, now_ms : Int) -> Bool

    WatermarkState::new

    source

    WatermarkState::observe

    source

    WindowAggregate

    source
    pub(all) struct WindowAggregate {
      window : TimeWindow
      key : String
      count : Int
      sum : Int
      late : Int
      dropped : Int
    } derive(Eq, 
    Debug    )

    WindowAggregate::add

    source

    WindowAggregate::avg_x100

    source
    fn WindowAggregate::avg_x100(self : WindowAggregate) -> Int

    WindowAggregate::new

    source
    fn WindowAggregate::new(window : TimeWindow, key : String) -> WindowAggregate

    WindowAggregate::to_json

    source
    fn WindowAggregate::to_json(self : WindowAggregate) -> String

    WindowSpec

    source
    pub(all) struct WindowSpec {
      size_ms : Int
      slide_ms : Int
    } derive(Eq, 
    Debug    )

    WindowSpec::max_windows_per_event

    source
    fn WindowSpec::max_windows_per_event(self : WindowSpec) -> Int

    WindowSpec::sliding

    source
    fn WindowSpec::sliding(size_ms : Int, slide_ms : Int) -> WindowSpec

    WindowSpec::tumbling

    source
    fn WindowSpec::tumbling(size_ms : Int) -> WindowSpec

    assign_event

    source
    fn assign_event(state : WatermarkState, spec : WindowSpec, event : StreamEvent) -> EventAssignment

    assign_windows

    source
    fn assign_windows(spec : WindowSpec, event_time_ms : Int) -> Array[TimeWindow]

    audit_assignments

    source
    fn audit_assignments(assignments : Array[EventAssignment]) -> StreamAudit

    default_policy

    source
    fn default_policy() -> WatermarkPolicy

    five_minute_sliding_window

    source
    fn five_minute_sliding_window() -> WindowSpec

    max_int

    source
    fn max_int(a : Int, b : Int) -> Int

    minute_window

    source
    fn minute_window() -> WindowSpec

    Source Files

    Powered by MoonBit

    Site sourceReport issueModulesSkillsCLIBuild queueStatistics

    © 2026 mooncakes.io