fix: Resolve event delivery configuration header passing between frontend and backend (#7514)

Co-authored-by: Gabriel Luiz Freitas Almeida <gabriel@langflow.org>
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
This commit is contained in:
Cristhian Zanforlin Lousa 2025-04-09 14:58:11 -03:00 • committed by GitHub
commit 59a7440045
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
12 changed files with 54 additions and 46 deletions

View file

@ -87,7 +87,7 @@ export default function NodeStatus({
const isBuilding = useFlowStore((state) => state.isBuilding);
const setNode = useFlowStore((state) => state.setNode);
const version = useDarkStore((state) => state.version);
const config = useGetConfig();
const eventDeliveryConfig = useUtilityStore((state) => state.eventDelivery);
const setErrorData = useAlertStore((state) => state.setErrorData);
const postTemplateValue = usePostTemplateValue({
@ -96,10 +96,6 @@ export default function NodeStatus({
node: data.node,
});
const shouldStreamEvents = () => {
return config.data?.event_delivery === EventDeliveryType.STREAMING;
};
// Start polling when connection is initiated
const startPolling = () => {
window.open(connectionLink, "_blank");
@ -169,8 +165,7 @@ export default function NodeStatus({
setValidationStatus(null);
buildFlow({
stopNodeId: nodeId,
stream: shouldStreamEvents(),
eventDelivery: config.data?.event_delivery,
eventDelivery: eventDeliveryConfig,
});
}
@ -265,8 +260,7 @@ export default function NodeStatus({
if (buildStatus === BuildStatus.BUILDING || isBuilding) return;
buildFlow({
stopNodeId: nodeId,
stream: shouldStreamEvents(),
eventDelivery: config.data?.event_delivery,
eventDelivery: eventDeliveryConfig,
});
track("Flow Build - Clicked", { stopNodeId: nodeId });
};

View file

@ -6,7 +6,7 @@ import axios, { AxiosError, AxiosInstance, AxiosRequestConfig } from "axios";
import * as fetchIntercept from "fetch-intercept";
import { useEffect } from "react";
import { Cookies } from "react-cookie";
import { BuildStatus } from "../../constants/enums";
import { BuildStatus, EventDeliveryType } from "../../constants/enums";
import useAlertStore from "../../stores/alertStore";
import useFlowStore from "../../stores/flowStore";
import { checkDuplicateRequestAndStoreRequest } from "./helpers/check-duplicate-requests";
@ -263,6 +263,7 @@ export type StreamingRequestParams = {
onError?: (statusCode: number) => void;
onNetworkError?: (error: Error) => void;
buildController: AbortController;
eventDeliveryConfig?: EventDeliveryType;
};
async function performStreamingRequest({

View file

@ -39,7 +39,7 @@ export const useGetConfig: useQueryFunctionType<undefined, ConfigResponse> = (
const setWebhookPollingInterval = useUtilityStore(
(state) => state.setWebhookPollingInterval,
);
const setEventDelivery = useUtilityStore((state) => state.setEventDelivery);
const { query } = UseRequestProcessor();
const getConfigFn = async () => {
@ -59,6 +59,7 @@ export const useGetConfig: useQueryFunctionType<undefined, ConfigResponse> = (
setWebhookPollingInterval(
data.webhook_polling_interval ?? DEFAULT_POLLING_INTERVAL,
);
setEventDelivery(data.event_delivery ?? EventDeliveryType.POLLING);
}
return data;
};

View file

@ -155,11 +155,7 @@ export default function IOModal({
const chatValue = useUtilityStore((state) => state.chatValueStore);
const setChatValue = useUtilityStore((state) => state.setChatValueStore);
const config = useGetConfig();
function shouldStreamEvents() {
return config.data?.event_delivery === EventDeliveryType.STREAMING;
}
const eventDeliveryConfig = useUtilityStore((state) => state.eventDelivery);
const sendMessage = useCallback(
async ({
@ -178,8 +174,7 @@ export default function IOModal({
files: files,
silent: true,
session: sessionId,
stream: shouldStreamEvents(),
eventDelivery: config.data?.event_delivery,
eventDelivery: eventDeliveryConfig,
}).catch((err) => {
console.error(err);
});

View file

@ -835,7 +835,6 @@ const useFlowStore = create<FlowStoreType>((set, get) => ({
edges: get().edges || undefined,
logBuilds: get().onFlowPage,
playgroundPage,
stream,
eventDelivery,
});
get().setIsBuilding(false);

View file

@ -1,3 +1,4 @@
import { EventDeliveryType } from "@/constants/enums";
import { Pagination, Tag } from "@/types/utils/types";
import { UtilityStoreType } from "@/types/zustand/utility";
import { create } from "zustand";
@ -43,4 +44,7 @@ export const useUtilityStore = create<UtilityStoreType>((set, get) => ({
currentSessionId: "",
setCurrentSessionId: (sessionId: string) =>
set({ currentSessionId: sessionId }),
eventDelivery: EventDeliveryType.POLLING,
setEventDelivery: (eventDelivery: EventDeliveryType) =>
set({ eventDelivery }),
}));

View file

@ -149,7 +149,6 @@ export type FlowStoreType = {
files,
silent,
session,
stream,
eventDelivery,
}: {
startNodeId?: string;
@ -158,7 +157,6 @@ export type FlowStoreType = {
files?: string[];
silent?: boolean;
session?: string;
stream?: boolean;
eventDelivery?: EventDeliveryType;
}) => Promise<void>;
getFlow: () => { nodes: Node[]; edges: EdgeType[]; viewport: Viewport };

View file

@ -1,3 +1,4 @@
import { EventDeliveryType } from "@/constants/enums";
import { Pagination, Tag } from "@/types/utils/types";
export type UtilityStoreType = {
@ -25,4 +26,6 @@ export type UtilityStoreType = {
setCurrentSessionId: (sessionId: string) => void;
setClientId: (clientId: string) => void;
clientId: string;
eventDelivery: EventDeliveryType;
setEventDelivery: (eventDelivery: EventDeliveryType) => void;
};

View file

@ -40,8 +40,7 @@ type BuildVerticesParams = {
logBuilds?: boolean;
session?: string;
playgroundPage?: boolean;
stream?: boolean;
eventDelivery?: EventDeliveryType;
eventDelivery: EventDeliveryType;
};
function getInactiveVertexData(vertexId: string): VertexBuildTypeAPI {
@ -148,7 +147,10 @@ export async function buildFlowVerticesWithFallback(
e.message === POLLING_MESSAGES.STREAMING_NOT_SUPPORTED
) {
// Fallback to polling
return await buildFlowVertices({ ...params, stream: false });
return await buildFlowVertices({
...params,
eventDelivery: EventDeliveryType.POLLING,
});
}
throw e;
}
@ -176,13 +178,16 @@ async function pollBuildEvents(
): Promise<void> {
let isDone = false;
while (!isDone) {
const response = await fetch(`${url}?stream=false`, {
method: "GET",
headers: {
"Content-Type": "application/json",
const response = await fetch(
`${url}?event_delivery=${EventDeliveryType.POLLING}`,
{
method: "GET",
headers: {
"Content-Type": "application/json",
},
signal: abortController.signal, // Add abort signal to fetch
},
signal: abortController.signal, // Add abort signal to fetch
});
);
if (!response.ok) {
const errorData = await response.json().catch(() => ({}));
@ -241,7 +246,6 @@ export async function buildFlowVertices({
logBuilds,
session,
playgroundPage,
stream = true,
eventDelivery,
}: BuildVerticesParams) {
const inputs = {};
@ -260,10 +264,10 @@ export async function buildFlowVertices({
queryParams.append("log_builds", logBuilds.toString());
}
// Add stream parameter when using direct event delivery
if (eventDelivery === EventDeliveryType.DIRECT) {
queryParams.append("stream", "true");
}
queryParams.append(
"event_delivery",
eventDelivery ?? EventDeliveryType.POLLING,
);
if (queryParams.toString()) {
buildUrl = `${buildUrl}?${queryParams.toString()}`;
@ -376,13 +380,12 @@ export async function buildFlowVertices({
}
});
useFlowStore.getState().setBuildController(buildController);
// Then stream the events
const eventsUrl = `${BASE_URL_API}build/${job_id}/events`;
const buildResults: Array<boolean> = [];
const verticesStartTimeMs: Map<string, number> = new Map();
if (stream) {
if (eventDelivery === EventDeliveryType.STREAMING) {
return performStreamingRequest({
method: "GET",
url: eventsUrl,