Class StreamEmissionEngine

java.lang.Object
org.ek9lang.compiler.phase7.generator.AbstractGenerator
org.ek9lang.compiler.phase7.generator.AbstractStreamStageGenerator
org.ek9lang.compiler.phase7.generator.StreamEmissionEngine
final class StreamEmissionEngine extends AbstractStreamStageGenerator
The buffered-emission engine: the stages that cannot stream, and the pipeline bodies they emit.

group, split, sort and async must see the whole source before they can emit anything, so rather than wrapping the loop body they RESTRUCTURE the pipeline - draining pre-loop into a buffer, then emitting from it through whatever stages remain.

These live together because they are MUTUALLY RECURSIVE and cannot be separated: a buffering stage may contain another downstream (sort|group, group|split), so each can re-enter the others. A dependency analysis of the original 5141-line generator found them as a single seven-method cycle; splitting them across classes would only make that cycle circular between classes instead of within one.

Depends on StreamStageDispatcher and StreamPipelineMachinery and is depended upon by the orchestrator - a one-way chain. Both collaborators are passed in rather than constructed, so the whole pipeline shares one dispatcher and one machinery instance.

  • Constructor Details

  • Method Details

    • applyAsyncBufferSortEmit

      void applyAsyncBufferSortEmit(String asyncExecutorVar, ISymbol asyncProducesType, List<EK9Parser.StreamPartContext> preSortStages, EK9Parser.StreamPartContext sortStage, List<EK9Parser.StreamPartContext> postSortStages, TerminalSetup terminalSetup, org.antlr.v4.runtime.tree.ParseTree probeSite, DebugInfo debugInfo, List<IRInstr> instructions)
      Async-specific entry to bufferSortEmit(HelperSource, ISymbol, List, EK9Parser.StreamPartContext, List, TerminalSetup, ParseTree, DebugInfo, List): drains the async executor (hasNext/next); its pipeline's finally region closes it.

    • applyAsyncToCollection

      boolean applyAsyncToCollection(BufferedStageParams params)
      Apply async concurrent execution to a collection, following the GROUP/SPLIT two-phase pattern.

      ASYNC submits each function item for concurrent execution on virtual threads, then drains results in COMPLETION order through post-ASYNC stages to the terminal (so async | head N yields the first N to complete). A head after the async with nothing between them that can drop an item caps what is submitted (AsyncCappingHeads, decision 15 d).

      Two-phase architecture:

      1. SUBMIT: iterate source, apply pre-ASYNC stages, submit each item to executor
      2. DRAIN: iterate results as they complete, apply post-ASYNC stages, pipe to terminal
      Returns:
      true if ASYNC was found and handled, false otherwise

    • applyGroupToCollection

      boolean applyGroupToCollection(BufferedStageParams params)
      Apply consecutive-key grouping (GROUP) to a collection. The group boundary fires when the key changes, where the key is the item itself or keyFn(item) for "| group by keyFn".
      Returns:
      true if a GROUP stage was found and handled, false otherwise

    • applyGroupingStage

      void applyGroupingStage(StreamSource source, ISymbol consumedElementType, BucketTracker tracker, StageGroup preStageGroup, StageGroup postStageGroup, BoundaryConditionBuilder boundaryBuilder, TerminalSetup terminalSetup, org.antlr.v4.runtime.tree.ParseTree probeSite, List<IRInstr> instructions, DebugInfo debugInfo)
      Shared drain+emit engine for the consecutive-grouping stream stages (GROUP, SPLIT). Drains the source iterator into a current-group List, emitting the group via tracker.addGroup and starting a fresh one whenever boundaryBuilder signals a boundary, then iterates the stored groups through the post-stage pipeline to the terminal. GROUP and SPLIT differ ONLY in how the per-item boundary is computed (the supplied callback).

    • applySortToCollection

      SortSetup applySortToCollection(List<EK9Parser.StreamPartContext> stages, IteratorSetup iteratorSetup, ISymbol elementType, IAggregateSymbol collectionType, org.antlr.v4.runtime.tree.ParseTree probeSite, DebugInfo debugInfo, List<IRInstr> instructions)

    • applySplitToCollection

      boolean applySplitToCollection(BufferedStageParams params)
      Apply split-by-predicate (SPLIT) to a collection, mirroring applyGroupToCollection(BufferedStageParams). The group boundary fires when the boundary predicate returns true for an item.
      Returns:
      true if a SPLIT stage was found and handled, false otherwise

    • applyWindowedAsyncToCollection

      void applyWindowedAsyncToCollection(List<EK9Parser.StreamPartContext> stages, int asyncIndex, List<EK9Parser.StreamPartContext> preAsyncStages, List<StageState> preAsyncStageStates, IteratorSetup iteratorSetup, ISymbol elementType, ISymbol asyncProducesType, String asyncExecutorVar, TerminalSetup terminalSetup, org.antlr.v4.runtime.tree.ParseTree probeSite, DebugInfo debugInfo, List<IRInstr> instructions, List<StageState> postAsyncStates)
      Step C — the interleaved WINDOWED async driver for an unbounded source (the telemetry case cat udp | async | filter | head N | collect). The eager submit-then-drain drains the whole source first, which never terminates on an unbounded source; this keeps at most W submissions in-flight and interleaves submit and drain so a post-async head N can stop it.

      The heads after the async (PullBoundingHeads) gate BOTH the drain and the refill condition, counters first (StreamDemand): once a head has its N items no further result is taken and no further item is pulled or submitted (Contract 12, "after the demand is met"), so a task whose result is not needed can never fail the statement. Every other stream loop is gated the same way.

      No runtime change — uses the existing executor (submit/next/hasNext/cancel/close). Bounded async, a pre-async head, and a post-async buffer-barrier all keep the eager path (routed in applyAsyncToCollection(BufferedStageParams)); a post-async barrier on an unbounded async is rejected at phase 3.

    • bufferSortEmit

      void bufferSortEmit(HelperSource source, ISymbol bucketType, List<EK9Parser.StreamPartContext> preSortStages, EK9Parser.StreamPartContext sortStage, List<EK9Parser.StreamPartContext> postSortStages, TerminalSetup terminalSetup, org.antlr.v4.runtime.tree.ParseTree probeSite, DebugInfo debugInfo, List<IRInstr> instructions)

    • buildAsyncSubmitAction

      StageWrapResult buildAsyncSubmitAction(String submitItemVar, ISymbol elementType, String asyncExecutorVar, String inFlightVar, String oneVar, ISymbol integerType, List<EK9Parser.StreamPartContext> preAsyncStages, List<StageState> preAsyncStageStates, org.ek9lang.compiler.phase7.generator.StreamEmissionEngine.AsyncCap cap, DebugInfo debugInfo)
      The per-item SUBMIT action shared by the windowed prime loop and the drain refill: executor.submit(item); inFlight++, wrapped in the pre-ASYNC stages (filter/map/etc.).

    • buildWindowRefill

      List<IRInstr> buildWindowRefill(IteratorSetup iteratorSetup, boolean isRawIterator, String submitItemVar, ISymbol elementType, String asyncExecutorVar, String inFlightVar, String oneVar, ISymbol integerType, List<EK9Parser.StreamPartContext> preAsyncStages, List<StageState> preAsyncStageStates, List<StageState.Counter> postHeads, org.ek9lang.compiler.phase7.generator.StreamEmissionEngine.AsyncCap cap, DebugInfo debugInfo)
      The windowed drain's REFILL step: if ([c1 < l1 AND ...] source.hasNext()) { bind submitItem = next(); [pre-async stages]; submit(submitItem); inFlight++ }. Keeps the in-flight window topped up as results drain, until a post-async head (postHeads) has its items; with none (the perpetual case) the guard is hasNext() alone. The bind + any raw-iterator cast register into the surrounding drain-body scope (per-iteration).

    • continueOverIterator

      void continueOverIterator(List<EK9Parser.StreamPartContext> stages, IteratorSetup sourceIter, ISymbol sourceElementType, TerminalSetup terminalSetup, org.antlr.v4.runtime.tree.ParseTree probeSite, DebugInfo debugInfo, List<IRInstr> instructions, boolean sourceUnbounded)
      THE pipeline driver over a terminal that is already resolved - the buffers' own re-entry: see the overload below. A materialised bucket stream's loops are probed at the pipeline's source, probeSite (D88).

    • continueOverIterator

      void continueOverIterator(List<EK9Parser.StreamPartContext> stages, IteratorSetup sourceIter, ISymbol sourceElementType, Function<List<IRInstr>, TerminalSetup> terminalResolver, boolean sourceUnbounded, org.antlr.v4.runtime.tree.ParseTree probeSite, DebugInfo debugInfo, List<IRInstr> instructions)
      THE pipeline driver over a source iterator: emit a stream given as sourceIter through stages to the terminal, with NO composition limit. Drains any leading sorts, then dispatches the FIRST collection buffer (GROUP/SPLIT/ASYNC) - each of which, when ITS downstream contains a further buffer, materialises its output as an iterator and calls back here - or streams the remaining per-item stages to the terminal. Because no stage encodes what follows it, arbitrary orderings/depths of SORT/GROUP/SPLIT/ASYNC (plus per-item stages) compose.

      The top-level single-source pipeline (StreamStatementGenerator.generateSingleSourcePipelineCore) is this driver over the source's own iterator; the buffers re-enter it over the iterator they materialised. terminalResolver is called exactly once, after the leading sorts and before the pipeline scope opens - a statement's terminal sink is constructed at that position (appending to the list it is given); a resolved terminal ignores it. probeSite - the pipeline's source as written - places the LOOP_BODY coverage probe of every loop the pipeline makes, the re-entries over a materialised bucket stream included (D88).

    • generateConditionAndBody

      ConditionAndPostLoop generateConditionAndBody(org.antlr.v4.runtime.tree.ParseTree probeSite, List<EK9Parser.StreamPartContext> stages, IteratorSetup iteratorSetup, String pipelineItemVar, ISymbol elementType, TerminalSetup terminalSetup, DebugInfo debugInfo, List<StageState> stageStates, boolean rawIterator, String pipelineScopeId, List<IRInstr> preLoopInstructions)
      Generate condition (hasNext) and body (next + stages + terminal pipe). Follows ForInGenerator.generateIterationConditionAndBody() pattern. Returns both condition case details and any post-loop instructions (e.g., JOIN emission).
      Parameters:
      rawIterator - true when iterating a raw org.ek9.lang.Iterator (buffered for-range), where next() returns Any and requires a cast to elementType

    • terminalBodyTail

      StageWrapResult terminalBodyTail(List<EK9Parser.StreamPartContext> stages, String itemVar, ISymbol itemType, TerminalSetup terminalSetup, String scopeId, List<IRInstr> preLoopInstructions, List<StageState> stageStates, DebugInfo debugInfo)
      The shared stream body-tail, used identically by the cat path (the StreamPipelineMachinery.buildStreamLoop(IteratorSetup, List, String, ISymbol, boolean, ParseTree, StreamLoopSink, DebugInfo) sink) and the for-range streaming body: plan the per-boundary item vars (declared pre-loop in preLoopInstructions so a post-loop consumer like JOIN's emit references a dominating var), pipe the final-typed item to the terminal, and wrap the result in the intermediate stages. Single source of truth for the planPipelineVars -> generateTerminalPipe -> wrapWithStages triad so the two paths cannot drift (R4).

    • terminalBodyTail

      StageWrapResult terminalBodyTail(List<EK9Parser.StreamPartContext> stages, String itemVar, ISymbol itemType, TerminalSetup terminalSetup, String scopeId, List<IRInstr> preLoopInstructions, List<StageState> stageStates, String asyncExecutorVar, DebugInfo debugInfo)
      terminalBodyTail(List, String, ISymbol, TerminalSetup, String, List, List, DebugInfo) over an async executor's results: asyncExecutorVar is handed to a HEAD among the stages so a satisfied head cancels the executor's outstanding work (null: no executor).