diff --git a/src/frontend/src/customization/utils/custom-poll-build-events.ts b/src/frontend/src/customization/utils/custom-poll-build-events.ts new file mode 100644 index 0000000000..4503baa9e6 --- /dev/null +++ b/src/frontend/src/customization/utils/custom-poll-build-events.ts @@ -0,0 +1,92 @@ +import { BUILD_POLLING_INTERVAL } from "@/constants/constants"; +import { BuildStatus, EventDeliveryType } from "@/constants/enums"; +import { VertexLayerElementType } from "@/types/zustand/flow"; + +export async function customPollBuildEvents( + url: string, + buildResults: Array, + verticesStartTimeMs: Map, + callbacks: { + onBuildStart?: (idList: VertexLayerElementType[]) => void; + onBuildUpdate?: (data: any, status: BuildStatus, buildId: string) => void; + onBuildComplete?: (allNodesValid: boolean) => void; + onBuildError?: ( + title: string, + list: string[], + idList?: VertexLayerElementType[], + ) => void; + onGetOrderSuccess?: () => void; + onValidateNodes?: (nodes: string[]) => void; + }, + abortController: AbortController, + onEvent, +): Promise { + let isDone = false; + while (!isDone) { + const response = await fetch( + `${url}?event_delivery=${EventDeliveryType.POLLING}`, + { + method: "GET", + headers: { + "Content-Type": "application/json", + Accept: "application/x-ndjson", + }, + signal: abortController.signal, // Add abort signal to fetch + }, + ); + + if (!response.ok) { + const errorData = await response.json().catch(() => ({})); + throw new Error( + errorData.detail || + "Langflow was not able to connect to the server. Please make sure your connection is working properly.", + ); + } + + // Get the response text - will be NDJSON format (one JSON per line) + const responseText = await response.text(); + + // Skip if empty response + if (!responseText.trim()) { + await new Promise((resolve) => setTimeout(resolve, 100)); + continue; + } + + // Split by newlines to get individual JSON objects + const eventLines = responseText.split("\n").filter((line) => line.trim()); + + // If no events, continue polling + if (eventLines.length === 0) { + await new Promise((resolve) => setTimeout(resolve, 100)); + continue; + } + + // Process all events in the NDJSON response + for (const eventStr of eventLines) { + // Process the event + const event = JSON.parse(eventStr); + const result = await onEvent( + event.event, + event.data, + buildResults, + verticesStartTimeMs, + callbacks, + ); + + if (!result) { + isDone = true; + abortController.abort(); + break; + } + + // Check if this was the end event + if (event.event === "end") { + isDone = true; + break; + } + } + + // Add a small delay between polls + await new Promise((resolve) => setTimeout(resolve, BUILD_POLLING_INTERVAL)); + } +} diff --git a/src/frontend/src/utils/buildUtils.ts b/src/frontend/src/utils/buildUtils.ts index 3c5a63a5c9..ccc22b62ac 100644 --- a/src/frontend/src/utils/buildUtils.ts +++ b/src/frontend/src/utils/buildUtils.ts @@ -12,6 +12,7 @@ import { customCancelBuildUrl, customEventsUrl, } from "@/customization/utils/custom-buildUtils"; +import { customPollBuildEvents } from "@/customization/utils/custom-poll-build-events"; import { useMessagesStore } from "@/stores/messagesStore"; import { BuildStatus, EventDeliveryType } from "../constants/enums"; import { getVerticesOrder, postBuildVertex } from "../controllers/API"; @@ -180,74 +181,14 @@ async function pollBuildEvents( }, abortController: AbortController, ): Promise { - let isDone = false; - while (!isDone) { - const response = await fetch( - `${url}?event_delivery=${EventDeliveryType.POLLING}`, - { - method: "GET", - headers: { - "Content-Type": "application/json", - Accept: "application/x-ndjson", - }, - signal: abortController.signal, // Add abort signal to fetch - }, - ); - - if (!response.ok) { - const errorData = await response.json().catch(() => ({})); - throw new Error( - errorData.detail || - "Langflow was not able to connect to the server. Please make sure your connection is working properly.", - ); - } - - // Get the response text - will be NDJSON format (one JSON per line) - const responseText = await response.text(); - - // Skip if empty response - if (!responseText.trim()) { - await new Promise((resolve) => setTimeout(resolve, 100)); - continue; - } - - // Split by newlines to get individual JSON objects - const eventLines = responseText.split("\n").filter((line) => line.trim()); - - // If no events, continue polling - if (eventLines.length === 0) { - await new Promise((resolve) => setTimeout(resolve, 100)); - continue; - } - - // Process all events in the NDJSON response - for (const eventStr of eventLines) { - // Process the event - const event = JSON.parse(eventStr); - const result = await onEvent( - event.event, - event.data, - buildResults, - verticesStartTimeMs, - callbacks, - ); - - if (!result) { - isDone = true; - abortController.abort(); - break; - } - - // Check if this was the end event - if (event.event === "end") { - isDone = true; - break; - } - } - - // Add a small delay between polls - await new Promise((resolve) => setTimeout(resolve, BUILD_POLLING_INTERVAL)); - } + return customPollBuildEvents( + url, + buildResults, + verticesStartTimeMs, + callbacks, + abortController, + onEvent, + ); } export async function buildFlowVertices({