diff --git a/flink-processor/src/main/java/org/mobilitydb/flink/meos/wirings/demo/MeosAllTiersCapstoneDemo.java b/flink-processor/src/main/java/org/mobilitydb/flink/meos/wirings/demo/MeosAllTiersCapstoneDemo.java
new file mode 100644
index 0000000..420c613
--- /dev/null
+++ b/flink-processor/src/main/java/org/mobilitydb/flink/meos/wirings/demo/MeosAllTiersCapstoneDemo.java
@@ -0,0 +1,221 @@
+package org.mobilitydb.flink.meos.wirings.demo;
+
+import jnr.ffi.Pointer;
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple4;
+import org.apache.flink.api.java.tuple.Tuple5;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.KeyedStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
+import org.apache.flink.streaming.api.windowing.time.Time;
+import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
+import org.mobilitydb.flink.meos.MeosOpsFreeCore;
+import org.mobilitydb.flink.meos.MeosOpsTBox;
+import org.mobilitydb.flink.meos.wirings.MeosBoundedStateMap;
+import org.mobilitydb.flink.meos.wirings.MeosCrossStreamJoin;
+import org.mobilitydb.flink.meos.wirings.MeosStatelessFilter;
+import org.mobilitydb.flink.meos.wirings.MeosWindowedAggregate;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.Arrays;
+
+/**
+ * Capstone end-to-end demo composing ALL FOUR tier wirings in a single
+ * Flink DataStream pipeline.
+ *
+ * Proves the wirings compose into a realistic pipeline shape, not
+ * just work in isolation. Each tier-wiring class drives one stage of
+ * the pipeline:
+ *
+ *
{@code
+ * Stream A (vehicles) Stream B (queries)
+ * │ │
+ * ① MeosStatelessFilter │
+ * (keep events in regions of interest)│
+ * │ │
+ * ② MeosBoundedStateMap │
+ * (per-vehicle running tbox union) │
+ * │ │
+ * ③ MeosWindowedAggregate │
+ * (30s tumbling per-vehicle aggregate)│
+ * │ │
+ * └─────────────┐ ┌──────┘
+ * ↓ ↓
+ * ④ MeosCrossStreamJoin
+ * (interval-join: vehicle aggregates vs region queries
+ * within ±1m time bound, match by region key)
+ * ↓
+ * output
+ * }
+ *
+ * The pipeline answers: "for each region, which vehicles had an
+ * aggregate trajectory (running union) overlapping the region's
+ * query bbox during the latest 30-second window?"
+ *
+ *
Tier per stage:
+ *
+ * - Stateless filter — drop events outside any region of
+ * interest (per-event predicate, no state).
+ * - Bounded-state map — per-vehicle running tbox union
+ * (MEOS handle persisted across events as byte[] state).
+ * - Windowed aggregate — per-vehicle 30s tumbling tbox
+ * (window-close-only aggregation, no handle persistence across
+ * windows).
+ * - Cross-stream join — interval-join vehicle aggregates
+ * against region queries (pre-keyed by region, ±1m bound).
+ *
+ *
+ * Run with:
+ *
+ *
{@code
+ * mvn -q exec:java \
+ * -Dexec.mainClass=org.mobilitydb.flink.meos.wirings.demo.MeosAllTiersCapstoneDemo \
+ * -Dmobilityflink.meos.enabled=true
+ * }
+ */
+public final class MeosAllTiersCapstoneDemo {
+
+ private static final Logger LOG = LoggerFactory.getLogger(MeosAllTiersCapstoneDemo.class);
+
+ /** Region IDs we care about — the stateless filter drops events outside this set. */
+ private static final java.util.Set REGIONS_OF_INTEREST =
+ new java.util.HashSet<>(Arrays.asList(1, 2));
+
+ /** Vehicle event stream — (vehicleId, regionId, eventTboxWKT, eventTimeMs). */
+ private static final Tuple4[] VEHICLE_EVENTS = new Tuple4[]{
+ // window 1: [0s, 30s)
+ Tuple4.of(10, 1, "TBOX XT([0,2],[2026-01-01,2026-01-01 00:00:15])", ts("00:00:00")),
+ Tuple4.of(10, 1, "TBOX XT([1,3],[2026-01-01 00:00:15,2026-01-01 00:00:25])", ts("00:00:15")),
+ Tuple4.of(20, 2, "TBOX XT([10,12],[2026-01-01,2026-01-01 00:00:15])", ts("00:00:05")),
+ Tuple4.of(99, 9, "TBOX XT([90,92],[2026-01-01,2026-01-01 00:00:15])", ts("00:00:08")), // region 9 — dropped by stage 1
+ Tuple4.of(20, 2, "TBOX XT([11,13],[2026-01-01 00:00:15,2026-01-01 00:00:25])", ts("00:00:20")),
+ // window 2: [30s, 60s)
+ Tuple4.of(10, 1, "TBOX XT([0,4],[2026-01-01 00:00:30,2026-01-01 00:00:45])", ts("00:00:30")),
+ Tuple4.of(20, 2, "TBOX XT([10,15],[2026-01-01 00:00:30,2026-01-01 00:00:45])", ts("00:00:35")),
+ };
+
+ /** Region query stream — (regionId, queryTboxWKT, eventTimeMs). */
+ private static final Tuple2[] REGION_QUERIES = new Tuple2[]{
+ Tuple2.of(1, "TBOX XT([1,3],[2026-01-01 00:00:10,2026-01-01 00:00:25])"),
+ Tuple2.of(2, "TBOX XT([11,13],[2026-01-01 00:00:10,2026-01-01 00:00:25])"),
+ Tuple2.of(1, "TBOX XT([2,4],[2026-01-01 00:00:35,2026-01-01 00:00:50])"),
+ Tuple2.of(2, "TBOX XT([12,14],[2026-01-01 00:00:35,2026-01-01 00:00:50])"),
+ };
+
+ private static long ts(String hms) {
+ String[] parts = hms.split(":");
+ long secs = Integer.parseInt(parts[0]) * 3600L
+ + Integer.parseInt(parts[1]) * 60L
+ + Integer.parseInt(parts[2]);
+ return 1767225600000L + secs * 1000L;
+ }
+
+ public static void main(String[] args) throws Exception {
+ if (!MeosOpsTBox.MEOS_AVAILABLE) {
+ LOG.error("MEOS not available — the demo requires libmeos.");
+ System.exit(1);
+ }
+
+ StreamExecutionEnvironment env =
+ StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setParallelism(1);
+
+ // ── Stream A: vehicle events ────────────────────────────────────────
+ DataStream> rawEvents =
+ env.fromCollection(Arrays.asList(VEHICLE_EVENTS))
+ .assignTimestampsAndWatermarks(
+ WatermarkStrategy
+ .>forBoundedOutOfOrderness(Duration.ofSeconds(1))
+ .withTimestampAssigner((e, t) -> e.f3));
+
+ // ── ① STATELESS FILTER ── keep only events in regions of interest ──
+ DataStream> inRegion =
+ rawEvents.filter(new MeosStatelessFilter>(
+ evt -> REGIONS_OF_INTEREST.contains(evt.f1)));
+
+ // ── ② BOUNDED-STATE MAP ── per-vehicle running tbox union ──────────
+ // State holds the MEOS-WKT text of the per-vehicle running union;
+ // emit (vehicleId, regionId, runningUnionWKT, eventTimeMs) per event.
+ DataStream> runningUnion = inRegion
+ .keyBy(t -> t.f0) // key by vehicleId
+ .process(new MeosBoundedStateMap, Tuple4>(
+ ptr -> MeosOpsTBox.tbox_out(ptr, 6).getBytes(StandardCharsets.UTF_8),
+ bytes -> MeosOpsTBox.tbox_in(new String(bytes, StandardCharsets.UTF_8)),
+ (prior, evt) -> {
+ Pointer eventTbox = MeosOpsTBox.tbox_in(evt.f2);
+ Pointer newUnion = (prior == null)
+ ? eventTbox
+ : MeosOpsFreeCore.union_tbox_tbox(prior, eventTbox, /*strict=*/0);
+ Tuple4 output =
+ Tuple4.of(evt.f0, evt.f1, MeosOpsTBox.tbox_out(newUnion, 6), evt.f3);
+ return new MeosBoundedStateMap.MeosStep<>(newUnion, output);
+ }))
+ .returns(org.apache.flink.api.common.typeinfo.TypeInformation.of(
+ new org.apache.flink.api.common.typeinfo.TypeHint>() {}));
+
+ // ── ③ WINDOWED AGGREGATE ── per-vehicle 30s tumbling tbox union ─────
+ // Within each 30s window: take the FINAL running-union value per
+ // vehicle as the per-window summary.
+ DataStream> windowed = runningUnion
+ .keyBy(t -> t.f0) // key by vehicleId
+ .window(TumblingEventTimeWindows.of(Time.seconds(30)))
+ .process(new MeosWindowedAggregate<
+ Integer,
+ Tuple4,
+ Tuple4,
+ TimeWindow
+ >((window, events, ctx) -> {
+ // Emit the LAST event in the window (the running union at window close).
+ Tuple4 last = null;
+ for (Tuple4 e : events) {
+ last = e;
+ }
+ return last;
+ }))
+ .returns(org.apache.flink.api.common.typeinfo.TypeInformation.of(
+ new org.apache.flink.api.common.typeinfo.TypeHint>() {}));
+
+ // ── Stream B: region queries (keyed by regionId for the join) ───────
+ DataStream> queryStream =
+ env.fromCollection(Arrays.asList(REGION_QUERIES))
+ .assignTimestampsAndWatermarks(
+ WatermarkStrategy
+ .>forBoundedOutOfOrderness(Duration.ofSeconds(1))
+ .withTimestampAssigner((e, t) -> ts("00:00:20"))); // single query-time
+
+ // ── ④ CROSS-STREAM JOIN ── vehicle aggregates × region queries ──────
+ // Pre-key both by regionId; interval-join within ±1m time bound.
+ // Per matched pair, emit (regionId, vehicleId, aggUnionWKT, queryWKT, vehicleTs).
+ KeyedStream, Integer> vehiclesKeyed =
+ windowed.keyBy(t -> t.f1); // key by regionId
+ KeyedStream, Integer> queriesKeyed =
+ queryStream.keyBy(q -> q.f0);
+
+ DataStream> overlaps =
+ vehiclesKeyed.intervalJoin(queriesKeyed)
+ .between(Time.minutes(-1), Time.minutes(1))
+ .process(new MeosCrossStreamJoin<
+ Tuple4,
+ Tuple2,
+ Tuple5
+ >((vehAgg, query, ctx) -> {
+ Pointer aggTbox = MeosOpsTBox.tbox_in(vehAgg.f2);
+ Pointer queryTbox = MeosOpsTBox.tbox_in(query.f1);
+ if (MeosOpsFreeCore.overlaps_tbox_tbox(aggTbox, queryTbox) != 0) {
+ return Tuple5.of(vehAgg.f1, vehAgg.f0, vehAgg.f2, query.f1, vehAgg.f3);
+ }
+ return null;
+ }))
+ .returns(org.apache.flink.api.common.typeinfo.TypeInformation.of(
+ new org.apache.flink.api.common.typeinfo.TypeHint>() {}));
+
+ overlaps.print("capstone-output");
+
+ env.execute("MeosWirings capstone (all 4 tiers composed)");
+ }
+}