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; +}