From 10fece4bc9d167366e904a3d0430f6b3294d2047 Mon Sep 17 00:00:00 2001 From: psy_inf Date: Thu, 17 Sep 2026 20:36:01 +0200 Subject: [PATCH 1/2] docs: add TODO.md, the tracked list of planned work Collects the open extension ideas in one place: more examples for the newer features, named merge slots, several named graph inputs, stage labels with typed access, tracking which stage short-circuited, a config-aware registerMergeFilter, an injectable registry with duplicate detection, the documentation items around 0..n outputs and large messages, and the two limitations the README lists. Each item says what the library does today, the gap, a proposal with an API sketch, its compatibility impact and the workaround available in the meantime. Nothing here is implemented. The README roadmap now summarizes the list and links to it; its short-circuit item moved into TODO.md unchanged. --- README.md | 27 +++-- TODO.md | 291 ++++++++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 309 insertions(+), 9 deletions(-) create mode 100644 TODO.md diff --git a/README.md b/README.md index aac00ae..608c6f5 100644 --- a/README.md +++ b/README.md @@ -575,15 +575,24 @@ Run the bundled example directly after building: ## Roadmap -Planned improvements, not yet implemented: - -- **Track which stage short-circuited.** When a graph drops a message, neither - `DslFilterGraph` nor `AnyFilterChain` currently tells the caller *which* - stage returned `std::nullopt`. They should record the stage (name and - location) and expose it, e.g. via an accessor or a richer result type, so - callers can diagnose where a message was filtered out. (The `Void` type - already distinguishes an *intentional* dead-end from a dropped message; this - item covers observing *unintentional* drops.) +Planned improvements, not yet implemented, live in [TODO.md](TODO.md) — each +with what the library does today, the gap, an API sketch and the workaround +available in the meantime. The current list: + +- **More examples** for `GraphContext`, `finish()`, stateful merges, + `JoinFilter`, nested graphs and the in-band tick pattern. +- **Named merge slots**, so a fan-in group matches by name, not position. +- **Several named graph inputs** (`in.orders`, `in.quotes`), mirroring + `GraphOutputs`. +- **Stage labels and typed access** (`Summarize@stats`, `graph.stage("stats")`). +- **Track which stage short-circuited**, so a caller can see where a message + was dropped rather than only that it was. +- **Config-aware `registerMergeFilter`**, a fresh combiner per instance. +- **Injectable registry with duplicate detection**, instead of a singleton that + silently overwrites. +- **Documentation** of 0..n outputs, fan-out copies and stateful stages. +- **Lifting the current limitations**: nested stage arguments in the DSL, and + type checking inside `GraphOutputs`. ## License diff --git a/TODO.md b/TODO.md new file mode 100644 index 0000000..3e3e9e4 --- /dev/null +++ b/TODO.md @@ -0,0 +1,291 @@ +# filterGraph — TODO + +Planned work, in one place. Each item states what the library does **today**, +the **gap**, a **proposal** with an API sketch, its **compatibility** impact, +and the **workaround** that exists in the meantime. + +The list started as a set of extension ideas that came up while designing +stateful fan-out/fan-in graphs (several parallel stages whose results a +stateful merge combines, plus a result-checking stage), so most items are about +merges, stage state and the end of a run. They are written for the library in +general, and the API sketches are sketches: nothing here is implemented. + +Examples use the generic stages of the [README](README.md) (`Parse`, +`Validate`, `Summarize`, …). Shipped features are documented in the +[README](README.md) and [CHANGELOG.md](CHANGELOG.md), not here. + +## Overview + +| # | Item | Impact | Compatibility | Workaround today | +|---|------|--------|---------------|------------------| +| 1 | [More examples for the newer features](#1-more-examples-for-the-newer-features) | **high**: several features have no runnable example | additive | read the tests | +| 2 | [Named merge slots](#2-named-merge-slots) | medium | additive if done carefully | rely on group order | +| 3 | [Several named graph inputs](#3-several-named-graph-inputs) | medium | additive (new graph shape) | `std::variant` input + select stages | +| 4 | [Stage labels and typed access](#4-stage-labels-and-typed-access) | medium | additive (DSL syntax) | side registry filled by creator lambdas | +| 5 | [Track which stage short-circuited](#5-track-which-stage-short-circuited) | medium | additive | tap the edge, or log in the stage | +| 6 | [Config-aware `registerMergeFilter`](#6-config-aware-registermergefilter) | low | additive | subclass + `FilterRegistrar` creator | +| 7 | [Injectable registry, duplicate detection](#7-injectable-registry-duplicate-detection) | low | mostly additive | unique names | +| 8 | [Documentation: 0..n outputs, large messages](#8-documentation-0n-outputs-large-messages) | doc only | — | — | +| 9 | [Known limitations to lift](#9-known-limitations-to-lift) | low–medium | additive | JSON config; read `GraphOutputs` carefully | + +Two earlier items of the same list are done and therefore not repeated here: +**type-checked merge inputs** (`TypedMergeFilter` / `UniformMergeFilter` / +`registerTypedMergeFilter`) and the **end-of-stream hook** +(`MessageFilter::finish()`). Both are described in the README. + +--- + +## 1. More examples for the newer features + +**Today.** [`apps/textPipeline`](apps/textPipeline/main.cpp) and +[EXAMPLE.md](EXAMPLE.md) cover the compile-time `FilterGraph`, registering +stages, a DSL graph with a tap, fan-in with an untyped and a typed merge, +several named outputs, `validateDslGraph` and the JSON format. Several features +that shipped since have no runnable example: `GraphContext`, +`MessageFilter::finish()`, a merge with per-instance state, `JoinFilter`, a +nested graph used as a stage, a `Void` sink, and the in-band tick pattern that +stands in for a `tick()` hook. + +**Gap.** Those are exactly the features that come up once stages hold state, +and the hardest to get right from a reference description alone. Their only +executable documentation today is the test suite +([`GraphContextTests.cpp`](tests/FilterGraphTests/GraphContextTests.cpp), +[`LifecycleTests.cpp`](tests/FilterGraphTests/LifecycleTests.cpp)), which reads +as assertions rather than as a walkthrough. + +**Proposal.** One runnable app per theme, each in the style of `textPipeline` +(numbered blocks, comments that explain the *why*), plus a matching EXAMPLE.md +section per app: + +- `apps/statefulPipeline` — a stage that accumulates across messages, flushes + in `finish()` and publishes its result through the `GraphContext`; a merge + with per-instance state; a derived application context recovered with + `as()`. +- `apps/compositePipeline` — `JoinFilter` from JSON next to the equivalent DSL + merge; a `DslFilterGraph` registered as a stage of an outer graph (showing + that `finish()` and the context propagate into it); a `Void`-terminated sink; + an in-band `Tick` alternative in a variant input type. + +**Compatibility.** Additive: new targets under `apps/`, no library change. + +**Workaround today.** The tests, plus the README sections +[Graph context](README.md#graph-context) and +[Ending a run](README.md#ending-a-run). + +## 2. Named merge slots + +**Today.** Slots are positional, in the order the DSL group lists them. A merge +stage can only know which upstream a slot came from by convention. Since +type-checked merges shipped, a mis-*typed* slot is caught when the graph is +built — but two slots of the *same* type in the wrong order still pass +validation and then run silently wrong. + +**Proposal.** Optional slot names in the group, checked against the names the +stage declares: + +```text +(raw: msg, checked: valid) -> Summarize -> out.stats +``` + +```cpp +// empty = positional, today's behaviour +virtual std::vector mergeInputNames() const { return {}; } +``` + +Behaviour: + +- **Stage declares names, group uses names:** slots are matched by name, so the + order in the group no longer matters. Unknown or missing names produce + diagnostics. +- **Stage declares names, group is positional:** allowed, matched by position + (keeps today's graphs valid). +- **Stage declares nothing:** names in the group are an error (*'Summarize' has + no named slots*). + +Together with the slot types a typed merge already declares, this gives every +slot a name *and* a type, which also makes `dsl::toMermaid` output +self-explanatory. + +**Compatibility.** Keep `MergeInputs` a `std::vector` and reorder the +values into declared order before calling the stage, so stage code does not +change. Making `MergeInputs` a struct with names would be a breaking change and +is not needed. + +**Workaround today.** Rely on group order, and give same-typed slots distinct +wrapper types so that the existing slot-type check can tell them apart. + +## 3. Several named graph inputs + +**Today.** A graph has exactly one input edge `in`, with one type. A graph that +consumes several kinds of message needs a single `std::variant` input, and then +every stage has to unpack it, or one "select" stage per alternative drops the +others. + +**Proposal.** A graph with named inputs, mirroring `GraphOutputs`: + +```text +in.orders -> ParseOrder -> order +in.quotes -> ParseQuote -> quote +(order, quote) -> Match -> out +``` + +```cpp +DslFilterGraph graph(text); +graph.push("quotes", Quote{...}); // runs only the stages reachable from in.quotes +``` + +Semantics: + +- One `push` is one run. +- Input edges not fed in this run are **holes**, exactly like dropped paths + today, so merges and outputs need no new rules. +- Validation checks each named input's type at construction, which needs a way + to declare them, e.g. `GraphInputs::of("orders", "quotes")`. +- `filter(GraphInputs&&)` stays available, for "push several inputs in one run". + +**Compatibility.** Additive: `DslFilterGraph` with a plain input type +keeps the single `in` edge. + +**Workaround today.** A `std::variant` input plus one select stage per +alternative, each dropping the alternatives it does not handle. + +## 4. Stage labels and typed access + +**Today.** The owner of a `DslFilterGraph` cannot reach a stage instance. The +DSL has no labels (`#` starts a comment), and the compiled plan is private. +Tests, diagnostics and statistics therefore have to go through edges or global +state. + +**Proposal.** An optional label per stage, and typed lookup: + +```text +(msg, valid) -> Summarize@stats -> out.stats +``` + +```cpp +auto& stats = graph.stage("stats"); // throws if unknown or the type differs +``` + +- Labels are unique per graph, which is checked at construction. +- `dsl::toMermaid` can show them, and diagnostics can say `stats` instead of + "stage 'Summarize' at 3:17". +- Lookup needs `AnyMessageFilterAdapter` to expose the wrapped filter + (`std::shared_ptr` + `std::type_index`, or a virtual + `target(std::type_index)` in the style of `std::function::target`). + +**Compatibility.** Additive. The `@` character is unused in the DSL today. + +**Workaround today.** Register the stage with a creator lambda that records +every instance it builds in a side registry, or have the stage publish what the +owner needs through the `GraphContext`. + +## 5. Track which stage short-circuited + +**Today.** When a graph drops a message, neither `DslFilterGraph` nor +`AnyFilterChain` tells the caller *which* stage returned `std::nullopt`. + +**Gap.** A graph that silently produces nothing is hard to diagnose: the caller +sees an empty `std::optional` and has to bisect the graph to find the stage that +dropped the message. (`Void` already distinguishes an *intentional* dead end +from a dropped message; this item is about observing *unintentional* drops.) + +**Proposal.** Record the stage that short-circuited — its name, and its location +in the DSL text — and expose it, e.g. through an accessor valid after a run, or +a richer result type. Whatever the shape, it should stay allocation-free on the +happy path and say something useful for a drop inside a nested graph. + +**Compatibility.** Additive as an accessor; a new result type would be breaking, +so that variant needs an overload or an opt-in. + +**Workaround today.** Tap the suspect edge with a logging stage ending in `end`, +or log inside the stage that decides to drop. + +## 6. Config-aware `registerMergeFilter` + +**Today.** `registerMergeFilter(name, combiner)` ignores the DSL config and +**copies one combiner into every instance**. A combiner lambda that captures a +`shared_ptr` therefore **shares state across all instances and graphs**. The +header says so; the README and EXAMPLE.md do not. + +A merge with config and per-instance state is already possible: derive from +`MessageFilter` (or from `TypedMergeFilter`) and register it +with `FilterRegistrar(name, creator)`. The gap is ergonomics and +documentation, not capability. + +**Proposal.** + +```cpp +// A fresh combiner per instance, built from the stage's DSL config. +template +void registerMergeFilter(const std::string& name, + std::function::Combiner(const nlohmann::json& config)> factory); +``` + +Plus a short EXAMPLE.md section on "stateful merges: subclass + +`FilterRegistrar`" (see item 1), and a note in the README on the copy semantics +of the existing overload. + +**Compatibility.** Additive overload. + +**Workaround today.** Subclass and register with a creator lambda. + +## 7. Injectable registry, duplicate detection + +**Today.** `FilterRegistry::instance()` is a process-wide singleton, and +`registerFilter` **silently overwrites** an existing name. + +**Gap.** + +- Two libraries registering the same name shadow each other without notice. +- A test that wants to swap a stage for a fake has to mutate global state and + restore it. +- Graphs cannot be built against different stage sets in one process. + +**Proposal.** + +- `registerFilter` reports a duplicate name (throw, or return `false`), with an + explicit `replaceFilter(name, creator)` for intentional overrides. +- `DslFilterGraph(text, const FilterRegistry& registry = FilterRegistry::instance())`, + and the same parameter on `validateDslGraph`, `validateGraph` and + `JsonFilterGraph`. +- A copyable `FilterRegistry` (or `FilterRegistry::derive()`), so a test can + start from the global set and replace single stages. + +**Compatibility.** Turning an overwrite into an error is behaviour-breaking. +Ship it behind a transition: warn first, or add +`registerFilter(..., OnDuplicate)`. + +**Workaround today.** Keep names unique, and register test fakes under their own +names. + +## 8. Documentation: 0..n outputs, large messages + +No API change. These points belong in the README / EXAMPLE.md, because they come +up as soon as stages carry state: + +- **0..n outputs per input.** A stage emits at most one value per run. A general + multi-output stage would conflict with merge semantics (which output pairs + with which slot?), so the recommendation is **not** to add one. Document the + convention instead: a stage that can produce several results has a collection + as its output type, and downstream stages iterate over it. +- **Fan-out copies.** Every reader of an edge except the last gets a copy + (`GraphPlan::read`). Cheap values are fine; large or shared payloads should + travel as `std::shared_ptr`. +- **Stateful stages.** A stage instance lives as long as its graph, and state in + its members persists across messages. Say explicitly that this is supported + and intended, and that each graph construction creates fresh instances. + +## 9. Known limitations to lift + +The [current limitations](README.md#current-limitations) the README lists, as +work items: + +- **Nested stage arguments in the DSL.** Arguments are flat `key=value` pairs; + nested objects and lists are not expressible, so a stage that needs them has + to be configured in JSON. Lifting this means a value grammar for objects and + arrays, plus diagnostics for it. +- **Type checking inside `GraphOutputs`.** Only the graph's input type and the + single `out` type are checked against the C++ template parameters. The types + behind `out.` are checked when they are read (`get` throws + `std::bad_any_cast`), not when the graph is built. Checking them up front + needs a way to declare the expected type per key. From 9f2866aa36e3cc1a3cd082576fc57656c2e95626 Mon Sep 17 00:00:00 2001 From: psy_inf Date: Thu, 17 Sep 2026 20:56:15 +0200 Subject: [PATCH 2/2] docs: add two runnable examples for the stateful and composite features apps/textPipeline covers the core building blocks; the features that shipped since had no runnable example. Two samples fill the gap, in the same style (numbered blocks, comments that explain the why), each walked through in EXAMPLE.md: - statefulPipeline (section 8): a stage that accumulates across messages and flushes in finish(), handing its result to the application through the GraphContext; a merge with per-instance state and per-instance arguments (subclass + FilterRegistrar) next to the shared-combiner semantics of registerMergeFilter, shown with a captured counter; an application's own derived context, recovered with as(). - compositePipeline (section 9): registerTypedMergeFilter and the JSON JoinFilter side by side; a DslFilterGraph registered as a stage of an outer graph, with finish() and the context reaching into it; a Void-terminated sink graph; the in-band tick message that stands in for a tick() hook. Also refreshes the textPipeline output in EXAMPLE.md, which predates the typed-merge stages, and drops the examples item from TODO.md. No library change; the new files follow the style of the existing sources. --- CHANGELOG.md | 15 + EXAMPLE.md | 384 +++++++++++++++++++++++++- README.md | 15 +- TODO.md | 87 ++---- apps/CMakeLists.txt | 2 + apps/compositePipeline/CMakeLists.txt | 23 ++ apps/compositePipeline/main.cpp | 297 ++++++++++++++++++++ apps/statefulPipeline/CMakeLists.txt | 23 ++ apps/statefulPipeline/main.cpp | 292 ++++++++++++++++++++ 9 files changed, 1063 insertions(+), 75 deletions(-) create mode 100644 apps/compositePipeline/CMakeLists.txt create mode 100644 apps/compositePipeline/main.cpp create mode 100644 apps/statefulPipeline/CMakeLists.txt create mode 100644 apps/statefulPipeline/main.cpp diff --git a/CHANGELOG.md b/CHANGELOG.md index e928ded..efcf857 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -49,6 +49,21 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 merge stage declares its slot types; `AnyMessageFilter::mergeInputTypes()` exposes them to the DSL. `MergeInputs` moved here from `MergeFilter.hpp` (which still provides it). +- **`apps/statefulPipeline` and `apps/compositePipeline`** — two runnable + examples for the features `apps/textPipeline` does not cover. + `statefulPipeline` shows a stage that accumulates across messages and flushes + in `finish()`, hands its result to the application through the + `GraphContext`, a merge with per-instance state (subclass + + `FilterRegistrar`) next to the shared-combiner semantics of + `registerMergeFilter`, and an application's own derived context recovered + with `as()`. `compositePipeline` shows `registerTypedMergeFilter` + and the JSON `JoinFilter` side by side, a `DslFilterGraph` registered as a + stage of an outer graph (with `finish()` and the context reaching into it), a + `Void`-terminated sink graph, and the in-band tick message. EXAMPLE.md walks + through both in sections 8 and 9. +- **`TODO.md`** — the planned work, with what the library does today, the gap, + an API sketch and the workaround for each item; the README roadmap summarizes + it. - **`MessageFilter::setContext` / `context()` / `sharedContext()`** — stages receive the graph's `GraphContext`. `FilterGraph`, `DslFilterGraph`, `JsonFilterGraph`, `AnyFilterChain`, `FanoutFilter` and `JoinFilter` create an diff --git a/EXAMPLE.md b/EXAMPLE.md index 5a259f6..2b3b3cd 100644 --- a/EXAMPLE.md +++ b/EXAMPLE.md @@ -1,11 +1,15 @@ # filterGraph by example -This walkthrough follows the runnable sample in -[`apps/textPipeline/main.cpp`](apps/textPipeline/main.cpp), a small text -pipeline that exercises every core building block: a **compile-time** -`FilterGraph`, then **runtime graphs described in the text DSL** — a tap -(fan-out), a merge (fan-in), several named outputs and up-front diagnostics — -and finally the same kind of pipeline in the **JSON format**. +This walkthrough follows three runnable samples under [`apps/`](apps/), all of +them small text pipelines: + +| Sections | Sample | Covers | +| --- | --- | --- | +| 1–7 | [`textPipeline`](apps/textPipeline/main.cpp) | the core building blocks: a **compile-time** `FilterGraph`, **runtime graphs in the text DSL** (a tap, a merge, several named outputs, up-front diagnostics) and the same pipeline in the **JSON format** | +| 8 | [`statefulPipeline`](apps/statefulPipeline/main.cpp) | stages that **carry state**: `finish()`, the `GraphContext`, a merge with per-instance state | +| 9 | [`compositePipeline`](apps/compositePipeline/main.cpp) | **composition**: `JoinFilter`, a graph nested as a stage, a `Void` sink, in-band ticks | + +Start at the top: sections 8 and 9 assume the vocabulary of 1–7. ## Core concept: a chain of stages @@ -440,21 +444,331 @@ returns every problem, located by a JSON pointer such as `/1/config/paths/0/0`. Unlike the DSL checks, it cannot see the graph's declared input/output types, and it stops type-checking across a `Fanout` or `Join`. -## Build & run this example +## 8. Stateful stages: `finish()` and the graph context + +The sections above transform one message at a time. Stages may also **carry +state**: a stage instance lives as long as the graph it belongs to, so state in +its members persists across messages, and every graph construction creates +fresh instances. The second sample, +[`apps/statefulPipeline/main.cpp`](apps/statefulPipeline/main.cpp), is about +those stages. + +A stage that accumulates something has no natural point at which to report it — +it only ever runs because a message arrived. `finish()` is that point: + +```cpp +class CollectFilter : public MessageFilter +{ +public: + std::optional filter(std::string&& line) override + { + ++mSummary.lines; + mSummary.words += countWords(line); + return std::move(line); // the line itself passes through unchanged + } + + void finish() override // once, after the last message + { + std::cout << std::format("[collect] end of stream: {} lines, {} words\n", + mSummary.lines, mSummary.words); + context().set(mSummary); // hand the result to the application + } + +private: + Summary mSummary; +}; +``` + +`finish()` produces no message, so a final *result* travels through the +[graph context](README.md#graph-context) instead — a type-keyed blackboard that +every stage of a graph shares: + +```cpp +DslFilterGraph graph("in -> Collect -> out"); + +for (auto line : lines) { graph.filter(std::move(line)); } +graph.finish(); // the owner's call + +const auto summary = graph.context().get(); // std::optional +``` + +```text +[collect] end of stream: 3 lines, 12 words +[app] read from the context: 3 lines, 12 words +``` + +Nothing is reported while messages flow, and a graph that is never finished +never flushes. + +### A merge with per-instance state + +A merge stage may hold state too, but *how* it is registered decides whether +that state is per instance or shared. Subclassing and registering a creator +gives every instance its own — here a `TypedMergeFilter` with a sliding window +whose size comes from the stage's arguments: + +```cpp +class TrendFilter : public TypedMergeFilter +{ +public: + explicit TrendFilter(std::size_t window) : mWindow(window) {} + + std::optional merge(std::optional&& length, + std::optional&& words) override + { /* push length into mLengths, drop the oldest, average what is left */ } + +private: + std::size_t mWindow; + std::deque mLengths; +}; + +static FilterRegistrar registerTrend("Trend", [](const nlohmann::json& config) { + return std::make_shared(config.value("window", std::size_t{3})); +}); +``` + +```text +in -> Length -> length +in -> Words -> words +(length, words) -> Trend(window=2) -> out.short +(length, words) -> Trend(window=3) -> out.long +``` + +Two instances of one stage type, each with its own window and its own history: + +```text +[trend] short: chars=19 words=4 avg(last 2)=19.0 +[trend] long: chars=19 words=4 avg(last 3)=19.0 +[trend] short: chars=10 words=2 avg(last 2)=14.5 +[trend] long: chars=10 words=2 avg(last 3)=14.5 +[trend] short: chars=30 words=6 avg(last 2)=20.0 +[trend] long: chars=30 words=6 avg(last 3)=19.7 +``` + +`registerMergeFilter` (and `registerTypedMergeFilter`) behave differently on +purpose: they **copy one combiner into every instance**, so whatever the +combiner captures is shared by all of them — across instances *and* across +graphs. The sample registers such a merge with a captured counter and uses it +twice in one graph: + +```text +[tally] call 1 of the one shared combiner (2 slots) / call 2 of the one shared combiner (2 slots) +[tally] call 3 of the one shared combiner (2 slots) / call 4 of the one shared combiner (2 slots) +[tally] call 5 of the one shared combiner (2 slots) / call 6 of the one shared combiner (2 slots) +``` + +The numbers run straight through both stages. For a stateless combiner that is +exactly what you want; for state, use the `Trend` pattern above. + +### An application's own context + +`GraphContext` is a polymorphic base, so an application can derive its own and +hand it to the graph. A stage recovers it with `as()`: + +```cpp +class AppContext : public GraphContext +{ +public: + explicit AppContext(std::string session) : sessionId(std::move(session)) {} + std::string sessionId; +}; + +auto context = std::make_shared("session-42"); +graph.setContext(context); // give it to the GRAPH, not to single stages +``` + +```cpp +const auto* app = context().as(); // nullptr if it is another type +return app ? std::format("[{}] {}", app->sessionId, line) : std::move(line); +``` + +Give the context to the graph: a graph overwrites its stages' contexts with its +own when they are built, so a context handed to a single stage would be +replaced. A stage that calls `as()` depends on that type, so reusable +stages should stick to the type-keyed `set`/`get`. + +## 9. Composition: joins, nested graphs, sinks and ticks + +Every graph is itself a `MessageFilter`, which is what makes graphs composable. +The third sample, +[`apps/compositePipeline/main.cpp`](apps/compositePipeline/main.cpp), puts the +composition features side by side. + +### A typed merge from a lambda, and its JSON twin + +Section 4 registered a merge from a combiner and section 4's *typed slots* +subsection subclassed `TypedMergeFilter`. `registerTypedMergeFilter` is the +third way: typed slots, registered from a lambda. + +```cpp +registerTypedMergeFilter( + "Concat", + [](std::optional&& upper, + std::optional&& reversed) -> std::optional { + return std::format("{} | {}", upper.value_or("-"), reversed.value_or("-")); + }); +``` + +```text +in -> Upper -> upper +in -> Reverse -> reversed +(upper, reversed) -> Concat -> out +``` + +The JSON format expresses the same shape as a **`JoinFilter`**: it scatters a +copy of the message through each configured path and hands the gathered results +to a C++ combiner (the paths come from configuration, the combiner cannot). + +```cpp +registerJoinFilter("Join", [](std::vector&& slots) { + /* one slot per path, in path order; an empty slot is a hole */ +}); +``` + +```json +[ + { "type": "Join", "config": { "paths": [ + [ { "type": "Upper" } ], + [ { "type": "Reverse" } ] + ] } } +] +``` + +Both print the same thing: + +```text +[dsl] COMPOSE ME | em esopmoc +[json] COMPOSE ME | em esopmoc +``` + +### A graph as a stage + +A `DslFilterGraph` can be registered like any other stage, which makes a whole +graph reusable inside another: + +```cpp +static FilterRegistrar> registerInner( + "Inner", [](const nlohmann::json&) { + return std::make_shared>( + "in -> Tag -> tagged -> Count -> out"); + }); +``` + +```text +in -> Inner -> inner -> Upper -> out +``` + +```mermaid +flowchart LR + In(["in"]) --> Inner + subgraph Inner["Inner (a graph as a stage)"] + direction LR + T["Tag"] --> Tagged(["tagged"]) --> C["Count"] + end + Inner --> InnerEdge(["inner"]) --> U["Upper"] --> Out(["out"]) +``` + +The outer graph hands its context to the nested one, and finishing the outer +graph finishes the inner stages too — so `Tag` sees a value the *application* +published, and `Count` reports when the outer graph is finished: + +```cpp +graph.context().set(Tag{"[tagged] "}); +graph.filter(std::string{"nested graphs compose"}); +graph.filter(std::string{"and share a context"}); +graph.finish(); +``` + +```text +[outer] [TAGGED] NESTED GRAPHS COMPOSE +[outer] [TAGGED] AND SHARE A CONTEXT +[nested] the inner stage saw 2 message(s) +``` + +### A graph that only has side effects + +When every path ends in a sink, the graph has no consumable output: its stages +are declared `MessageFilter` and its paths end in `end`. +`Void` is a deliberate dead end, unlike `std::nullopt`, which means a message +was dropped. + +```cpp +class WriteFilter : public MessageFilter +{ +public: + std::optional filter(std::string&& text) override + { + std::cout << "[sink] " << text << '\n'; + return Void{}; + } +}; + +DslFilterGraph graph(R"dsl( + in -> Upper -> upper -> Write -> end + in -> Reverse -> reversed -> Write -> end +)dsl"); + +const auto ran = graph.filter(std::string{"side effects only"}); // optional +``` + +```text +[sink] SIDE EFFECTS ONLY +[sink] ylno stceffe edis +[sink] the graph ran: true +``` + +### Time: the in-band tick message + +There is no `tick()` hook, because time is domain-specific (event time, wall +clock, a sensor clock). A stage that must act while no data arrives is fed +**ticks as messages**: the graph's input type is a variant of the payload and a +`Tick`, so ticks travel the same paths as everything else. + +```cpp +struct Tick {}; +using Event = std::variant; + +class BatchFilter : public MessageFilter +{ +public: + std::optional filter(Event&& event) override + { + if (const auto* line = std::get_if(&event)) + { + mBatch.push_back(*line); + return std::nullopt; // buffered: nothing downstream runs + } + return flushBatch(); // a Tick emits what has accumulated + } + // finish() reports whatever is still buffered when the stream ends +}; + +DslFilterGraph graph("in -> Batch -> out"); +``` + +```text +[batch] first, second +[batch] third +[batch] 1 line(s) left unflushed at the end of the stream +``` + +## Build & run these examples ```powershell # Configure + build (pick a preset for your toolchain from CMakePresets.json) cmake --preset windows-msvc-release-user-mode cmake --build --preset windows-msvc-release-user-mode -# Run the sample +# Run the samples ./out/build/windows-msvc-release-user-mode/apps/textPipeline/textPipeline +./out/build/windows-msvc-release-user-mode/apps/statefulPipeline/statefulPipeline +./out/build/windows-msvc-release-user-mode/apps/compositePipeline/compositePipeline ``` On Linux/macOS, use a matching preset such as `unixlike-gcc-release` or `unixlike-clang-release`. -The complete output: +The complete output of `textPipeline` (sections 1–7): ```text [compile-time] !HPARGRETLIF ,OLLEH @@ -462,9 +776,61 @@ The complete output: [main] !hparGretlif ,olleH [tap] AB [merge] HELLO, FILTERGRAPH! | !hparGretlif ,olleH (1 hole) +[typed] HELLO, FILTERGRAPH! (19 chars) +[typed check] 3:20: slot 1 of 'Report' expects 'class std::basic_string' but edge 'length' carries 'unsigned __int64' +[typed check] 3:20: slot 2 of 'Report' expects 'unsigned __int64' but edge 'upper' carries 'class std::basic_string' [outputs] upper=HELLO, FILTERGRAPH! length=19 long=dropped [check] 1:7: unknown stage type 'Uppercas' — did you mean 'Uppercase'? [check] 2:7: could not construct 'MinLength': [json.exception.out_of_range.403] key 'minLength' not found [json tap] HELLO, FILTERGRAPH! [json main] !hparGretlif ,olleH ``` + +(The `[typed check]` lines print the full `typeid(...).name()` spelling, which +depends on the compiler; it is shortened here.) + +Of `statefulPipeline` (section 8): + +```text +[collect] end of stream: 3 lines, 12 words +[app] read from the context: 3 lines, 12 words + +[trend] short: chars=19 words=4 avg(last 2)=19.0 +[trend] long: chars=19 words=4 avg(last 3)=19.0 +[trend] short: chars=10 words=2 avg(last 2)=14.5 +[trend] long: chars=10 words=2 avg(last 3)=14.5 +[trend] short: chars=30 words=6 avg(last 2)=20.0 +[trend] long: chars=30 words=6 avg(last 3)=19.7 + +[tally] call 1 of the one shared combiner (2 slots) / call 2 of the one shared combiner (2 slots) +[tally] call 3 of the one shared combiner (2 slots) / call 4 of the one shared combiner (2 slots) +[tally] call 5 of the one shared combiner (2 slots) / call 6 of the one shared combiner (2 slots) + +[stamped] [session-42] the quick brown fox +[stamped] [session-42] jumps over +[stamped] [session-42] the lazy dog and keeps running +[collect] end of stream: 3 lines, 15 words +[app] session session-42 saw 3 lines +``` + +(15 words, not 12: in that last graph `Stamp` runs before `Collect`, so each +line carries the session stamp as an extra word.) + +And of `compositePipeline` (section 9): + +```text +[dsl] COMPOSE ME | em esopmoc +[json] COMPOSE ME | em esopmoc + +[outer] [TAGGED] NESTED GRAPHS COMPOSE +[outer] [TAGGED] AND SHARE A CONTEXT +[nested] the inner stage saw 2 message(s) + +[sink] SIDE EFFECTS ONLY +[sink] ylno stceffe edis +[sink] the graph ran: true + +[batch] first, second +[batch] third +[batch] 1 line(s) left unflushed at the end of the stream +``` diff --git a/README.md b/README.md index 608c6f5..2535172 100644 --- a/README.md +++ b/README.md @@ -11,8 +11,11 @@ paths, merge them back together, and drop/short-circuit messages — all without hard-coding the pipeline shape in source code. > **New here? Start with the [example walkthrough (EXAMPLE.md)](EXAMPLE.md)** — -> a step-by-step, diagrammed tour of the runnable -> [`apps/textPipeline`](apps/textPipeline/main.cpp) sample. +> a step-by-step, diagrammed tour of three runnable samples: +> [`apps/textPipeline`](apps/textPipeline/main.cpp) (the core building blocks), +> [`apps/statefulPipeline`](apps/statefulPipeline/main.cpp) (stages that carry +> state) and [`apps/compositePipeline`](apps/compositePipeline/main.cpp) +> (composing graphs out of graphs). > **A note on the word "filter".** Here "filter" follows the Unix-pipeline and > media-graph (DirectShow / GStreamer / FFmpeg) tradition: a stage that reads a @@ -567,20 +570,22 @@ Testing is enabled by default (`-DENABLE_TESTING=ON`); pass `-DENABLE_TESTING=OFF` to skip building the tests. Tests use Catch2 (fetched via CPM) and live under `tests/`. -Run the bundled example directly after building: +Run the bundled examples directly after building: ```powershell ./out/build/windows-msvc-release-user-mode/apps/textPipeline/textPipeline +./out/build/windows-msvc-release-user-mode/apps/statefulPipeline/statefulPipeline +./out/build/windows-msvc-release-user-mode/apps/compositePipeline/compositePipeline ``` +[EXAMPLE.md](EXAMPLE.md) walks through all three, with their output. + ## Roadmap Planned improvements, not yet implemented, live in [TODO.md](TODO.md) — each with what the library does today, the gap, an API sketch and the workaround available in the meantime. The current list: -- **More examples** for `GraphContext`, `finish()`, stateful merges, - `JoinFilter`, nested graphs and the in-band tick pattern. - **Named merge slots**, so a fan-in group matches by name, not position. - **Several named graph inputs** (`in.orders`, `in.quotes`), mirroring `GraphOutputs`. diff --git a/TODO.md b/TODO.md index 3e3e9e4..e149843 100644 --- a/TODO.md +++ b/TODO.md @@ -18,61 +18,25 @@ Examples use the generic stages of the [README](README.md) (`Parse`, | # | Item | Impact | Compatibility | Workaround today | |---|------|--------|---------------|------------------| -| 1 | [More examples for the newer features](#1-more-examples-for-the-newer-features) | **high**: several features have no runnable example | additive | read the tests | -| 2 | [Named merge slots](#2-named-merge-slots) | medium | additive if done carefully | rely on group order | -| 3 | [Several named graph inputs](#3-several-named-graph-inputs) | medium | additive (new graph shape) | `std::variant` input + select stages | -| 4 | [Stage labels and typed access](#4-stage-labels-and-typed-access) | medium | additive (DSL syntax) | side registry filled by creator lambdas | -| 5 | [Track which stage short-circuited](#5-track-which-stage-short-circuited) | medium | additive | tap the edge, or log in the stage | -| 6 | [Config-aware `registerMergeFilter`](#6-config-aware-registermergefilter) | low | additive | subclass + `FilterRegistrar` creator | -| 7 | [Injectable registry, duplicate detection](#7-injectable-registry-duplicate-detection) | low | mostly additive | unique names | -| 8 | [Documentation: 0..n outputs, large messages](#8-documentation-0n-outputs-large-messages) | doc only | — | — | -| 9 | [Known limitations to lift](#9-known-limitations-to-lift) | low–medium | additive | JSON config; read `GraphOutputs` carefully | - -Two earlier items of the same list are done and therefore not repeated here: +| 1 | [Named merge slots](#1-named-merge-slots) | medium | additive if done carefully | rely on group order | +| 2 | [Several named graph inputs](#2-several-named-graph-inputs) | medium | additive (new graph shape) | `std::variant` input + select stages | +| 3 | [Stage labels and typed access](#3-stage-labels-and-typed-access) | medium | additive (DSL syntax) | side registry filled by creator lambdas | +| 4 | [Track which stage short-circuited](#4-track-which-stage-short-circuited) | medium | additive | tap the edge, or log in the stage | +| 5 | [Config-aware `registerMergeFilter`](#5-config-aware-registermergefilter) | low | additive | subclass + `FilterRegistrar` creator | +| 6 | [Injectable registry, duplicate detection](#6-injectable-registry-duplicate-detection) | low | mostly additive | unique names | +| 7 | [Documentation: 0..n outputs, large messages](#7-documentation-0n-outputs-large-messages) | doc only | — | — | +| 8 | [Known limitations to lift](#8-known-limitations-to-lift) | low–medium | additive | JSON config; read `GraphOutputs` carefully | + +Items of the same list that are done, and therefore not repeated here: **type-checked merge inputs** (`TypedMergeFilter` / `UniformMergeFilter` / -`registerTypedMergeFilter`) and the **end-of-stream hook** -(`MessageFilter::finish()`). Both are described in the README. +`registerTypedMergeFilter`), the **end-of-stream hook** +(`MessageFilter::finish()`), both described in the README, and **examples for +the newer features** — `apps/statefulPipeline` and `apps/compositePipeline`, +walked through in [EXAMPLE.md](EXAMPLE.md) sections 8 and 9. --- -## 1. More examples for the newer features - -**Today.** [`apps/textPipeline`](apps/textPipeline/main.cpp) and -[EXAMPLE.md](EXAMPLE.md) cover the compile-time `FilterGraph`, registering -stages, a DSL graph with a tap, fan-in with an untyped and a typed merge, -several named outputs, `validateDslGraph` and the JSON format. Several features -that shipped since have no runnable example: `GraphContext`, -`MessageFilter::finish()`, a merge with per-instance state, `JoinFilter`, a -nested graph used as a stage, a `Void` sink, and the in-band tick pattern that -stands in for a `tick()` hook. - -**Gap.** Those are exactly the features that come up once stages hold state, -and the hardest to get right from a reference description alone. Their only -executable documentation today is the test suite -([`GraphContextTests.cpp`](tests/FilterGraphTests/GraphContextTests.cpp), -[`LifecycleTests.cpp`](tests/FilterGraphTests/LifecycleTests.cpp)), which reads -as assertions rather than as a walkthrough. - -**Proposal.** One runnable app per theme, each in the style of `textPipeline` -(numbered blocks, comments that explain the *why*), plus a matching EXAMPLE.md -section per app: - -- `apps/statefulPipeline` — a stage that accumulates across messages, flushes - in `finish()` and publishes its result through the `GraphContext`; a merge - with per-instance state; a derived application context recovered with - `as()`. -- `apps/compositePipeline` — `JoinFilter` from JSON next to the equivalent DSL - merge; a `DslFilterGraph` registered as a stage of an outer graph (showing - that `finish()` and the context propagate into it); a `Void`-terminated sink; - an in-band `Tick` alternative in a variant input type. - -**Compatibility.** Additive: new targets under `apps/`, no library change. - -**Workaround today.** The tests, plus the README sections -[Graph context](README.md#graph-context) and -[Ending a run](README.md#ending-a-run). - -## 2. Named merge slots +## 1. Named merge slots **Today.** Slots are positional, in the order the DSL group lists them. A merge stage can only know which upstream a slot came from by convention. Since @@ -114,7 +78,7 @@ is not needed. **Workaround today.** Rely on group order, and give same-typed slots distinct wrapper types so that the existing slot-type check can tell them apart. -## 3. Several named graph inputs +## 2. Several named graph inputs **Today.** A graph has exactly one input edge `in`, with one type. A graph that consumes several kinds of message needs a single `std::variant` input, and then @@ -149,7 +113,7 @@ keeps the single `in` edge. **Workaround today.** A `std::variant` input plus one select stage per alternative, each dropping the alternatives it does not handle. -## 4. Stage labels and typed access +## 3. Stage labels and typed access **Today.** The owner of a `DslFilterGraph` cannot reach a stage instance. The DSL has no labels (`#` starts a comment), and the compiled plan is private. @@ -179,7 +143,7 @@ auto& stats = graph.stage("stats"); // throws if unknown or the every instance it builds in a side registry, or have the stage publish what the owner needs through the `GraphContext`. -## 5. Track which stage short-circuited +## 4. Track which stage short-circuited **Today.** When a graph drops a message, neither `DslFilterGraph` nor `AnyFilterChain` tells the caller *which* stage returned `std::nullopt`. @@ -200,7 +164,7 @@ so that variant needs an overload or an opt-in. **Workaround today.** Tap the suspect edge with a logging stage ending in `end`, or log inside the stage that decides to drop. -## 6. Config-aware `registerMergeFilter` +## 5. Config-aware `registerMergeFilter` **Today.** `registerMergeFilter(name, combiner)` ignores the DSL config and **copies one combiner into every instance**. A combiner lambda that captures a @@ -221,15 +185,16 @@ void registerMergeFilter(const std::string& name, std::function::Combiner(const nlohmann::json& config)> factory); ``` -Plus a short EXAMPLE.md section on "stateful merges: subclass + -`FilterRegistrar`" (see item 1), and a note in the README on the copy semantics -of the existing overload. +[EXAMPLE.md](EXAMPLE.md) section 8 now shows the subclass + `FilterRegistrar` +pattern and the shared-combiner semantics; what is still missing is the +overload above, and a note in the README on the copy semantics of the existing +one. **Compatibility.** Additive overload. **Workaround today.** Subclass and register with a creator lambda. -## 7. Injectable registry, duplicate detection +## 6. Injectable registry, duplicate detection **Today.** `FilterRegistry::instance()` is a process-wide singleton, and `registerFilter` **silently overwrites** an existing name. @@ -258,7 +223,7 @@ Ship it behind a transition: warn first, or add **Workaround today.** Keep names unique, and register test fakes under their own names. -## 8. Documentation: 0..n outputs, large messages +## 7. Documentation: 0..n outputs, large messages No API change. These points belong in the README / EXAMPLE.md, because they come up as soon as stages carry state: @@ -275,7 +240,7 @@ up as soon as stages carry state: its members persists across messages. Say explicitly that this is supported and intended, and that each graph construction creates fresh instances. -## 9. Known limitations to lift +## 8. Known limitations to lift The [current limitations](README.md#current-limitations) the README lists, as work items: diff --git a/apps/CMakeLists.txt b/apps/CMakeLists.txt index 8f84c96..9ed5350 100644 --- a/apps/CMakeLists.txt +++ b/apps/CMakeLists.txt @@ -1 +1,3 @@ add_subdirectory(textPipeline) +add_subdirectory(statefulPipeline) +add_subdirectory(compositePipeline) diff --git a/apps/compositePipeline/CMakeLists.txt b/apps/compositePipeline/CMakeLists.txt new file mode 100644 index 0000000..5d54f94 --- /dev/null +++ b/apps/compositePipeline/CMakeLists.txt @@ -0,0 +1,23 @@ +project(compositePipeline) + +set(CMAKE_CXX_STANDARD 20) +file(GLOB_RECURSE HEADER_FILES CONFIGURE_DEPENDS "*.h*") +file(GLOB_RECURSE CPP_FILES CONFIGURE_DEPENDS "*.cpp") + +add_executable(${PROJECT_NAME} ${HEADER_FILES} ${CPP_FILES} ) + + + +target_link_libraries(${PROJECT_NAME} + PUBLIC + filterGraph::filterGraph +) +target_include_directories(${PROJECT_NAME} + PRIVATE + $ + $ +) + +enable_coverage(${PROJECT_NAME}) + +install(TARGETS ${PROJECT_NAME}) \ No newline at end of file diff --git a/apps/compositePipeline/main.cpp b/apps/compositePipeline/main.cpp new file mode 100644 index 0000000..720b600 --- /dev/null +++ b/apps/compositePipeline/main.cpp @@ -0,0 +1,297 @@ +// Putting graphs together out of graphs — the composition features: +// - JoinFilter: scatter one message through JSON-configured paths and combine +// their results in C++, next to the equivalent DSL fan-out + merge +// - registerTypedMergeFilter: a typed merge registered from a lambda +// - a DslFilterGraph registered as a stage of an outer graph, with finish() +// and the GraphContext propagating into it +// - Void: a graph that only has side effects, with no consumable output +// - the in-band tick message, the pattern that stands in for a tick() hook +// +// See EXAMPLE.md, "Composition: joins, nested graphs, sinks and ticks". +#include +#include +#include +#include +#include +#include +#include +#include + +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +using filterGraph::DslFilterGraph; +using filterGraph::FilterRegistrar; +using filterGraph::GraphContext; +using filterGraph::JsonFilterGraph; +using filterGraph::MessageFilter; +using filterGraph::Void; +using filterGraph::registerJoinFilter; +using filterGraph::registerTypedMergeFilter; + +namespace { + +// --- Messages and context values --------------------------------------- + +// A tick alternative in the input type is how a time-driven stage is fed: +// there is no tick() hook, because time is domain-specific (event time, wall +// clock, a sensor clock). Ticks travel through the graph like any other +// message, so the same graph shape handles both. +struct Tick +{ +}; + +using Event = std::variant; + +// Published by the application, read by a stage inside the nested graph. +struct Tag +{ + std::string value; +}; + +// --- Stages ------------------------------------------------------------ + +class UpperFilter : public MessageFilter +{ +public: + std::optional filter(std::string&& text) override + { + std::ranges::transform(text, text.begin(), [](unsigned char c) { return static_cast(std::toupper(c)); }); + return std::move(text); + } +}; + +class ReverseFilter : public MessageFilter +{ +public: + std::optional filter(std::string&& text) override + { + std::ranges::reverse(text); + return std::move(text); + } +}; + +// Prefixes the tag the application put into the context. This stage sits +// inside the nested graph, so seeing the tag proves that the outer graph's +// context reached it. +class TagFilter : public MessageFilter +{ +public: + std::optional filter(std::string&& text) override + { + return std::format("{}{}", context().getOr(Tag{" "}).value, text); + } +}; + +// Counts what passes through and reports when the stream ends. It lives inside +// the nested graph, so its report shows that finish() reached the inner stages. +class CountFilter : public MessageFilter +{ +public: + std::optional filter(std::string&& text) override + { + ++mSeen; + return std::move(text); + } + + void finish() override + { + std::cout << std::format("[nested] the inner stage saw {} message(s)\n", mSeen); + } + +private: + std::size_t mSeen = 0; +}; + +// A sink: Void says "this path deliberately ends here", which is distinct from +// returning std::nullopt ("this message was dropped"). A Void stage must be the +// last one in its path. +class WriteFilter : public MessageFilter +{ +public: + std::optional filter(std::string&& text) override + { + std::cout << "[sink] " << text << '\n'; + return Void{}; + } +}; + +// A time-driven stage, fed by in-band ticks: it buffers the lines it receives +// and emits the batch when a Tick arrives. Between ticks it returns +// std::nullopt, so nothing downstream runs. +class BatchFilter : public MessageFilter +{ +public: + std::optional filter(Event&& event) override + { + if (const auto* line = std::get_if(&event)) + { + mBatch.push_back(*line); + return std::nullopt; // buffered; nothing to emit yet + } + + if (mBatch.empty()) + { + return std::nullopt; // a tick with nothing buffered + } + + std::string batch; + for (const auto& line : mBatch) + { + batch += (batch.empty() ? "" : ", ") + line; + } + mBatch.clear(); + return batch; + } + + void finish() override + { + if (!mBatch.empty()) + { + std::cout << std::format("[batch] {} line(s) left unflushed at the end of the stream\n", mBatch.size()); + } + } + +private: + std::vector mBatch; +}; + +// --- Registration ------------------------------------------------------ + +static FilterRegistrar registerUpper("Upper"); +static FilterRegistrar registerReverse("Reverse"); +static FilterRegistrar registerTag("Tag"); +static FilterRegistrar registerCount("Count"); +static FilterRegistrar registerWrite("Write"); +static FilterRegistrar registerBatch("Batch"); + +// A DslFilterGraph is itself a MessageFilter, so a whole graph can be +// registered as a stage and used inside another graph. +static FilterRegistrar> registerInner("Inner", [](const nlohmann::json&) { + return std::make_shared>("in -> Tag -> tagged -> Count -> out"); +}); + +// A typed merge registered from a lambda: it declares one slot type per +// argument, so the DSL checks the group's edges when the graph is built. +static const bool sRegisterConcat = [] { + registerTypedMergeFilter( + "Concat", + [](std::optional&& upper, std::optional&& reversed) -> std::optional { + return std::format("{} | {}", upper.value_or("-"), reversed.value_or("-")); + }); + return true; +}(); + +// The JSON counterpart of that fan-out + merge: a Join scatters a copy of the +// message through each configured path and hands the gathered results to a +// combiner written in C++ (paths come from configuration, the combiner cannot). +static const bool sRegisterJoin = [] { + registerJoinFilter("Join", [](std::vector&& slots) -> std::optional { + std::string joined; + for (auto& slot : slots) + { + joined += (joined.empty() ? "" : " | "); + joined += slot.has_value() ? std::any_cast(std::move(slot)) : std::string{"-"}; + } + return joined; + }); + return true; +}(); + +} // namespace + +int main() +{ + // 1) Fan-out and merge in the DSL: both statements read `in`, so each gets + // its own copy, and the typed merge combines the two results. + { + DslFilterGraph graph(R"dsl( + in -> Upper -> upper + in -> Reverse -> reversed + (upper, reversed) -> Concat -> out + )dsl"); + + std::cout << "[dsl] " << *graph.filter(std::string{"compose me"}) << '\n'; + } + + // 2) The same shape in JSON, where the branching lives in a Join stage's + // config. The DSL expresses it as graph structure; JSON needs a + // composite stage, and the combiner is the same kind of C++ callable. + { + JsonFilterGraph graph(nlohmann::json::parse(R"json([ + { "type": "Join", "config": { "paths": [ + [ { "type": "Upper" } ], + [ { "type": "Reverse" } ] + ] } } + ])json")); + + std::cout << "[json] " << *graph.filter(std::string{"compose me"}) << "\n\n"; + } + + // 3) A nested graph as a stage. The outer graph hands its context to the + // stage — including into the inner graph's own stages — and finishing + // the outer graph finishes the inner ones too. + { + DslFilterGraph graph("in -> Inner -> inner -> Upper -> out"); + graph.context().set(Tag{"[tagged] "}); + + std::cout << "[outer] " << *graph.filter(std::string{"nested graphs compose"}) << '\n'; + std::cout << "[outer] " << *graph.filter(std::string{"and share a context"}) << '\n'; + + graph.finish(); // reaches the Count stage inside the nested graph + std::cout << '\n'; + } + + // 4) A graph with no consumable output: every path ends in a sink, so the + // graph's output type is Void. filter() returns a Void value (the graph + // ran), not a message. + { + DslFilterGraph graph(R"dsl( + in -> Upper -> upper -> Write -> end + in -> Reverse -> reversed -> Write -> end + )dsl"); + + const auto ran = graph.filter(std::string{"side effects only"}); + std::cout << std::format("[sink] the graph ran: {}\n\n", ran.has_value()); + } + + // 5) In-band ticks: the input type is a variant of the payload and a Tick, + // so a stage that has to act while no data arrives is driven by messages + // like every other stage. Between ticks the graph produces nothing. + { + DslFilterGraph graph("in -> Batch -> out"); + + const std::vector stream{ + std::string{"first"}, + std::string{"second"}, + Tick{}, + std::string{"third"}, + Tick{}, + std::string{"unflushed"}, + }; + + for (auto event : stream) + { + if (auto batch = graph.filter(std::move(event))) + { + std::cout << "[batch] " << *batch << '\n'; + } + } + + graph.finish(); // reports what is still buffered + } + + return 0; +} diff --git a/apps/statefulPipeline/CMakeLists.txt b/apps/statefulPipeline/CMakeLists.txt new file mode 100644 index 0000000..b443876 --- /dev/null +++ b/apps/statefulPipeline/CMakeLists.txt @@ -0,0 +1,23 @@ +project(statefulPipeline) + +set(CMAKE_CXX_STANDARD 20) +file(GLOB_RECURSE HEADER_FILES CONFIGURE_DEPENDS "*.h*") +file(GLOB_RECURSE CPP_FILES CONFIGURE_DEPENDS "*.cpp") + +add_executable(${PROJECT_NAME} ${HEADER_FILES} ${CPP_FILES} ) + + + +target_link_libraries(${PROJECT_NAME} + PUBLIC + filterGraph::filterGraph +) +target_include_directories(${PROJECT_NAME} + PRIVATE + $ + $ +) + +enable_coverage(${PROJECT_NAME}) + +install(TARGETS ${PROJECT_NAME}) \ No newline at end of file diff --git a/apps/statefulPipeline/main.cpp b/apps/statefulPipeline/main.cpp new file mode 100644 index 0000000..09a581d --- /dev/null +++ b/apps/statefulPipeline/main.cpp @@ -0,0 +1,292 @@ +// Stages that carry state across messages — the features that come up as soon +// as a graph does more than transform one message at a time: +// - a stateful stage: state in its members, flushed in finish() +// - GraphContext: how a stage hands a final result back to the application +// - a merge stage with per-instance state (subclass + FilterRegistrar) +// - the shared-state semantics of registerMergeFilter's combiner +// - a derived application context, recovered with GraphContext::as() +// +// See EXAMPLE.md, "Stateful stages: finish() and the graph context". +#include +#include +#include +#include +#include + +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +using filterGraph::DslFilterGraph; +using filterGraph::FilterRegistrar; +using filterGraph::GraphContext; +using filterGraph::MergeInputs; +using filterGraph::MessageFilter; +using filterGraph::TypedMergeFilter; +using filterGraph::registerMergeFilter; + +namespace { + +// --- Values shared through the context --------------------------------- + +// A context value is keyed by its own type, so it is a dedicated struct rather +// than a bare std::size_t that unrelated stages would silently share. +struct Summary +{ + std::size_t lines{}; + std::size_t words{}; +}; + +// An application context: a GraphContext with members of its own. Stages that +// need the session id ask for it via context().as(). +class AppContext : public GraphContext +{ +public: + explicit AppContext(std::string session) + : sessionId(std::move(session)) + { + } + + std::string sessionId; +}; + +std::size_t countWords(std::string_view line) +{ + std::size_t words = 0; + bool inWord = false; + for (const char c : line) + { + const bool isSpace = std::isspace(static_cast(c)) != 0; + if (!isSpace && !inWord) + { + ++words; + } + inWord = !isSpace; + } + return words; +} + +// --- Stages ------------------------------------------------------------ + +// A stateful stage: one instance lives as long as its graph, so its members +// accumulate across messages. It passes each line through unchanged and only +// reports when the stream ends: finish() is that point. The result goes into +// the graph context, because finish() produces no message. +class CollectFilter : public MessageFilter +{ +public: + std::optional filter(std::string&& line) override + { + ++mSummary.lines; + mSummary.words += countWords(line); + return std::move(line); + } + + void finish() override + { + std::cout << std::format("[collect] end of stream: {} lines, {} words\n", mSummary.lines, mSummary.words); + context().set(mSummary); // the application reads it after finish() + } + +private: + Summary mSummary; +}; + +// Stamps each line with the session id of the application's own context. A +// stage using as() depends on that type, so reusable stages should +// stick to the type-keyed set/get instead. +class StampFilter : public MessageFilter +{ +public: + std::optional filter(std::string&& line) override + { + const auto* app = context().as(); + return app ? std::format("[{}] {}", app->sessionId, line) : std::move(line); + } +}; + +class LengthFilter : public MessageFilter +{ +public: + std::optional filter(std::string&& line) override + { + return line.size(); + } +}; + +class WordsFilter : public MessageFilter +{ +public: + std::optional filter(std::string&& line) override + { + return countWords(line); + } +}; + +// A merge stage with per-instance state and per-instance configuration: it +// keeps a sliding window of the lengths it has seen. This is the way to a +// stateful merge — subclass (here TypedMergeFilter, so the DSL still checks +// the slot types) and register a creator that reads the stage's arguments, so +// that every instance gets its own state. +class TrendFilter : public TypedMergeFilter +{ +public: + explicit TrendFilter(std::size_t window) + : mWindow(window) + { + } + + std::optional merge(std::optional&& length, std::optional&& words) override + { + mLengths.push_back(length.value_or(0)); + if (mLengths.size() > mWindow) + { + mLengths.pop_front(); + } + + const double average = + std::accumulate(mLengths.begin(), mLengths.end(), 0.0) / static_cast(mLengths.size()); + return std::format("chars={} words={} avg(last {})={:.1f}", length.value_or(0), words.value_or(0), mWindow, average); + } + +private: + std::size_t mWindow; + std::deque mLengths; +}; + +// --- Registration ------------------------------------------------------ + +static FilterRegistrar registerCollect("Collect"); +static FilterRegistrar registerStamp("Stamp"); +static FilterRegistrar registerLength("Length"); +static FilterRegistrar registerWords("Words"); + +// A creator lambda gives every instance its own state, configured from the +// stage's arguments: `Trend(window=2)`. +static FilterRegistrar registerTrend("Trend", [](const nlohmann::json& config) { + return std::make_shared(config.value("window", std::size_t{3})); +}); + +// By contrast, registerMergeFilter copies ONE combiner into every instance, so +// state the combiner captures is shared by all of them — across instances and +// across graphs. That is what block 3 demonstrates; a merge that needs its own +// state uses the Trend pattern above instead. +static const bool sRegisterTally = [] { + auto calls = std::make_shared(0); + registerMergeFilter("Tally", [calls](MergeInputs&& inputs) -> std::optional { + ++*calls; + return std::format("call {} of the one shared combiner ({} slots)", *calls, inputs.size()); + }); + return true; +}(); + +const std::vector& lines() +{ + static const std::vector input{ + "the quick brown fox", + "jumps over", + "the lazy dog and keeps running", + }; + return input; +} + +} // namespace + +int main() +{ + // 1) A stateful stage, flushed at the end of the stream. Nothing is + // reported while messages flow; finish() reports once and publishes the + // summary to the graph's context, where the application picks it up. + { + DslFilterGraph graph("in -> Collect -> out"); + + for (auto line : lines()) + { + graph.filter(std::move(line)); + } + + graph.finish(); // the owner's call: a graph that is never finished never flushes + + const auto summary = graph.context().get(); + std::cout << std::format("[app] read from the context: {} lines, {} words\n\n", + summary->lines, + summary->words); + } + + // 2) Two instances of the same stateful merge stage, with different + // arguments. Each has its own sliding window, because the registered + // creator builds a fresh TrendFilter per instance. + { + DslFilterGraph graph(R"dsl( + in -> Length -> length + in -> Words -> words + (length, words) -> Trend(window=2) -> out.short + (length, words) -> Trend(window=3) -> out.long + )dsl"); + + for (auto line : lines()) + { + auto outputs = graph.filter(std::move(line)); + std::cout << std::format("[trend] short: {}\n[trend] long: {}\n", + *outputs->get("short"), + *outputs->get("long")); + } + std::cout << '\n'; + } + + // 3) The same graph shape with a merge registered from a combiner lambda. + // Both stage instances run the same captured counter, so the numbers + // keep climbing across instances — the state is shared, not per stage. + { + DslFilterGraph graph(R"dsl( + in -> Length -> length + in -> Words -> words + (length, words) -> Tally -> out.first + (length, words) -> Tally -> out.second + )dsl"); + + for (auto line : lines()) + { + auto outputs = graph.filter(std::move(line)); + std::cout << std::format("[tally] {} / {}\n", + *outputs->get("first"), + *outputs->get("second")); + } + std::cout << '\n'; + } + + // 4) An application context: derive from GraphContext, hand it to the graph + // (not to individual stages — the graph overwrites a stage's context + // with its own), and recover it inside a stage with as(). + // Collect's summary lands in the very same context. Stamp runs first + // here, so the session stamp counts as a word of every line. + { + DslFilterGraph graph("in -> Stamp -> stamped -> Collect -> out"); + + auto context = std::make_shared("session-42"); + graph.setContext(context); + + for (auto line : lines()) + { + std::cout << "[stamped] " << *graph.filter(std::move(line)) << '\n'; + } + + graph.finish(); + std::cout << std::format("[app] session {} saw {} lines\n", + context->sessionId, + context->get()->lines); + } + + return 0; +}