Press n or j to go to the next uncovered block, b, p or k for the previous block.
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 | 24x 9x 2x 2x 2x 2x 1x 1x 1x 8x 4x 4x 3x 2x 1x 1x 1x 3x 3x 6x 5x 5x 5x 3x 3x 3x 3x 4x 3x 3x | // execution-phases/stream-phase.ts
// Verantwortlichkeit: Stream-/Output-Verarbeitung während der Ausführung
// - Runtime-Callbacks wrappen
// - Geparste Stream-Zeilen an StreamHandler delegieren
// - stderr-Fallback-Puffer verarbeiten
import type { Logger } from "@shared/logger";
import type { PinStateChange } from "@shared/types/arduino.types";
import type { ParsedStderrOutput, ArduinoOutputParser } from "../../arduino-output-parser";
import type { RegistryManager } from "../../registry-manager";
import type { StreamHandler } from "../stream-handler";
import type { ExecutionState } from "../execution-manager";
interface StreamCallbacks {
onOutput: (line: string, isComplete?: boolean) => void;
onError: (line: string) => void;
onPinState?: (pin: number, type: PinStateChange, value: number) => void;
}
interface StreamDependencies {
registryManager: RegistryManager;
streamHandler: StreamHandler;
}
interface StreamCallbackDependencies {
registryManager: Pick<RegistryManager, "isWaiting">;
logger: Logger;
}
/**
* Erstellt Runtime-Callbacks mit Queueing, Telemetrie-Erkennung und Batcher-Routing.
*/
export function createStreamCallbacks(
onOutput: (line: string, isComplete?: boolean) => void,
onError: (line: string) => void,
onPinState: ((pin: number, type: PinStateChange, value: number) => void) | undefined,
state: ExecutionState | undefined,
deps: StreamCallbackDependencies,
): StreamCallbacks {
return {
onOutput: (line: string, isComplete?: boolean) => {
if (typeof line === "string" && line.startsWith("[[SIM_TELEMETRY:") && line.endsWith("]]")) {
try {
const jsonStr = line.slice("[[SIM_TELEMETRY:".length, -2);
const metrics = JSON.parse(jsonStr);
if (state?.telemetryCallback) {
state.telemetryCallback(metrics);
}
return;
} catch (err) {
deps.logger.warn(`Failed to parse telemetry marker: ${err}`);
}
}
if (state?.serialOutputBatcher) {
state.serialOutputBatcher.enqueue(line);
} else if (onOutput && state?.processKilled === false) {
onOutput(line, isComplete);
}
},
onPinState: (pin: number, stateType: PinStateChange, value: number) => {
if (state && deps.registryManager.isWaiting()) {
state.messageQueue.push({
type: "pinState",
data: { pin, stateType, value },
});
E} else if (onPinState) {
onPinState(pin, stateType, value);
}
},
onError: (line: string) => {
Eif (onError) {
onError(line);
}
},
};
}
/**
* Delegiert eine geparste stderr/stdout-Zeile an den bestehenden StreamHandler.
*/
export function delegateParsedLineToStreamHandler(
parsed: ParsedStderrOutput,
state: ExecutionState | undefined,
callbacks: StreamCallbacks,
deps: StreamDependencies,
): void {
if (!state) return;
const streamState = {
pinStateBatcher: state.pinStateBatcher,
serialOutputBatcher: state.serialOutputBatcher,
backpressurePaused: state.backpressurePaused,
isPaused: state.state === "paused",
baudrate: state.baudrate,
registryManager: deps.registryManager,
};
deps.streamHandler.handleParsedLine(parsed, streamState, callbacks);
state.backpressurePaused = streamState.backpressurePaused;
}
/**
* Verarbeitet gepufferte stderr-Daten, wenn kein Line-Streaming verfügbar ist.
*/
export function handleStderrFallbackData(
data: Buffer,
state: ExecutionState,
callbacks: StreamCallbacks,
deps: StreamDependencies & { stderrParser: ArduinoOutputParser },
): void {
state.stderrFallbackBuffer += data.toString();
const lines = state.stderrFallbackBuffer.split(/\r?\n/);
state.stderrFallbackBuffer = lines.pop() || "";
for (const line of lines) {
if (!line) continue;
const parsed = deps.stderrParser.parseStderrLine(line, state.processStartTime);
delegateParsedLineToStreamHandler(parsed, state, callbacks, deps);
}
}
|