Skip to content

feat: add finish(), the end-of-stream lifecycle hook - #11

Merged
psyinf merged 1 commit into
mainfrom
feature/lifecycle-finish
Sep 17, 2026
Merged

psyinf merged 1 commit into
mainfrom
feature/lifecycle-finish

Conversation

@psyinf

@psyinf psyinf commented Sep 17, 2026

Copy link
Copy Markdown
Owner

Stack (bottom → top)

  1. feat: add GraphContext, a graph-scoped side channel for stages #9GraphContext
  2. feat: type-checked merge inputs #10 — type-checked merge inputs
  3. feat: add finish(), the end-of-stream lifecycle hook #11finish() lifecycle hook (this PR)

Summary

A stage acted only when a message reached it, so a stage that accumulates something (statistics, a batch, an open file) had no defined point at which to flush it. The alternatives were both weak: a destructor (order across a graph is implicit and exceptions cannot propagate) or an in-band end-of-stream message that every stage type has to understand and pass on.

MessageFilter::finish() is that point. It is called once after the last message, defaults to a no-op, and produces no message.

DslFilterGraph<Record, Void> graph(text);
for (auto&& record : records) { graph.filter(std::move(record)); }
graph.finish(); // finishes every stage, in run order

Behaviour

  • Forwarded everywhere: FilterGraph, DslFilterGraph, JsonFilterGraph, AnyFilterChain, FanoutFilter, JoinFilter and nested graphs pass it to their stages.
  • Run order: DslFilterGraph finishes its stages in run order, so a stage is finished after the stages it reads from.
  • Every stage is finished even if one throws (detail::FinishScope); the first exception is rethrown once the others have run. One failing stage cannot keep the rest from flushing.
  • No output: finish() returns nothing, so nothing is routed downstream — which would need new merge semantics (every other slot a hole). A stage that wants to hand a final result to the application publishes it through the GraphContext. The name leaves the door open for a flush()-style variant later.
  • Owner's decision: a graph that is never finished never flushes, and finishing twice finishes every stage twice. Neither is policed.

Breaking change

AnyMessageFilter::finish() is pure virtual, like setContext in #9: a composite that forgot to forward it would otherwise silently fail to flush its inner stages. Custom AnyMessageFilter implementations must implement it. (Say the word if you'd rather have a no-op default here — it is a one-line change.)

No tick()

Deliberately left out. Time is domain-specific (event time, wall clock, a sensor clock), so a generic signature would fit few users. The README documents the in-band tick message instead: a Tick alternative in the graph's input type, fed through the graph like any other message.

Testing

New LifecycleTests (6 cases): run order in a DSL graph, a nested graph forwarding to its own stages, the compile-time FilterGraph, a JsonFilterGraph with fanout branches and join paths, every stage finishing when one throws, and finishing twice. Full suite: 83/83 locally (MSVC debug, ASan).

@psyinf
psyinf added this pull request to stack #12 September 17, 2026 17:44
@psyinf
psyinf force-pushed the feature/lifecycle-finish branch from cf2c2f0 to 7269745 Compare September 17, 2026 18:15
@psyinf
psyinf force-pushed the feature/lifecycle-finish branch from 7269745 to 3dd2804 Compare September 17, 2026 18:22
Base automatically changed from feature/typed-merge-inputs to main September 17, 2026 18:26
A stage only acted when a message reached it, so a stage holding state
(statistics, a batch, an open file) had no defined point to flush it;
the alternatives were destructors or an in-band end-of-stream message.

- MessageFilter::finish(): called once after the last message, defaults
  to a no-op, produces no message
- DslFilterGraph::finish() finishes every stage in run order, so a stage
  is finished after the stages it reads from
- FilterGraph, JsonFilterGraph, AnyFilterChain, FanoutFilter, JoinFilter
  and nested graphs forward it
- detail::FinishScope: every stage is finished even if one throws, and
  the first exception is rethrown afterwards
- breaking: AnyMessageFilter::finish() is pure virtual, like setContext,
  so a composite cannot silently fail to forward it

No tick(): time is domain-specific, so the README documents the in-band
tick message instead.
@psyinf
psyinf force-pushed the feature/lifecycle-finish branch from 3dd2804 to 24ee99b Compare September 17, 2026 18:26
@psyinf
psyinf merged commit bfd89ab into main Sep 17, 2026
3 checks passed
@psyinf
psyinf deleted the feature/lifecycle-finish branch September 17, 2026 18:31
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant