17267738980

Committed 27 Aug 2025 01:12PM UTC coverage: 95.388% (-0.1%) from 95.519%

Build # 17267738980

Build Type

Pull #462

github

Committed by

web-flow

Commit Message

Merge 22c6668dc into 56ca7b899

Pull Request Pull Request #462: feat: 🎸 add [detach]

Run Details

2220 of 2404 branches covered (92.35%)

Branch coverage included in aggregate %.

34 of 38 new or added lines in 6 files covered. (89.47%)

4 existing lines in 2 files now uncovered.

4584 of 4729 relevant lines covered (96.93%)

80855.68 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

97.01

/packages/core/src/server/knownPipeline.js

import { join, basename, dirname } from 'path';
import debug from 'debug';
import sizeof from 'object-sizeof';
import { PassThrough } from 'readable-stream';
import { pipeline } from 'stream';
import once from 'once';
import merge from 'lodash/merge.js';
import cloneDeep from 'lodash/cloneDeep.js';
import { metricsHandle } from './metrics.js';
import errorHandler from './errorHandler.js';
import { isFile } from '../file.js';
import breaker from '../statements/breaker.js';
import settings from '../settings.js';

const dispositionFrom = ({ extension }) => (extension ? `attachment; filename="dump.${extension}"` : 'inline');

const encodingFrom = (headers) => (headers
    && headers['accept-encoding']
    && headers['accept-encoding'].match(/\bgzip\b/) ? 'gzip' : 'identity'
);

const typeFrom = ({ mimeType }) => (mimeType || 'application/json');

const onlyOne = (item) => (Array.isArray(item) ? item.shift() : item);

const knownPipeline = (ezs) => (request, response, next) => {
    if (request.catched
      || !request.methodMatch(['POST', 'OPTIONS', 'HEAD'])
      || request.serverPath === false
      || !request.isPipeline()
    ) {
        return next();
    }
    request.catched = true;
    const { headers, fusible, method, pathName } = request;
    const { query } = request.urlParsed;

    debug('ezs:info')(`Create middleware 'knownPipeline' for ${method} ${pathName}`);
    const triggerError = errorHandler(ezs, request, response);
    const files = ezs.memoize(`knownPipeline>${pathName}`,
        () => pathName
            .slice(1)
            .split(',')
            .map((file) => join(request.serverPath, dirname(file), basename(file, '.ini').concat('.ini')))
            .filter((file) => isFile(file)));
    if (files.length === 0) {
        triggerError(new Error(`Cannot find ${pathName}`), 404);
        return false;
    }
    debug('ezs:debug')(
        `PID ${process.pid} will execute ${pathName} commands with ${sizeof(query)}B of global parameters`,
    );

    const meta = ezs.memoize(`executePipeline>${files}`,
        () => files.map((file) => ezs.metaFile(file)).reduce((prev, cur) => merge(cur, prev), {}));
    const contentEncoding = encodingFrom(headers);
    const contentDisposition = dispositionFrom(meta);
    const contentType = typeFrom(meta);
    const { prepend, append } = meta;
    response.setHeader('Access-Control-Allow-Origin', '*');
    response.setHeader('Access-Control-Allow-Methods', 'POST, OPTIONS');
    response.setHeader('Access-Control-Allow-Headers', '*');
    response.setHeader('Access-Control-Expose-Headers', '*');
    response.setHeader('Content-Encoding', contentEncoding);
    response.setHeader('Content-Disposition', contentDisposition);
    response.setHeader('Content-Type', contentType);
    response.setHeader('X-Request-ID', fusible);

    response.socket.setNoDelay(false);

    if (method !== 'POST') {
        response.writeHead(200);
        response.end();
        return true;
    }
    const {
        mainStatement,
        tracerEnable,
        metricsEnable,
    } = settings;
    const environment = { ...query, headers, request: { fusible, method, pathName } };
    const statements = files.map((file) => {
        debug('ezs:debug')(`${file} will be process by [${mainStatement}]`);
        // parseCommand returns the same object for the same string
        // but here, we want to modify the parameters, so the object must be cloned
        const mainCommand = cloneDeep(ezs.parseCommand(mainStatement));
        mainCommand.args.file = file; // Mandatory parameter
        return ezs.createCommand(mainCommand, environment);
    });
    const prepend2Pipeline = ezs.parseCommand(onlyOne(prepend));
    if (prepend2Pipeline) {
        statements.unshift(ezs.createCommand(prepend2Pipeline, environment));
    }
    const append2Pipeline = ezs.parseCommand(onlyOne(append));
    if (append2Pipeline) {
        statements.push(ezs.createCommand(append2Pipeline, environment));
    }
    if (tracerEnable) {
        statements.unshift(ezs('tracer', { print: '-', last: '>' }));
        statements.push(ezs('tracer', { print: '.', last: '!' }));
    }
    if (metricsEnable) {
        ezs.use({metrics: metricsHandle(pathName)});
        statements.unshift(ezs('metrics', { bucket: 'input' }));
        statements.push(ezs('metrics', { bucket: 'output' }));
    }
    statements.unshift(ezs(breaker, { fusible }));
    statements.push(ezs(breaker, { fusible }));
    const rawStream = new PassThrough();
    let emptyStream = true;
    const responseToBeContinued = setInterval(() => response.writeContinue(), settings.response.checkInterval);
    const responseStarted = once(() => clearInterval(responseToBeContinued));

    statements.push(ezs((data, feed) => {
        if (!response.headersSent) {
            response.writeHead(200);
        }
        responseStarted();
        emptyStream = false;
        return feed.send(data);
    }));

    const decodedStream = rawStream
        .pipe(ezs('truncate', { length: request.headers['content-length'] }))
        .pipe(ezs.uncompress(request.headers));

    const outputStream = new PassThrough();
    outputStream.pipe(response);
    const transformedStream = ezs.createPipeline(decodedStream, statements)
        .on('unpipe', () => {
            request.unpipe(rawStream);
            rawStream.end();
        })
        .pipe(ezs.catch((e) => e))
        .on('error', (e) => {
            outputStream.unpipe(response);
            responseStarted();
            triggerError(e, 400);
            rawStream.destroy();
            decodedStream.destroy();
            transformedStream.destroy();
        });

    pipeline(
        transformedStream,
        ezs.toBuffer(),
        ezs.compress(response.getHeaders()),
        outputStream,
        (e) => {
            if (e) {
                outputStream.unpipe(response);
                responseStarted();
                triggerError(e, 500);
            }
        }
    );

    request
        .once('aborted', () => {
            request.unpipe(rawStream);
            rawStream.end();
        })
        .on('error', (e) => {
            request.unpipe(rawStream);
            rawStream.end();
            triggerError(e, 500);
        })
        .once('close', () => {
            if (emptyStream) {
                //transformedStream.destroy(new Error('No Content'));
            }
        })
        .once('end', () => {
            rawStream.end();
        });
    request.pipe(rawStream);
    request.resume();
};

export default knownPipeline;

1	import { join, basename, dirname } from 'path';
2	import debug from 'debug';
3	import sizeof from 'object-sizeof';
4	import { PassThrough } from 'readable-stream';
5	import { pipeline } from 'stream';
6	import once from 'once';
7	import merge from 'lodash/merge.js';
8	import cloneDeep from 'lodash/cloneDeep.js';
9	import { metricsHandle } from './metrics.js';
10	import errorHandler from './errorHandler.js';
11	import { isFile } from '../file.js';
12	import breaker from '../statements/breaker.js';
13	import settings from '../settings.js';
14
15	const dispositionFrom = ({ extension }) => (extension ? `attachment; filename="dump.${extension}"` : 'inline');	1,074✔
16
17	const encodingFrom = (headers) => (headers	1,074✔
18	&& headers['accept-encoding']
19	&& headers['accept-encoding'].match(/\bgzip\b/) ? 'gzip' : 'identity'
20	);
21
22	const typeFrom = ({ mimeType }) => (mimeType \|\| 'application/json');	1,074✔
23
24	const onlyOne = (item) => (Array.isArray(item) ? item.shift() : item);	1,074!
25
26	const knownPipeline = (ezs) => (request, response, next) => {	1,074✔
27	if (request.catched	376✔
28	\|\| !request.methodMatch(['POST', 'OPTIONS', 'HEAD'])
29	\|\| request.serverPath === false
30	\|\| !request.isPipeline()
31	) {
32	return next();	66✔
33	}
34	request.catched = true;	310✔
35	const { headers, fusible, method, pathName } = request;	310✔
36	const { query } = request.urlParsed;	310✔
37
38	debug('ezs:info')(`Create middleware 'knownPipeline' for ${method} ${pathName}`);	310✔
39	const triggerError = errorHandler(ezs, request, response);	310✔
40	const files = ezs.memoize(`knownPipeline>${pathName}`,	310✔
41	() => pathName	222✔
42	.slice(1)
43	.split(',')
44	.map((file) => join(request.serverPath, dirname(file), basename(file, '.ini').concat('.ini')))	258✔
45	.filter((file) => isFile(file)));	258✔
46	if (files.length === 0) {	310✔
47	triggerError(new Error(`Cannot find ${pathName}`), 404);	54✔
48	return false;	54✔
49	}
50	debug('ezs:debug')(	256✔
51	`PID ${process.pid} will execute ${pathName} commands with ${sizeof(query)}B of global parameters`,
52	);
53
54	const meta = ezs.memoize(`executePipeline>${files}`,	256✔
55	() => files.map((file) => ezs.metaFile(file)).reduce((prev, cur) => merge(cur, prev), {}));	192✔
56	const contentEncoding = encodingFrom(headers);	256✔
57	const contentDisposition = dispositionFrom(meta);	256✔
58	const contentType = typeFrom(meta);	256✔
59	const { prepend, append } = meta;	256✔
60	response.setHeader('Access-Control-Allow-Origin', '*');	256✔
61	response.setHeader('Access-Control-Allow-Methods', 'POST, OPTIONS');	256✔
62	response.setHeader('Access-Control-Allow-Headers', '*');	256✔
63	response.setHeader('Access-Control-Expose-Headers', '*');	256✔
64	response.setHeader('Content-Encoding', contentEncoding);	256✔
65	response.setHeader('Content-Disposition', contentDisposition);	256✔
66	response.setHeader('Content-Type', contentType);	256✔
67	response.setHeader('X-Request-ID', fusible);	256✔
68
69	response.socket.setNoDelay(false);	256✔
70
71	if (method !== 'POST') {	256✔
72	response.writeHead(200);	6✔
73	response.end();	6✔
74	return true;	6✔
75	}
76	const {
77	mainStatement,
78	tracerEnable,
79	metricsEnable,
80	} = settings;	250✔
81	const environment = { ...query, headers, request: { fusible, method, pathName } };	250✔
82	const statements = files.map((file) => {	250✔
83	debug('ezs:debug')(`${file} will be process by [${mainStatement}]`);	286✔
84	// parseCommand returns the same object for the same string
85	// but here, we want to modify the parameters, so the object must be cloned
86	const mainCommand = cloneDeep(ezs.parseCommand(mainStatement));	286✔
87	mainCommand.args.file = file; // Mandatory parameter	286✔
88	return ezs.createCommand(mainCommand, environment);	286✔
89	});
90	const prepend2Pipeline = ezs.parseCommand(onlyOne(prepend));	250✔
91	if (prepend2Pipeline) {	250✔
92	statements.unshift(ezs.createCommand(prepend2Pipeline, environment));	6✔
93	}
94	const append2Pipeline = ezs.parseCommand(onlyOne(append));	250✔
95	if (append2Pipeline) {	250✔
96	statements.push(ezs.createCommand(append2Pipeline, environment));	6✔
97	}
98	if (tracerEnable) {	250✔
99	statements.unshift(ezs('tracer', { print: '-', last: '>' }));	6✔
100	statements.push(ezs('tracer', { print: '.', last: '!' }));	6✔
101	}
102	if (metricsEnable) {	250✔
103	ezs.use({metrics: metricsHandle(pathName)});	12✔
104	statements.unshift(ezs('metrics', { bucket: 'input' }));	12✔
105	statements.push(ezs('metrics', { bucket: 'output' }));	12✔
106	}
107	statements.unshift(ezs(breaker, { fusible }));	250✔
108	statements.push(ezs(breaker, { fusible }));	250✔
109	const rawStream = new PassThrough();	250✔
110	let emptyStream = true;	250✔
111	const responseToBeContinued = setInterval(() => response.writeContinue(), settings.response.checkInterval);	250✔
112	const responseStarted = once(() => clearInterval(responseToBeContinued));	250✔
113
114	statements.push(ezs((data, feed) => {	250✔
115	if (!response.headersSent) {	183,880✔
116	response.writeHead(200);	211✔
117	}
118	responseStarted();	183,880✔
119	emptyStream = false;	183,880✔
120	return feed.send(data);	183,880✔
121	}));
122
123	const decodedStream = rawStream	250✔
124	.pipe(ezs('truncate', { length: request.headers['content-length'] }))
125	.pipe(ezs.uncompress(request.headers));
126
127	const outputStream = new PassThrough();	250✔
128	outputStream.pipe(response);	250✔
129	const transformedStream = ezs.createPipeline(decodedStream, statements)	250✔
130	.on('unpipe', () => {
131	request.unpipe(rawStream);	247✔
132	rawStream.end();	247✔
133	})
134	.pipe(ezs.catch((e) => e))	12✔
135	.on('error', (e) => {
136	outputStream.unpipe(response);	40✔
137	responseStarted();	40✔
138	triggerError(e, 400);	40✔
139	rawStream.destroy();	40✔
140	decodedStream.destroy();	40✔
141	transformedStream.destroy();	40✔
142	});
143
144	pipeline(	250✔
145	transformedStream,
146	ezs.toBuffer(),
147	ezs.compress(response.getHeaders()),
148	outputStream,
149	(e) => {
150	if (e) {	247✔
151	outputStream.unpipe(response);	30✔
152	responseStarted();	30✔
153	triggerError(e, 500);	30✔
154	}
155	}
156	);
157
158	request	250✔
159	.once('aborted', () => {
160	request.unpipe(rawStream);	5✔
161	rawStream.end();	5✔
162	})
163	.on('error', (e) => {
UNCOV 164	request.unpipe(rawStream);	×
UNCOV 165	rawStream.end();	×
UNCOV 166	triggerError(e, 500);	×
167	})
168	.once('close', () => {
169	if (emptyStream) {	223✔
170	//transformedStream.destroy(new Error('No Content'));
171	}
172	})
173	.once('end', () => {
174	rawStream.end();	222✔
175	});
176	request.pipe(rawStream);	250✔
177	request.resume();	250✔
178	};
179
180	export default knownPipeline;

Inist-CNRS / ezs / 17267738980

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous