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, DebugInfo debugInfo, List<IRInstr> instructions)
      Async-specific entry to bufferSortEmit(String, String, String, String, CallDetails, ISymbol, List, EK9Parser.StreamPartContext, List, TerminalSetup, DebugInfo, List): drains the async executor (hasNext/next) and closes it after emission. A thin delegate, so the existing async-then-stage-then-sort IR stays byte-identical.

    • 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 submission order through post-ASYNC stages to the terminal.

      Two-phase architecture:

      1. SUBMIT: iterate source, apply pre-ASYNC stages, submit each item to executor
      2. DRAIN: iterate results in order, 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, 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, DebugInfo debugInfo, List<IRInstr> instructions, boolean sourceUnbounded)

    • 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, DebugInfo debugInfo, List<IRInstr> instructions)
      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 (the more flag) can stop it.

      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(String sourceVar, String sourceTypeName, String hasNextMethod, String nextMethod, CallDetails closeCall, ISymbol bucketType, List<EK9Parser.StreamPartContext> preSortStages, EK9Parser.StreamPartContext sortStage, List<EK9Parser.StreamPartContext> postSortStages, TerminalSetup terminalSetup, 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, 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, String moreVar, DebugInfo debugInfo)
      The windowed drain's REFILL step: if (more AND source.hasNext()) { bind submitItem = next(); [pre-async stages]; submit(submitItem); inFlight++ }. Keeps the in-flight window topped up as results drain. When moreVar == null (no post-async head, 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, DebugInfo debugInfo, List<IRInstr> instructions, boolean sourceUnbounded)
      Recursive pipeline TAIL driver: emit a stream given as sourceIter through stages to the terminal, with NO composition limit. Drains any leading sorts, then dispatches the next 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. This is the recursive peer of applyGroup/Split/AsyncToCollection: it mirrors the top-level dispatch in StreamStatementGenerator.generateSingleSourcePipelineCore (lines ~1012-1121) but is parameterised on (stages, iter, elementType) with the terminal already resolved, and it owns its own scope + temp ids. It runs ONLY when a downstream buffer is actually present (a previously-unreachable/dropped case with no golden), and its ids only ever advance the monotonic counter after the top-level scope is open, so the top-level path stays byte-identical. stmtCtx/catCtx are null here (no body coverage probe over a materialised bucket stream — the same convention the buffered drains use).

    • generateConditionAndBody

      ConditionAndPostLoop generateConditionAndBody(EK9Parser.StreamStatementContext ctx, EK9Parser.StreamCatContext catCtx, List<EK9Parser.StreamPartContext> stages, IteratorSetup iteratorSetup, String pipelineItemVar, ISymbol elementType, TerminalSetup terminalSetup, DebugInfo debugInfo, List<StageState> stageStates, boolean rawIterator, String pipelineScopeId, List<IRInstr> preLoopInstructions, boolean sourceUnbounded)
      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, StageState, String, ISymbol, boolean, EK9Parser.StreamCatContext, 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).