mudhorn/n8n
1
0
Fork 0
mirror of https://github.com/n8n-io/n8n.git synced 2025-03-05 20:50:17 -08:00
Code Issues Actions Packages Projects Releases Wiki Activity

fix(core): Webhooks responding with binary data should not prematurely end the response stream (#9063)

Browse source
This commit is contained in:
कारतोफ्फेलस्क्रिप्ट™ 2024-05-07 13:48:20 +02:00 committed by GitHub
parent 225fdbb379
commit 23b676d7cb
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 15 additions and 18 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

31
packages/cli/src/WebhookHelpers.ts
View file

@ -9,7 +9,7 @@
import type express from 'express'; import type express from 'express';
import { Container } from 'typedi'; import { Container } from 'typedi';
import get from 'lodash/get'; import get from 'lodash/get';
import { pipeline } from 'stream/promises'; import { finished } from 'stream/promises';
import formidable from 'formidable'; import formidable from 'formidable';
import { BinaryDataService, NodeExecuteFunctions } from 'n8n-core'; import { BinaryDataService, NodeExecuteFunctions } from 'n8n-core';
@ -30,6 +30,7 @@ import type {
IWebhookResponseData, IWebhookResponseData,
IWorkflowDataProxyAdditionalKeys, IWorkflowDataProxyAdditionalKeys,
IWorkflowExecuteAdditionalData, IWorkflowExecuteAdditionalData,
WebhookResponseMode,
Workflow, Workflow,
WorkflowExecuteMode, WorkflowExecuteMode,
} from 'n8n-workflow'; } from 'n8n-workflow';
@ -272,7 +273,7 @@ export async function executeWebhook(
additionalKeys, additionalKeys,
undefined, undefined,
'onReceived', 'onReceived',
); ) as WebhookResponseMode;
const responseCode = workflow.expression.getSimpleParameterValue( const responseCode = workflow.expression.getSimpleParameterValue(
workflowStartNode, workflowStartNode,
webhookData.webhookDescription.responseCode as string, webhookData.webhookDescription.responseCode as string,
@ -291,7 +292,7 @@ export async function executeWebhook(
'firstEntryJson', 'firstEntryJson',
); );
if (!['onReceived', 'lastNode', 'responseNode'].includes(responseMode as string)) { if (!['onReceived', 'lastNode', 'responseNode'].includes(responseMode)) {
// If the mode is not known we error. Is probably best like that instead of using // If the mode is not known we error. Is probably best like that instead of using
// the default that people know as early as possible (probably already testing phase) // the default that people know as early as possible (probably already testing phase)
// that something does not resolve properly. // that something does not resolve properly.
@ -562,7 +563,8 @@ export async function executeWebhook(
if (binaryData?.id) { if (binaryData?.id) {
res.header(response.headers); res.header(response.headers);
const stream = await Container.get(BinaryDataService).getAsStream(binaryData.id); const stream = await Container.get(BinaryDataService).getAsStream(binaryData.id);
await pipeline(stream, res); stream.pipe(res, { end: false });
await finished(stream);
responseCallback(null, { noWebhookResponse: true }); responseCallback(null, { noWebhookResponse: true });
} else if (Buffer.isBuffer(response.body)) { } else if (Buffer.isBuffer(response.body)) {
res.header(response.headers); res.header(response.headers);
@ -595,6 +597,7 @@ export async function executeWebhook(
}); });
} }
process.nextTick(() => res.end());
didSendResponse = true; didSendResponse = true;
}) })
.catch(async (error) => { .catch(async (error) => {
@ -659,17 +662,9 @@ export async function executeWebhook(
return data; return data;
} }
if (responseMode === 'responseNode') { // in `responseNode` mode `responseCallback` is called by `responsePromise`
if (!didSendResponse) { if (responseMode === 'responseNode' && responsePromise) {
// Return an error if no Webhook-Response node did send any data await Promise.allSettled([responsePromise.promise()]);
responseCallback(null, {
data: {
message: 'Workflow executed successfully',
},
responseCode,
});
didSendResponse = true;
}
return undefined; return undefined;
} }
@ -795,14 +790,16 @@ export async function executeWebhook(
res.setHeader('Content-Type', binaryData.mimeType); res.setHeader('Content-Type', binaryData.mimeType);
if (binaryData.id) { if (binaryData.id) {
const stream = await Container.get(BinaryDataService).getAsStream(binaryData.id); const stream = await Container.get(BinaryDataService).getAsStream(binaryData.id);
await pipeline(stream, res); stream.pipe(res, { end: false });
await finished(stream);
} else { } else {
res.end(Buffer.from(binaryData.data, BINARY_ENCODING)); res.write(Buffer.from(binaryData.data, BINARY_ENCODING));
} }
responseCallback(null, { responseCallback(null, {
noWebhookResponse: true, noWebhookResponse: true,
}); });
process.nextTick(() => res.end());
} }
} else if (responseData === 'noData') { } else if (responseData === 'noData') {
// Return without data // Return without data

2
packages/workflow/src/Interfaces.ts
View file

@ -1834,7 +1834,7 @@ export interface IWebhookResponseData {
} }
export type WebhookResponseData = 'allEntries' | 'firstEntryJson' | 'firstEntryBinary' | 'noData'; export type WebhookResponseData = 'allEntries' | 'firstEntryJson' | 'firstEntryBinary' | 'noData';
export type WebhookResponseMode = 'onReceived' | 'lastNode'; export type WebhookResponseMode = 'onReceived' | 'lastNode' | 'responseNode';
export interface INodeTypes { export interface INodeTypes {
getByName(nodeType: string): INodeType | IVersionedNodeType; getByName(nodeType: string): INodeType | IVersionedNodeType;