diff --git a/.qlty/qlty.toml b/.qlty/qlty.toml
new file mode 100644
index 00000000..116b11d3
--- /dev/null
+++ b/.qlty/qlty.toml
@@ -0,0 +1,10 @@
+config_version = "0"
+
+exclude_patterns = [
+ "test/**",
+]
+
+[coverage]
+ignores = [
+ "benchmark/**",
+]
diff --git a/MIGRATION.md b/MIGRATION.md
index da131331..000813ec 100644
--- a/MIGRATION.md
+++ b/MIGRATION.md
@@ -14,7 +14,67 @@ Minimal supported Node.js version is 22.12.0.
* `http2-wrapper` and `cacheable-lookup` were replaced with built-in implementations (`agent.http2`, `dnsCache` options changed)
* credentials are stripped when a hook redirects request to a different origin
-`response.body` passed to `afterResponse` action is still a `Buffer` (with default `responseType: 'buffer'`), but `response.rawBody` is now a `Uint8Array`.
+#### Streaming resources to storage
+Version 7 stops storing resource content in memory ([#386](https://github.com/website-scraper/node-website-scraper/issues/386)). Resources which need no modification (images, fonts, media, scripts - everything except html and css) are streamed directly from the network to storage, and html/css content is freed as soon as the resource is saved. Memory usage no longer grows with the size of the scraped website.
+
+#### saveResource action
+
+Resource content is now exposed as a Readable stream via `resource.getContentStream()` instead of `resource.getText()`. For streamed resources the stream can be consumed only once - consume or destroy it within the action. `getContentStream()` hands such a stream out only once, a second call throws.
+
+```javascript
+// before
+registerAction('saveResource', async ({resource}) => {
+ await saveItSomewhere(resource.getFilename(), resource.getText());
+});
+
+// after
+import { buffer } from 'stream/consumers';
+registerAction('saveResource', async ({resource}) => {
+ await saveItSomewhere(resource.getFilename(), await buffer(resource.getContentStream()));
+});
+// or pipe resource.getContentStream() to your storage to avoid buffering
+```
+
+`saveResource` actions are no longer called one resource at a time: streamed resources are saved as their responses arrive, concurrently. If your action writes to a shared sink which is not concurrency-safe (an archive, a single file, a serial connection) - make it concurrency-safe or set `requestConcurrency: 1`.
+
+Note: when more than one `saveResource` action is registered, resource content is buffered in memory once so that every action can read it - memory usage is then again bounded by resource size for those resources.
+
+#### afterResponse action
+
+The action is called when response headers arrive, before the body is downloaded. It receives `{response}` where response is `{url, statusCode, headers, getBody()}` instead of the full got response. Returning a string is not supported anymore - return an object or null. Return an object without `body` to keep the resource streaming (no buffering).
+
+Other contract changes:
+
+* When a returned string `body` has no `encoding`, it is decoded with the encoding derived from the response `content-type` header (`utf8` if it contains utf-8, case-insensitive, `binary` otherwise). Previously `binary` was always the default - pass `encoding: 'binary'` explicitly to keep the old behavior.
+
+```javascript
+// before
+registerAction('afterResponse', ({response}) => {
+ if (response.statusCode === 404) return null;
+ return { body: response.body, metadata: { headers: response.headers } };
+});
+
+// after
+registerAction('afterResponse', ({response}) => {
+ if (response.statusCode === 404) return null;
+ return { metadata: { headers: response.headers } };
+});
+// call `await response.getBody()` only if you need to inspect or replace the body
+```
+
+#### scrape() result
+
+Resource content is not retained after save: `resource.getText()` returns `null` in the result and in `onResourceSaved`. Read saved files from your storage instead.
+
+#### generateFilename action
+
+`responseData` no longer contains `body` - filenames are generated from headers (`url`, `statusCode`, `mimeType`, `encoding`, `metadata`) before the body is consumed.
+
+#### Other behavior changes
+
+* Binary files are written byte-for-byte from the network (previously the content was round-tripped through a latin1 string). Output for correctly-served files is unchanged; edge-case encodings may differ.
+* Transient request errors are retried as before. For streamed resources (everything except html and css) retries happen only until response headers are received - a connection which dies mid-body is not retried, since the body is already being written to storage.
+* `responseType` in the `request` option is ignored (responses are streamed).
## From version 4 to 5
diff --git a/README.md b/README.md
index 65578d5f..71b8ed17 100644
--- a/README.md
+++ b/README.md
@@ -119,7 +119,7 @@ The difference between [maxRecursiveDepth](#maxRecursiveDepth) and [maxDepth](#m
only html resources with depth 2 will be filtered out, last image will be downloaded
#### request
-Object, custom options for http module [got](https://github.com/sindresorhus/got#options) which is used inside website-scraper. Allows to set retries, cookies, userAgent, encoding, etc.
+Object, custom options for http module [got](https://github.com/sindresorhus/got#options) which is used inside website-scraper. Allows to set retries, cookies, userAgent, encoding, etc. Note that `responseType` is controlled by the scraper (responses are streamed) and is ignored if passed here.
```javascript
// use same request options for all resources
scrape({
@@ -244,7 +244,7 @@ class MyPlugin {
registerAction('afterFinish', async () => {});
registerAction('error', async ({error}) => {console.error(error)});
registerAction('beforeRequest', async ({resource, requestOptions}) => ({requestOptions}));
- registerAction('afterResponse', async ({response}) => response.body);
+ registerAction('afterResponse', async ({response}) => ({metadata: {headers: response.headers}}));
registerAction('onResourceSaved', ({resource}) => {});
registerAction('onResourceError', ({resource, error}) => {});
registerAction('saveResource', async ({resource}) => {});
@@ -315,15 +315,19 @@ registerAction('beforeRequest', async ({resource, requestOptions}) => {
```
##### afterResponse
-Action afterResponse is called after each response, it allows to customize resource or reject its saving.
+Action afterResponse is called after response headers were received, before the body is consumed. It allows to customize resource or reject its saving.
Parameters - object which includes:
-* response - response object from http module [got](https://github.com/sindresorhus/got#response)
+* response - object which includes:
+ * `url` (string) - final url of the response (after redirects)
+ * `statusCode` (number)
+ * `headers` (object)
+ * `getBody()` - async function which buffers the response body and resolves with a `Buffer`. Note that calling it loads the whole body into memory - don't call it if you only need headers or status code, so the resource keeps streaming to storage
Return resolved `Promise` with:
- * object if the resource should be saved, object should contain next properties:
- * `body` (string, required)
- * `encoding` (`binary` or `utf8`) is used to save the file, binary is used by default.
+ * object if the resource should be saved, object may contain next properties:
+ * `body` (string or Buffer, optional) - replaces resource content; if omitted, the original body is used without buffering it in memory. `null` body skips the resource, same as returning `null`
+ * `encoding` (`binary` or `utf8`) is used to decode a string `body`; defaults to the encoding derived from the response `content-type` header (`utf8` if it contains utf-8, case-insensitive, `binary` otherwise)
* `metadata` (object) - everything you want to save for this resource (like headers, original text, timestamps, etc.), scraper will not use this field at all, it is only for the result
* or null if the resource should be skipped
@@ -335,8 +339,6 @@ registerAction('afterResponse', ({response}) => {
return null;
} else {
return {
- body: response.body,
- encoding: 'utf8',
metadata: {
headers: response.headers,
someOtherData: [ 1, 2, 3 ]
@@ -344,6 +346,15 @@ registerAction('afterResponse', ({response}) => {
}
}
});
+
+// Modify html body before it is processed
+registerAction('afterResponse', async ({response}) => {
+ if (response.headers['content-type']?.includes('text/html')) {
+ const body = await response.getBody();
+ return { body: transformHtml(body.toString()), encoding: 'utf8' };
+ }
+ return {}; // everything else keeps streaming to storage
+});
```
##### onResourceSaved
@@ -374,7 +385,7 @@ Action generateFilename is called to determine path in file system where the res
Parameters - object which includes:
* resource - [Resource](https://github.com/website-scraper/node-website-scraper/blob/master/lib/resource.js) object
-* responseData - object returned from afterResponse action, contains `url`, `mimeType`, `body`, `metadata` properties
+* responseData - object which contains `url`, `statusCode`, `mimeType`, `encoding`, `metadata` properties. Note that it does not contain the response body - filenames are generated before the body is consumed
Should return object which includes:
* filename - String, relative to `directory` path for specified resource
@@ -419,12 +430,23 @@ Action saveResource is called to save file to some storage. Use it to save files
Parameters - object which includes:
* resource - [Resource](https://github.com/website-scraper/node-website-scraper/blob/master/lib/resource.js) object
-If multiple actions `saveResource` added - resource will be saved to multiple storages.
+Resource content is exposed as a Readable stream returned by `resource.getContentStream()`. For resources which are streamed directly from the network (all types except html and css) the stream can be consumed only once - consume it (or destroy it) within the action. `getContentStream()` hands such a stream out only once, a second call throws - buffer the content yourself if you need it more than once.
+
+If multiple actions `saveResource` added - resource will be saved to multiple storages (the scraper buffers streamed content in that case so every action can read it).
```javascript
+import { pipeline } from 'stream/promises';
+
registerAction('saveResource', async ({resource}) => {
const filename = resource.getFilename();
- const text = resource.getText();
- await saveItSomewhere(filename, text);
+ await pipeline(resource.getContentStream(), createMyStorageWriteStream(filename));
+});
+
+// or, if your storage needs the whole content at once
+import { buffer } from 'stream/consumers';
+
+registerAction('saveResource', async ({resource}) => {
+ const content = await buffer(resource.getContentStream());
+ await saveItSomewhere(resource.getFilename(), content);
});
```
@@ -434,6 +456,8 @@ Array of [Resource](https://github.com/website-scraper/node-website-scraper/blob
- `filename`: filename where page was saved (relative to `directory`)
- `children`: array of children Resources
+Note that resource content is not retained in memory after the resource was saved - read the saved files from your storage if you need the content.
+
## Log and debug
This module uses [debug](https://github.com/visionmedia/debug) to log events. To enable logs you should use environment variable `DEBUG`.
The next command will log everything from website-scraper
diff --git a/benchmark/memory.mjs b/benchmark/memory.mjs
new file mode 100644
index 00000000..2c7616e3
--- /dev/null
+++ b/benchmark/memory.mjs
@@ -0,0 +1,91 @@
+/*
+ * Memory benchmark for issue #386.
+ *
+ * Starts a local http server serving an html page referencing FILES_COUNT
+ * binary files of FILE_SIZE_MB each (generated chunk-wise on the fly, so the
+ * server itself stays flat), scrapes it and reports peak memory usage.
+ *
+ * Usage:
+ * npm run benchmark:memory
+ * FILES_COUNT=50 FILE_SIZE_MB=10 REQUEST_CONCURRENCY=8 npm run benchmark:memory
+ */
+
+import http from 'http';
+import fs from 'fs';
+import os from 'os';
+import path from 'path';
+import scrape from 'website-scraper';
+
+const FILES_COUNT = parseInt(process.env.FILES_COUNT || '100', 10);
+const FILE_SIZE_MB = parseInt(process.env.FILE_SIZE_MB || '20', 10);
+const REQUEST_CONCURRENCY = parseInt(process.env.REQUEST_CONCURRENCY || '8', 10);
+
+const CHUNK = Buffer.alloc(64 * 1024, 0xab);
+const CHUNKS_PER_FILE = Math.ceil(FILE_SIZE_MB * 1024 * 1024 / CHUNK.length);
+
+function startServer () {
+ const server = http.createServer((req, res) => {
+ if (req.url === '/') {
+ const imgs = Array.from({length: FILES_COUNT}, (_, i) => `
`).join('\n');
+ res.writeHead(200, {'content-type': 'text/html'});
+ res.end(`${imgs}`);
+ return;
+ }
+
+ res.writeHead(200, {'content-type': 'application/octet-stream'});
+ let sent = 0;
+ const writeChunk = () => {
+ while (sent < CHUNKS_PER_FILE) {
+ sent++;
+ if (!res.write(CHUNK)) {
+ res.once('drain', writeChunk);
+ return;
+ }
+ }
+ res.end();
+ };
+ writeChunk();
+ });
+
+ return new Promise((resolve) => {
+ server.listen(0, '127.0.0.1', () => resolve(server));
+ });
+}
+
+function formatMB (bytes) {
+ return `${Math.round(bytes / 1024 / 1024)} MB`;
+}
+
+const server = await startServer();
+const baseUrl = `http://127.0.0.1:${server.address().port}/`;
+const directory = fs.mkdtempSync(path.join(os.tmpdir(), 'website-scraper-benchmark-'));
+fs.rmSync(directory, {recursive: true, force: true}); // scraper requires the directory to not exist
+
+const peak = {rss: 0, heapUsed: 0, external: 0, arrayBuffers: 0};
+const sampler = setInterval(() => {
+ const usage = process.memoryUsage();
+ for (const key of Object.keys(peak)) {
+ peak[key] = Math.max(peak[key], usage[key]);
+ }
+}, 200);
+
+console.log(`scraping ${FILES_COUNT} files x ${FILE_SIZE_MB} MB (${FILES_COUNT * FILE_SIZE_MB} MB total), concurrency ${REQUEST_CONCURRENCY}`);
+const startedAt = Date.now();
+
+try {
+ await scrape({
+ urls: [baseUrl],
+ directory,
+ requestConcurrency: REQUEST_CONCURRENCY
+ });
+} finally {
+ clearInterval(sampler);
+ server.close();
+ fs.rmSync(directory, {recursive: true, force: true});
+}
+
+console.log(`done in ${((Date.now() - startedAt) / 1000).toFixed(1)}s`);
+console.log(`peak rss: ${formatMB(peak.rss)}`);
+console.log(`peak heapUsed: ${formatMB(peak.heapUsed)}`);
+console.log(`peak external: ${formatMB(peak.external)}`);
+console.log(`peak arrayBuffers: ${formatMB(peak.arrayBuffers)}`);
diff --git a/eslint.config.js b/eslint.config.js
index fbd8b306..af7c4f68 100644
--- a/eslint.config.js
+++ b/eslint.config.js
@@ -10,7 +10,7 @@ export default [
'@stylistic': stylistic
},
languageOptions: {
- ecmaVersion: 2020,
+ ecmaVersion: 2025,
sourceType: 'module',
globals: {
...globals.node,
diff --git a/lib/config/defaults.js b/lib/config/defaults.js
index 8c731519..32bd9a3b 100644
--- a/lib/config/defaults.js
+++ b/lib/config/defaults.js
@@ -46,7 +46,6 @@ const config = {
],
request: {
throwHttpErrors: false,
- responseType: 'buffer',
//cookieJar: true,
decompress: true,
headers: {
diff --git a/lib/plugins/save-resource-to-fs-plugin.js b/lib/plugins/save-resource-to-fs-plugin.js
index 19b2d1d4..4350bcd3 100644
--- a/lib/plugins/save-resource-to-fs-plugin.js
+++ b/lib/plugins/save-resource-to-fs-plugin.js
@@ -1,10 +1,10 @@
import path from 'path';
import fs from 'fs';
-import { outputFile } from '../utils/fs.js';
+import { outputFileStream } from '../utils/fs.js';
class SaveResourceToFileSystemPlugin {
apply (registerAction) {
- let absoluteDirectoryPath, loadedResources = [];
+ let absoluteDirectoryPath, wroteToDirectory = false;
registerAction('beforeStart', ({options}) => {
if (!options.directory || typeof options.directory !== 'string') {
@@ -20,13 +20,24 @@ class SaveResourceToFileSystemPlugin {
registerAction('saveResource', async ({resource}) => {
const filename = path.join(absoluteDirectoryPath, resource.getFilename());
- const text = resource.getText();
- await outputFile(filename, text, resource.getEncoding());
- loadedResources.push(resource);
+ wroteToDirectory = true; // outputFileStream creates directories even if the write then fails
+ try {
+ await outputFileStream(filename, resource.getContentStream());
+ } catch (err) {
+ try {
+ await fs.promises.rm(filename, {force: true}); // don't leave partially written files
+ } catch {
+ // cleanup failure must not mask the original save error
+ }
+ throw err;
+ }
});
registerAction('error', async () => {
- if (loadedResources.length > 0) {
+ // remove what this plugin created (a failed save may leave directories behind even
+ // when no resource was saved), but nothing when it wrote nothing - the directory
+ // may then hold files placed there by other plugins
+ if (wroteToDirectory) {
await fs.promises.rm(absoluteDirectoryPath, {force: true, recursive: true});
}
});
diff --git a/lib/request.js b/lib/request.js
index 16defebc..65ff0cb8 100644
--- a/lib/request.js
+++ b/lib/request.js
@@ -1,4 +1,6 @@
import got from 'got';
+import { Readable } from 'stream';
+import { buffer as readStreamToBuffer } from 'stream/consumers';
import logger from './logger.js';
import { extend } from './utils/index.js';
@@ -6,29 +8,14 @@ function getMimeType (contentType) {
return contentType ? contentType.split(';')[0] : null;
}
-function defaultResponseHandler ({response}) {
- return Promise.resolve(response);
-}
-
function extractEncodingFromHeader (headers) {
const contentTypeHeader = headers['content-type'];
- return contentTypeHeader && contentTypeHeader.includes('utf-8') ? 'utf8' : 'binary';
-}
-
-function getEncoding (response) {
- if (response && typeof response === 'object') {
- if (response.headers && typeof response.headers === 'object') {
- return extractEncodingFromHeader(response.headers);
- } else if (response.encoding) {
- return response.encoding;
- }
- }
-
- return 'binary';
+ // charset values are case-insensitive (RFC 9110), e.g. "UTF-8" is common
+ return contentTypeHeader && contentTypeHeader.toLowerCase().includes('utf-8') ? 'utf8' : 'binary';
}
-function throwTypeError (result) {
+function throwAfterResponseTypeError (result) {
let type = typeof result;
if (result instanceof Error) {
@@ -37,80 +24,210 @@ function throwTypeError (result) {
type = 'array';
}
- throw new Error(`Wrong response handler result. Expected string or object, but received ${type}`);
+ throw new Error(`Wrong afterResponse result. Expected null or object with { body?, encoding?, metadata? }, but received ${type}`);
}
-function toBuffer (data) {
- return Buffer.isBuffer(data) ? data : Buffer.from(data.buffer, data.byteOffset, data.byteLength);
+/*
+ * Attaches got retry handling to a request stream and resolves with { response, stream } when
+ * response headers arrive, before the body is consumed. Attaching the 'retry' listener is what
+ * enables got retries for streams - transient errors before/at response time are retried with
+ * a fresh stream. The listener is removed once headers arrive: a consumer which streams the body
+ * (e.g. pipeline to storage) cannot be handed a replacement stream, see readBodyWithRetry for
+ * consumers which buffer the body and can.
+ */
+function waitForResponse (url, initialStream) {
+ return new Promise((resolve, reject) => {
+ const attempt = (stream) => {
+ const onRetry = (retryCount, error, createRetryStream) => {
+ stream.off('response', onResponse);
+ stream.off('error', onError);
+ logger.debug(`[request] retrying request for ${url}, attempt ${retryCount}, reason: ${error.message}`);
+ attempt(createRetryStream());
+ };
+ const onResponse = (response) => {
+ stream.off('retry', onRetry);
+ stream.off('error', onError);
+ // a body error emitted before the consumer attaches its own handlers must not crash the process
+ stream.on('error', (err) => logger.debug(`[request] response stream error for ${response.url}: ${err.message}`));
+ resolve({ response, stream });
+ };
+ const onError = (error) => {
+ stream.off('retry', onRetry);
+ stream.off('response', onResponse);
+ reject(error);
+ };
+ stream.once('retry', onRetry);
+ stream.once('response', onResponse);
+ stream.once('error', onError);
+ };
+ attempt(initialStream);
+ });
}
-function getData (result) {
- let data = result;
- if (result && typeof result === 'object' && 'body' in result) {
- data = result.body;
- }
+function requestWithRetry (url, requestOptions) {
+ return waitForResponse(url, got.stream(url, requestOptions));
+}
- return data;
+/*
+ * Reads the whole body of a response stream with the 'retry' listener attached, so that got
+ * retries a transient mid-body error (e.g. connection reset) within its retry rules instead
+ * of emitting it. Resolves with { body } or, when got retried, with { retryStream }.
+ */
+function readBodyOrRetry (url, stream) {
+ let onRetry;
+ const retried = new Promise((resolve) => {
+ onRetry = (retryCount, error, createRetryStream) => {
+ logger.debug(`[request] retrying request for ${url} after body error, attempt ${retryCount}, reason: ${error.message}`);
+ resolve({ retryStream: createRetryStream() });
+ };
+ stream.once('retry', onRetry);
+ });
+ // got destroys the stream right before emitting 'retry', so on retry the read fails
+ // a tick later - the race has already settled with retryStream by then
+ const read = readStreamToBuffer(stream).then((body) => ({ body }));
+
+ return Promise.race([read, retried]).finally(() => stream.off('retry', onRetry));
}
-function transformResult (result) {
- const encoding = getEncoding(result);
- const data = getData(result);
+/*
+ * Buffers the body of a response stream, re-reading it from the retried response whenever
+ * got retries the request mid-body (limit, errorCodes and backoff are got's).
+ */
+async function readBodyWithRetry (url, stream) {
+ let current = stream;
+ for (;;) {
+ const { body, retryStream } = await readBodyOrRetry(url, current);
+ if (body) {
+ return body;
+ }
+ ({ stream: current } = await waitForResponse(url, retryStream));
+ }
+}
- // Check for no data
- if (data === null || data === undefined) {
+function afterResponseBodyToBuffer (body, encoding) {
+ if (Buffer.isBuffer(body)) {
+ return body;
+ }
+ if (body instanceof Uint8Array) {
+ return Buffer.from(body.buffer, body.byteOffset, body.byteLength);
+ }
+ if (typeof body !== 'string') {
+ throwAfterResponseTypeError(body);
+ }
+ return Buffer.from(body, encoding);
+}
+
+/*
+ * Maps afterResponse result to responseData, or null if the resource should be skipped.
+ * bufferedBodyPromise is set when the action read the body via getBody.
+ */
+async function buildAfterResponseData ({ result, responseData, bufferedBodyPromise }) {
+ if (result === null || result === undefined) {
return null;
}
- // Then stringify it.
- let body = null;
- if (data instanceof Uint8Array) {
- body = toBuffer(data).toString(encoding);
- } else if (typeof data === 'string') {
- body = data;
- } else {
- throwTypeError(result);
+ if (typeof result !== 'object' || Array.isArray(result) || result instanceof Error) {
+ throwAfterResponseTypeError(result);
+ }
+
+ const hasBody = 'body' in result;
+ if (hasBody && (result.body === null || result.body === undefined)) {
+ return null; // no body - skip the resource, same as returning null
}
- return {
- body,
- encoding,
- metadata: result.metadata || data.metadata || null
+ const overrides = {
+ encoding: result.encoding || responseData.encoding,
+ metadata: result.metadata || null
};
+
+ if (hasBody) {
+ overrides.stream = Readable.from(afterResponseBodyToBuffer(result.body, overrides.encoding));
+ } else if (bufferedBodyPromise !== null) {
+ // body was read by the action - original stream is consumed, serve the buffered copy
+ overrides.stream = Readable.from(await bufferedBodyPromise);
+ }
+
+ if (overrides.stream) {
+ overrides.readBody = () => readStreamToBuffer(overrides.stream);
+ }
+
+ return extend(responseData, overrides);
+}
+
+async function applyAfterResponse ({ responseData, headers, afterResponse }) {
+ const originalStream = responseData.stream;
+ // memoize the promise, not the value - concurrent getBody calls must not
+ // both start reading the same one-shot stream
+ let bufferedBodyPromise = null;
+ let actionReturned = false;
+ const getBody = () => {
+ if (actionReturned) {
+ // the live stream may already be consumed by storage - reading it here would corrupt both
+ return Promise.reject(new Error(`getBody() can only be called while the afterResponse action is running (${responseData.url})`));
+ }
+ return bufferedBodyPromise ??= responseData.readBody();
+ };
+
+ let transformed = null;
+ try {
+ const result = await afterResponse({
+ response: {
+ url: responseData.url,
+ statusCode: responseData.statusCode,
+ headers,
+ getBody
+ }
+ });
+ actionReturned = true;
+ transformed = await buildAfterResponseData({ result, responseData, bufferedBodyPromise });
+ return transformed;
+ } finally {
+ actionReturned = true; // also when the action threw
+ if (!transformed || transformed.stream !== originalStream) {
+ // the live stream is not handed over to the caller - destroy it to release
+ // the socket (no-op when the body was already fully read via getBody)
+ originalStream.destroy();
+ }
+ }
}
-async function getRequest ({url, referer, options = {}, afterResponse = defaultResponseHandler}) {
- const requestOptions = extend(options, {});
+/*
+ * Resolves when response headers arrive with
+ * { url, statusCode, mimeType, encoding, metadata, stream, readBody }
+ * The body is NOT buffered - it is available as a Readable in stream, or via readBody()
+ * which buffers it with got retries for mid-body errors. Use one or the other.
+ * Returns null if the resource should be skipped (afterResponse returned null).
+ */
+async function getRequest ({ url, referer, options = {}, afterResponse }) {
+ const requestOptions = extend(options, {}); // got v16 does not allow url in options
+ delete requestOptions.responseType; // meaningless for streams
if (referer) {
- requestOptions.headers = requestOptions.headers || {};
- requestOptions.headers.referer = referer;
+ requestOptions.headers = extend(requestOptions.headers, { referer });
}
logger.debug(`[request] sending request for url ${url}, referer ${referer}`);
- const response = await got(url, requestOptions);
- // got returns Uint8Array for responseType 'buffer', keep Buffer for afterResponse handlers
- if (response.body instanceof Uint8Array) {
- response.body = toBuffer(response.body);
- }
+ const { response, stream } = await requestWithRetry(url, requestOptions);
logger.debug(`[request] received response for ${response.url}, statusCode ${response.statusCode}`);
- const responseHandlerResult = transformResult(await afterResponse({response}));
- if (!responseHandlerResult) {
- return null;
- }
- return {
+ const responseData = {
url: response.url,
+ statusCode: response.statusCode,
mimeType: getMimeType(response.headers['content-type']),
- body: responseHandlerResult.body,
- metadata: responseHandlerResult.metadata,
- encoding: responseHandlerResult.encoding
+ encoding: extractEncodingFromHeader(response.headers),
+ metadata: null,
+ stream,
+ readBody: () => readBodyWithRetry(url, stream)
};
+
+ if (!afterResponse) {
+ return responseData;
+ }
+
+ return applyAfterResponse({ responseData, headers: response.headers, afterResponse });
}
export default {
- get: getRequest,
- getEncoding,
- transformResult
+ get: getRequest
};
diff --git a/lib/resource.js b/lib/resource.js
index cb94435f..20dcb5fb 100644
--- a/lib/resource.js
+++ b/lib/resource.js
@@ -1,3 +1,4 @@
+import { Readable } from 'stream';
import types from './config/resource-types.js';
class Resource {
@@ -57,6 +58,46 @@ class Resource {
this.text = text;
}
+ setContentStream (stream) {
+ this.contentStream = stream;
+ this.contentStreamTaken = false;
+ }
+
+ setContentBuffer (buffer) {
+ this.contentBuffer = buffer;
+ }
+
+ /**
+ * Returns resource content as a Readable stream.
+ * For buffered (html/css text or contentBuffer) resources a fresh stream is created on each call.
+ * For streamed resources the live response stream is returned - it can be consumed only once,
+ * so it is handed out only once: a second call throws instead of returning an ended stream.
+ * Returns null when content was already cleared (see clearContent).
+ */
+ getContentStream () {
+ if (this.contentStream) {
+ if (this.contentStreamTaken) {
+ throw new Error(`Content stream of ${this} was already taken - it is a live response stream which can be consumed only once. Buffer it in your action if you need the content more than once`);
+ }
+ this.contentStreamTaken = true;
+ return this.contentStream;
+ }
+ if (this.contentBuffer) {
+ return Readable.from(this.contentBuffer);
+ }
+ if (typeof this.text === 'string') {
+ return Readable.from(Buffer.from(this.text, this.encoding));
+ }
+ return null;
+ }
+
+ clearContent () {
+ this.text = null;
+ this.contentStream = null;
+ this.contentStreamTaken = false;
+ this.contentBuffer = null;
+ }
+
getDepth () {
return this.depth;
}
diff --git a/lib/scraper.js b/lib/scraper.js
index ff2d810d..4e4cc28f 100644
--- a/lib/scraper.js
+++ b/lib/scraper.js
@@ -3,6 +3,7 @@ import logger from './logger.js';
import defaults from './config/defaults.js';
import recursiveSources from './config/recursive-sources.js';
import Resource from './resource.js';
+import { RequestError } from 'got';
import request from './request.js';
import ResourceHandler from './resource-handler/index.js';
import {
@@ -14,6 +15,7 @@ import {
import * as utils from './utils/index.js';
const { extend, union, urlsEqual, getTypeByMime, getTypeByFilename, series } = utils;
+import { buffer as readStreamToBuffer } from 'stream/consumers';
import NormalizedUrlMap from './utils/normalized-url-map.js';
const actionNames = [
@@ -52,6 +54,7 @@ class Scraper {
this.requestedResourcePromises = new NormalizedUrlMap(); // Map url -> request promise
this.loadedResources = new NormalizedUrlMap(); // Map url -> resource
this.requestQueue = new PromiseQueue({concurrency: this.options.requestConcurrency});
+ this.fatalError = null; // first error which should abort the scrape, see Scraper.waitForLoad
}
normalizeOptions (options) {
@@ -122,15 +125,61 @@ class Scraper {
}
}
- async saveResource (resource) {
+ async persistResource (resource) {
resource.setSaved();
try {
await this.resourceHandler.handleResource(resource);
+
+ if (resource.contentStream && this.actions.saveResource.length > 1) {
+ // live stream can be consumed only once - buffer content (as a Buffer,
+ // strings cap at ~512 MiB) so each saveResource action can read it
+ resource.setContentBuffer(await readStreamToBuffer(resource.contentStream));
+ resource.setContentStream(null);
+ }
+
logger.info('saving resource ' + resource + ' to fs');
- await this.runActions('saveResource', {resource});
- // ignore promise here, just notifying external code about resource saved
- this.runActions('onResourceSaved', {resource});
+ await this.runSaveResourceActions(resource);
+ } finally {
+ const stream = resource.contentStream;
+ if (stream && !stream.readableEnded) {
+ // action resolved without consuming the stream - release the socket
+ stream.destroy();
+ }
+ resource.clearContent();
+ }
+
+ // ignore promise here, just notifying external code about resource saved
+ this.runActions('onResourceSaved', {resource});
+ }
+
+ /*
+ * Runs saveResource actions. For a live stream the save also fails when the stream
+ * itself errors (e.g. connection reset mid-body), so that an action which does not
+ * propagate stream errors (e.g. .pipe() without an error handler) cannot hang the scrape.
+ */
+ async runSaveResourceActions (resource) {
+ const stream = resource.contentStream;
+ if (!stream) {
+ return this.runActions('saveResource', {resource});
+ }
+
+ let onStreamError;
+ const streamFailed = new Promise((resolve, reject) => {
+ onStreamError = reject;
+ stream.once('error', onStreamError);
+ });
+
+ try {
+ return await Promise.race([this.runActions('saveResource', {resource}), streamFailed]);
+ } finally {
+ stream.off('error', onStreamError);
+ }
+ }
+
+ async saveResource (resource) {
+ try {
+ await this.persistResource(resource);
} catch (err) {
logger.warn('failed to save resource ' + resource);
await this.handleError(err, resource);
@@ -141,54 +190,98 @@ class Scraper {
const self = this;
const url = resource.getUrl();
+ // the queue task encompasses response headers and full body consumption
+ // (buffering for html/css, streaming to storage for other types)
+ // so that requestConcurrency bounds concurrent transfers
const requestPromise = Promise.resolve()
- .then(async () => {
+ .then(() => {
const referer = resource.parent ? resource.parent.getUrl() : null;
- return this.requestQueue.add(async () => {
- const {requestOptions} = await this.runActions('beforeRequest', {resource, requestOptions: this.options.request});
- return request.get({
+ return this.requestQueue.add(async function requestCompleted () {
+ const {requestOptions} = await self.runActions('beforeRequest', {resource, requestOptions: self.options.request});
+ const responseData = await request.get({
url,
referer,
options: requestOptions,
- afterResponse: this.actions.afterResponse.length ? this.runActions.bind(this, 'afterResponse') : undefined
+ afterResponse: self.actions.afterResponse.length ? self.runActions.bind(self, 'afterResponse') : undefined
});
- });
- }).then(async function requestCompleted (responseData) {
- if (!responseData) {
- logger.debug('no response returned for url ' + url);
- return null;
- }
-
- if (!urlsEqual(responseData.url, url)) { // Url may be changed in redirects
- logger.debug('url changed. old url = ' + url + ', new url = ' + responseData.url);
- if (self.requestedResourcePromises.has(responseData.url)) {
- return self.requestedResourcePromises.get(responseData.url);
+ if (!responseData) {
+ logger.debug('no response returned for url ' + url);
+ return null;
}
- resource.setUrl(responseData.url);
- self.requestedResourcePromises.set(responseData.url, requestPromise);
- }
-
- resource.setType(getTypeByMime(responseData.mimeType));
+ if (!urlsEqual(responseData.url, url)) { // Url may be changed in redirects
+ logger.debug('url changed. old url = ' + url + ', new url = ' + responseData.url);
- const { filename } = await self.runActions('generateFilename', { resource, responseData });
- resource.setFilename(filename);
+ if (self.requestedResourcePromises.has(responseData.url)) {
+ responseData.stream.destroy(); // duplicate of an already requested resource - release the socket
+ // awaiting the other resource's promise here would deadlock at low concurrency
+ // because it may still be waiting for this queue slot - resolve it outside the queue task
+ return {redirectedTo: responseData.url};
+ }
- // if type was not determined by mime we can try to get it from filename after it was generated
- if (!resource.getType()) {
- resource.setType(getTypeByFilename(filename));
- }
+ resource.setUrl(responseData.url);
+ self.requestedResourcePromises.set(responseData.url, requestPromise);
+ }
- if (responseData.metadata) {
- resource.setMetadata(responseData.metadata);
+ resource.setType(getTypeByMime(responseData.mimeType));
+
+ const { stream, readBody, ...responseInfo } = responseData;
+
+ try {
+ const { filename } = await self.runActions('generateFilename', { resource, responseData: responseInfo });
+ resource.setFilename(filename);
+
+ // if type was not determined by mime we can try to get it from filename after it was generated
+ if (!resource.getType()) {
+ resource.setType(getTypeByFilename(filename));
+ }
+
+ if (responseData.metadata) {
+ resource.setMetadata(responseData.metadata);
+ }
+
+ resource.setEncoding(responseData.encoding);
+
+ if (resource.isHtml() || resource.isCss()) {
+ // html and css need full text to find and update links to child resources,
+ // they are saved after their children are resolved - see Scraper.waitForLoad.
+ // readBody retries mid-body errors (the stream is not handed over to anyone)
+ resource.setText((await readBody()).toString(responseData.encoding));
+ self.loadResource(resource);
+ } else {
+ // other types need no modification - stream them to storage immediately
+ resource.setContentStream(stream);
+ self.loadResource(resource);
+ try {
+ await self.saveResource(resource);
+ } catch (err) {
+ // saveResource already handled the error (warn log + onResourceError)
+ // and rethrew because ignoreErrors is false. This task's rejection
+ // would be discarded by Promise.allSettled in downloadChildrenResources,
+ // so decide here whether the scrape must abort:
+ // - a network failure while reading the body (got RequestError) of a
+ // child resource is a request error - same as a child failing before
+ // the body arrived, it does not abort the scrape
+ // - anything else (storage/plugin failure, or a root resource failing)
+ // is remembered so that waitForLoad aborts the scrape
+ const isChildNetworkError = err instanceof RequestError && Boolean(resource.parent);
+ if (!isChildNetworkError) {
+ self.fatalError = self.fatalError || err;
+ }
+ }
+ }
+ } catch (err) {
+ stream.destroy(); // the stream was never handed over to a consumer - release the socket
+ throw err;
+ }
+ return resource;
+ });
+ }).then(function resolveRedirectDuplicate (result) {
+ if (result && result.redirectedTo) {
+ return self.requestedResourcePromises.get(result.redirectedTo);
}
-
- resource.setEncoding(responseData.encoding);
- resource.setText(responseData.body);
-
- self.loadResource(resource); // Add resource to list for future downloading, see Scraper.waitForLoad
- return resource;
+ return result;
}).catch(function handleError (err) {
logger.error('failed to request resource ' + resource);
return self.handleError(err, resource);
@@ -244,6 +337,13 @@ class Scraper {
// 1. Get all not saved resources and save them
// 2. Recursion if any new not saved resource were added during this time. If not, loading is done.
async waitForLoad () {
+ if (this.fatalError) {
+ // saving a streamed resource failed inside a request queue task - its rejection
+ // was discarded by Promise.allSettled in downloadChildrenResources, re-throw
+ // it here to abort the scrape (see Scraper.createNewRequest)
+ throw this.fatalError;
+ }
+
const resourcesToSave = Array.from(this.loadedResources.values()).filter((r) => !r.isSaved());
const loadingIsFinished = resourcesToSave.length === 0;
diff --git a/lib/utils/fs.js b/lib/utils/fs.js
index 38618024..497cd2a9 100644
--- a/lib/utils/fs.js
+++ b/lib/utils/fs.js
@@ -1,13 +1,15 @@
import path from 'path';
import fs from 'fs/promises';
+import { createWriteStream } from 'fs';
+import { pipeline } from 'stream/promises';
-async function outputFile (file, data, encoding) {
+async function outputFileStream (file, stream) {
const dir = path.dirname(file);
await fs.mkdir(dir, { recursive: true});
- return fs.writeFile(file, data, { encoding: encoding });
+ return pipeline(stream, createWriteStream(file));
}
export {
- outputFile
+ outputFileStream
};
diff --git a/package.json b/package.json
index 94a0541f..a9504806 100644
--- a/package.json
+++ b/package.json
@@ -1,6 +1,6 @@
{
"name": "website-scraper",
- "version": "6.0.0",
+ "version": "7.0.0",
"description": "Download website to a local directory (including all css, images, js, etc.)",
"readmeFilename": "README.md",
"type": "module",
@@ -11,6 +11,7 @@
"scripts": {
"test": "c8 --all --reporter=text --reporter=lcov mocha --recursive --timeout 7000 ./test/unit/ ./test/functional",
"test-e2e": "mocha --timeout 300000 ./test/e2e/*-test.js",
+ "benchmark:memory": "node benchmark/memory.mjs",
"eslint": "eslint lib index.mjs"
},
"repository": {
diff --git a/test/functional/base/check-it-works.js b/test/functional/base/check-it-works.js
index 758b8af8..8ff3b11b 100644
--- a/test/functional/base/check-it-works.js
+++ b/test/functional/base/check-it-works.js
@@ -31,7 +31,9 @@ describe('Functional: check it works', function() {
return scrape(options).then((result) => {
result[0].url.should.be.eql('http://example.com/');
result[0].filename.should.be.eql('index.html');
- result[0].text.should.be.eql('TEST PROMISES');
+ // content is not retained in memory after save - read the saved file instead
+ (result[0].text === null).should.be.true;
+ fs.readFileSync(testDirname + '/index.html').toString().should.be.eql('TEST PROMISES');
});
});
});
diff --git a/test/functional/error-handling/error-handling.test.js b/test/functional/error-handling/error-handling.test.js
index 5cd588b6..2affe3a8 100644
--- a/test/functional/error-handling/error-handling.test.js
+++ b/test/functional/error-handling/error-handling.test.js
@@ -91,6 +91,30 @@ describe('Functional error handling', function() {
});
});
+ describe('Error before anything was saved', function () {
+ it('should not remove the directory when the fs plugin wrote nothing to it', function () {
+ nock.cleanAll();
+ nock('http://example.com/').get('/').replyWithError('REQUEST FAILED!');
+
+ class RequestLogPlugin {
+ apply (registerAction) {
+ registerAction('beforeRequest', ({requestOptions}) => {
+ fs.mkdirSync(testDirname, {recursive: true});
+ fs.appendFileSync(testDirname + '/requests.log', 'request\n');
+ return {requestOptions};
+ });
+ }
+ }
+
+ return scrape({...options, ignoreErrors: false, plugins: [new RequestLogPlugin()]}).then(function () {
+ false.should.be.true;
+ }).catch(function (err) {
+ err.message.should.be.eql('REQUEST FAILED!');
+ fs.existsSync(testDirname + '/requests.log').should.be.eql(true);
+ });
+ });
+ });
+
describe('Resource Handler Error', function () {
let scraper;
let handleResourceStub;
diff --git a/test/functional/request-response-customizations/after-response-action.test.js b/test/functional/request-response-customizations/after-response-action.test.js
index 55a33cb5..be4036fe 100644
--- a/test/functional/request-response-customizations/after-response-action.test.js
+++ b/test/functional/request-response-customizations/after-response-action.test.js
@@ -32,8 +32,8 @@ describe('Functional: afterResponse action in plugin', function() {
if (response.statusCode === 404) {
return null;
} else {
+ // no body returned - the resource keeps streaming
return {
- body: response.body,
metadata: {
headers: response.headers,
someOtherData: [ 1, 2, 3 ]
diff --git a/test/functional/streaming/streaming.test.js b/test/functional/streaming/streaming.test.js
new file mode 100644
index 00000000..07272c17
--- /dev/null
+++ b/test/functional/streaming/streaming.test.js
@@ -0,0 +1,441 @@
+import * as chai from 'chai';
+const should = chai.should();
+import '../../utils/assertions.js';
+import nock from 'nock';
+import fs from 'fs';
+import path from 'path';
+import http from 'http';
+import { buffer as readStreamBuffer } from 'stream/consumers';
+import scrape from 'website-scraper';
+
+const testDirname = './test/functional/streaming/.tmp';
+
+// deterministic multi-MB binary payload
+function generateBigBinary (sizeInBytes) {
+ const chunk = Buffer.alloc(64 * 1024);
+ for (let i = 0; i < chunk.length; i++) {
+ chunk[i] = (i * 7 + 13) % 256;
+ }
+ return Buffer.concat(Array.from({length: Math.ceil(sizeInBytes / chunk.length)}, () => chunk), sizeInBytes);
+}
+
+describe('Functional: streaming resources to storage', function () {
+
+ beforeEach(function () {
+ nock.cleanAll();
+ nock.disableNetConnect();
+ });
+
+ afterEach(function () {
+ nock.cleanAll();
+ nock.enableNetConnect();
+ fs.rmSync(testDirname, {recursive: true, force: true});
+ });
+
+ it('should save a large binary resource byte-identical', async function () {
+ const pngBody = generateBigBinary(5 * 1024 * 1024);
+ nock('http://example.com/').get('/').reply(200, '
', {'content-type': 'text/html'});
+ nock('http://example.com/').get('/a.png').reply(200, pngBody, {'content-type': 'image/png'});
+
+ await scrape({
+ urls: ['http://example.com/'],
+ directory: testDirname
+ });
+
+ const savedBytes = fs.readFileSync(testDirname + '/images/a.png');
+ savedBytes.equals(pngBody).should.be.eql(true);
+ });
+
+ it('should free resource content from memory after save', async function () {
+ nock('http://example.com/').get('/').reply(200, '', {'content-type': 'text/html'});
+ nock('http://example.com/').get('/a.png').reply(200, 'image content', {'content-type': 'image/png'});
+
+ const result = await scrape({
+ urls: ['http://example.com/'],
+ directory: testDirname
+ });
+
+ (result[0].getText() === null).should.be.true;
+ should.not.exist(result[0].getContentStream());
+
+ const child = result[0].children.find((c) => c.url === 'http://example.com/a.png');
+ (child.getText() === null).should.be.true;
+ should.not.exist(child.getContentStream());
+
+ // content is on disk instead
+ fs.readFileSync(testDirname + '/images/a.png').toString().should.be.eql('image content');
+ });
+
+ it('should save body returned from afterResponse (buffered path)', async function () {
+ nock('http://example.com/').get('/a.png').reply(200, 'original content', {'content-type': 'image/png'});
+
+ class ReplaceBodyPlugin {
+ apply (add) {
+ add('afterResponse', async ({response}) => {
+ const body = await response.getBody();
+ return { body: Buffer.concat([body, Buffer.from(' + replaced')]) };
+ });
+ }
+ }
+
+ await scrape({
+ urls: [{url: 'http://example.com/a.png', filename: 'a.png'}],
+ directory: testDirname,
+ plugins: [new ReplaceBodyPlugin()]
+ });
+
+ fs.readFileSync(testDirname + '/images/a.png').toString().should.be.eql('original content + replaced');
+ });
+
+ it('should keep resource streaming and attach metadata when afterResponse returns no body', async function () {
+ const pngBody = generateBigBinary(1024 * 1024);
+ nock('http://example.com/').get('/a.png').reply(200, pngBody, {'content-type': 'image/png'});
+
+ class MetadataPlugin {
+ apply (add) {
+ add('afterResponse', ({response}) => {
+ return { metadata: { statusCode: response.statusCode } };
+ });
+ }
+ }
+
+ const result = await scrape({
+ urls: [{url: 'http://example.com/a.png', filename: 'a.png'}],
+ directory: testDirname,
+ plugins: [new MetadataPlugin()]
+ });
+
+ result[0].metadata.should.be.eql({statusCode: 200});
+ fs.readFileSync(testDirname + '/images/a.png').equals(pngBody).should.be.eql(true);
+ });
+
+ it('should provide full content to multiple saveResource actions', async function () {
+ const pngBody = generateBigBinary(1024 * 1024);
+ nock('http://example.com/').get('/a.png').reply(200, pngBody, {'content-type': 'image/png'});
+
+ const saved = [];
+ class MultiStoragePlugin {
+ apply (add) {
+ add('saveResource', async ({resource}) => {
+ saved.push(await readStreamBuffer(resource.getContentStream()));
+ });
+ add('saveResource', async ({resource}) => {
+ saved.push(await readStreamBuffer(resource.getContentStream()));
+ });
+ }
+ }
+
+ await scrape({
+ urls: [{url: 'http://example.com/a.png', filename: 'a.png'}],
+ directory: testDirname,
+ plugins: [new MultiStoragePlugin()]
+ });
+
+ saved.should.have.length(2);
+ saved[0].equals(pngBody).should.be.eql(true);
+ saved[1].equals(pngBody).should.be.eql(true);
+ });
+
+ it('should complete scraping when saveResource action does not consume the stream', async function () {
+ nock('http://example.com/').get('/a.png').reply(200, generateBigBinary(1024 * 1024), {'content-type': 'image/png'});
+
+ const seenResources = [];
+ class IgnoreContentPlugin {
+ apply (add) {
+ add('saveResource', async ({resource}) => {
+ seenResources.push(resource.url);
+ });
+ }
+ }
+
+ await scrape({
+ urls: [{url: 'http://example.com/a.png', filename: 'a.png'}],
+ directory: testDirname,
+ plugins: [new IgnoreContentPlugin()]
+ });
+
+ seenResources.should.be.eql(['http://example.com/a.png']);
+ });
+
+ it('should reject the scrape when saving a streamed child resource fails and ignoreErrors is false', async function () {
+ nock('http://example.com/').get('/').reply(200, '', {'content-type': 'text/html'});
+ nock('http://example.com/').get('/a.png').reply(200, 'image content', {'content-type': 'image/png'});
+
+ const errors = [];
+ class FailingStoragePlugin {
+ apply (add) {
+ add('saveResource', async ({resource}) => {
+ if (resource.getFilename().endsWith('.png')) {
+ throw new Error('SAVE FAILED');
+ }
+ });
+ add('onResourceError', ({error}) => errors.push(error.message));
+ }
+ }
+
+ try {
+ await scrape({
+ urls: ['http://example.com/'],
+ directory: testDirname,
+ ignoreErrors: false,
+ plugins: [new FailingStoragePlugin()]
+ });
+ throw new Error('expected scrape to reject');
+ } catch (err) {
+ err.message.should.be.eql('SAVE FAILED');
+ }
+ errors.should.be.eql(['SAVE FAILED']); // handled exactly once
+ });
+
+ // nock cannot simulate a connection dying mid-body, so these tests use a real local server
+ describe('mid-stream failures', function () {
+ let server, baseUrl;
+
+ beforeEach(function (done) {
+ nock.enableNetConnect(/127\.0\.0\.1|localhost/);
+ server = http.createServer((req, res) => {
+ switch (req.url) {
+ case '/':
+ res.writeHead(200, {'content-type': 'text/html'});
+ res.end('
');
+ break;
+ case '/fail.png':
+ res.writeHead(200, {'content-type': 'image/png'});
+ res.write('partial data');
+ setTimeout(() => res.destroy(), 10);
+ break;
+ default:
+ res.writeHead(200, {'content-type': 'image/png'});
+ res.end('fine content');
+ }
+ });
+ server.listen(0, '127.0.0.1', () => {
+ baseUrl = `http://127.0.0.1:${server.address().port}`;
+ done();
+ });
+ });
+
+ afterEach(function (done) {
+ server.close(done);
+ });
+
+ it('should continue scraping and leave no partial file when child fails mid-stream with ignoreErrors', async function () {
+ await scrape({
+ urls: [baseUrl + '/'],
+ directory: testDirname,
+ request: {retry: {limit: 0}},
+ ignoreErrors: true
+ });
+
+ fs.existsSync(testDirname + '/index.html').should.be.eql(true);
+ fs.existsSync(testDirname + '/images/fail.png').should.be.eql(false); // failed, partial file removed
+ fs.readFileSync(testDirname + '/images/fine.png').toString().should.be.eql('fine content');
+ });
+
+ it('should continue scraping when a child fails mid-stream without ignoreErrors (same as a failed child request)', async function () {
+ const errors = [];
+ class ErrorCollectorPlugin {
+ apply (add) {
+ add('onResourceError', ({resource, error}) => errors.push({url: resource.getUrl(), message: error.message}));
+ }
+ }
+
+ const result = await scrape({
+ urls: [baseUrl + '/'],
+ directory: testDirname,
+ request: {retry: {limit: 0}},
+ ignoreErrors: false,
+ plugins: [new ErrorCollectorPlugin()]
+ });
+
+ result.should.have.length(1);
+ fs.existsSync(testDirname + '/index.html').should.be.eql(true);
+ fs.existsSync(testDirname + '/images/fail.png').should.be.eql(false); // failed, partial file removed
+ fs.readFileSync(testDirname + '/images/fine.png').toString().should.be.eql('fine content');
+ errors.should.have.length(1); // handled exactly once
+ errors[0].url.should.be.eql(baseUrl + '/fail.png');
+ });
+
+ it('should fail the save instead of hanging when a saveResource action does not propagate stream errors', async function () {
+ const errors = [];
+ // replaces the default fs saver, so it is the only saveResource action and receives the live stream
+ class PipeWithoutErrorHandlingPlugin {
+ apply (add) {
+ add('beforeStart', () => fs.mkdirSync(testDirname, {recursive: true}));
+ add('saveResource', ({resource}) => new Promise((resolve) => {
+ const ws = fs.createWriteStream(path.join(testDirname, path.basename(resource.getFilename())));
+ // .pipe() does not forward source errors - 'finish' never fires if the body dies
+ resource.getContentStream().pipe(ws).on('finish', resolve);
+ }));
+ add('onResourceError', ({resource, error}) => errors.push({url: resource.getUrl(), message: error.message}));
+ }
+ }
+
+ const result = await scrape({
+ urls: [baseUrl + '/'],
+ request: {retry: {limit: 0}},
+ ignoreErrors: true,
+ plugins: [new PipeWithoutErrorHandlingPlugin()]
+ });
+
+ result.should.have.length(1);
+ fs.readFileSync(testDirname + '/fine.png').toString().should.be.eql('fine content');
+ errors.should.have.length(1);
+ errors[0].url.should.be.eql(baseUrl + '/fail.png');
+ });
+
+ it('should reject and remove directory when a root resource fails mid-stream without ignoreErrors', async function () {
+ try {
+ await scrape({
+ urls: [
+ {url: baseUrl + '/fine.png', filename: 'fine.png'},
+ {url: baseUrl + '/fail.png', filename: 'fail.png'}
+ ],
+ directory: testDirname,
+ request: {retry: {limit: 0}},
+ ignoreErrors: false
+ });
+ throw new Error('expected scrape to reject');
+ } catch (err) {
+ err.message.should.not.be.eql('expected scrape to reject');
+ }
+
+ fs.existsSync(testDirname).should.be.eql(false);
+ });
+ });
+
+ describe('mid-body retries', function () {
+ let server, baseUrl, attempts;
+
+ beforeEach(function (done) {
+ nock.enableNetConnect(/127\.0\.0\.1|localhost/);
+ attempts = {};
+ server = http.createServer((req, res) => {
+ attempts[req.url] = (attempts[req.url] || 0) + 1;
+ const isHtml = req.url === '/';
+ res.writeHead(200, {'content-type': isHtml ? 'text/html' : 'image/png'});
+ if (attempts[req.url] === 1) {
+ res.write(isHtml ? '' : 'partial');
+ setTimeout(() => res.destroy(), 10);
+ } else {
+ res.end(isHtml ? '' : 'image content');
+ }
+ });
+ server.listen(0, '127.0.0.1', () => {
+ baseUrl = `http://127.0.0.1:${server.address().port}`;
+ done();
+ });
+ });
+
+ afterEach(function (done) {
+ server.close(done);
+ });
+
+ it('should retry html which fails mid-body, but not a streamed resource', async function () {
+ const errors = [];
+ class ErrorCollectorPlugin {
+ apply (add) {
+ add('onResourceError', ({resource}) => errors.push(resource.getUrl()));
+ }
+ }
+
+ await scrape({
+ urls: [baseUrl + '/'],
+ directory: testDirname,
+ request: {retry: {limit: 1}},
+ ignoreErrors: false,
+ plugins: [new ErrorCollectorPlugin()]
+ });
+
+ attempts['/'].should.be.eql(2); // html body retried and buffered from the 2nd attempt
+ fs.readFileSync(testDirname + '/index.html').toString().should.contain('
');
+ attempts['/a.png'].should.be.eql(1); // streamed to storage - no mid-body retry
+ fs.existsSync(testDirname + '/images/a.png').should.be.eql(false);
+ errors.should.be.eql([baseUrl + '/a.png']);
+ });
+ });
+
+ describe('stream cleanup when an action fails', function () {
+ let server, baseUrl, pngSocket;
+
+ beforeEach(function (done) {
+ nock.enableNetConnect(/127\.0\.0\.1|localhost/);
+ server = http.createServer((req, res) => {
+ if (req.url === '/a.png') {
+ // never ends the body - the socket stays open unless the client destroys it
+ pngSocket = req.socket;
+ res.writeHead(200, {'content-type': 'image/png'});
+ res.write('partial data');
+ } else {
+ res.writeHead(200, {'content-type': 'text/html'});
+ res.end('');
+ }
+ });
+ server.listen(0, '127.0.0.1', () => {
+ baseUrl = `http://127.0.0.1:${server.address().port}`;
+ done();
+ });
+ });
+
+ afterEach(function (done) {
+ if (pngSocket && !pngSocket.destroyed) {
+ pngSocket.destroy();
+ }
+ server.close(done);
+ });
+
+ it('should destroy the response stream when generateFilename throws', async function () {
+ class FailingFilenamePlugin {
+ apply (add) {
+ add('generateFilename', ({resource}) => {
+ if (resource.getUrl().endsWith('.png')) {
+ throw new Error('FILENAME FAILED');
+ }
+ return {filename: 'index.html'};
+ });
+ }
+ }
+
+ await scrape({
+ urls: [baseUrl + '/'],
+ directory: testDirname,
+ ignoreErrors: true,
+ plugins: [new FailingFilenamePlugin()]
+ });
+
+ // the failed resource's socket must be released, otherwise the process never exits
+ await new Promise((resolve, reject) => {
+ const startedAt = Date.now();
+ (function checkSocketClosed () {
+ if (pngSocket.destroyed) {
+ return resolve();
+ }
+ if (Date.now() - startedAt > 2000) {
+ return reject(new Error('response socket was not destroyed'));
+ }
+ setTimeout(checkSocketClosed, 20);
+ })();
+ });
+ });
+ });
+
+ it('should handle redirect to an already requested resource with requestConcurrency 1', async function () {
+ this.timeout(3000); // a deadlock in the request queue would exceed this
+
+ nock('http://example.com/').get('/').reply(200,
+ '
', {'content-type': 'text/html'});
+ nock('http://example.com/').get('/a.png').reply(200, 'a content', {'content-type': 'image/png'});
+ nock('http://example.com/').get('/b.png').reply(301, undefined, {location: 'http://example.com/a.png'});
+ nock('http://example.com/').get('/a.png').reply(200, 'a content', {'content-type': 'image/png'});
+
+ await scrape({
+ urls: ['http://example.com/'],
+ directory: testDirname,
+ requestConcurrency: 1
+ });
+
+ fs.readFileSync(testDirname + '/images/a.png').toString().should.be.eql('a content');
+ const html = fs.readFileSync(testDirname + '/index.html').toString();
+ html.should.contain('src="images/a.png"');
+ });
+});
diff --git a/test/unit/plugins.test.js b/test/unit/plugins.test.js
index e69de29b..4e550768 100644
--- a/test/unit/plugins.test.js
+++ b/test/unit/plugins.test.js
@@ -0,0 +1,158 @@
+import * as chai from 'chai';
+chai.should();
+import fs from 'fs';
+import { Readable } from 'stream';
+import SaveResourceToFileSystemPlugin from '../../lib/plugins/save-resource-to-fs-plugin.js';
+import Resource from '../../lib/resource.js';
+
+const testDirname = './test/unit/.plugins-test';
+
+function applyPlugin (plugin) {
+ const actions = {};
+ plugin.apply((name, handler) => actions[name] = handler);
+ return actions;
+}
+
+describe('SaveResourceToFileSystemPlugin', () => {
+ afterEach(() => {
+ fs.rmSync(testDirname, {recursive: true, force: true});
+ });
+
+ describe('beforeStart', () => {
+ it('should throw if directory is not a string', () => {
+ const actions = applyPlugin(new SaveResourceToFileSystemPlugin());
+ (() => actions.beforeStart({options: {directory: null}})).should.throw('Incorrect directory');
+ (() => actions.beforeStart({options: {directory: 42}})).should.throw('Incorrect directory');
+ });
+
+ it('should throw if directory already exists', () => {
+ fs.mkdirSync(testDirname, {recursive: true});
+ const actions = applyPlugin(new SaveResourceToFileSystemPlugin());
+ (() => actions.beforeStart({options: {directory: testDirname}})).should.throw(/Directory (.*?) exists/);
+ });
+ });
+
+ describe('saveResource', () => {
+ it('should save content stream to file creating nested directories', async () => {
+ const actions = applyPlugin(new SaveResourceToFileSystemPlugin());
+ actions.beforeStart({options: {directory: testDirname}});
+
+ const resource = new Resource('http://example.com/a.png', 'images/nested/a.png');
+ const content = Buffer.from([0xff, 0x00, 0xab, 0x10]);
+ resource.setContentStream(Readable.from(content));
+
+ await actions.saveResource({resource});
+
+ fs.readFileSync(testDirname + '/images/nested/a.png').should.be.eql(content);
+ });
+
+ it('should save buffered text resources using resource encoding', async () => {
+ const actions = applyPlugin(new SaveResourceToFileSystemPlugin());
+ actions.beforeStart({options: {directory: testDirname}});
+
+ const resource = new Resource('http://example.com/', 'index.html');
+ resource.setEncoding('utf8');
+ resource.setText('тест');
+
+ await actions.saveResource({resource});
+
+ fs.readFileSync(testDirname + '/index.html').toString('utf8').should.be.eql('тест');
+ });
+
+ it('should remove partially written file and rethrow when stream errors', async () => {
+ const actions = applyPlugin(new SaveResourceToFileSystemPlugin());
+ actions.beforeStart({options: {directory: testDirname}});
+
+ const resource = new Resource('http://example.com/a.png', 'a.png');
+ resource.setContentStream(new Readable({
+ read () {
+ this.push('partial data');
+ this.destroy(new Error('mid-stream failure'));
+ }
+ }));
+
+ try {
+ await actions.saveResource({resource});
+ throw new Error('expected saveResource to reject');
+ } catch (err) {
+ err.message.should.be.eql('mid-stream failure');
+ }
+ fs.existsSync(testDirname + '/a.png').should.be.eql(false);
+ });
+
+ it('should propagate the original save error when cleanup rm fails', async () => {
+ const actions = applyPlugin(new SaveResourceToFileSystemPlugin());
+ actions.beforeStart({options: {directory: testDirname}});
+
+ // first resource occupies "about" as a directory
+ const first = new Resource('http://example.com/about/', 'about/index.html');
+ first.setContentStream(Readable.from('page'));
+ await actions.saveResource({resource: first});
+
+ // second resource wants "about" as a plain file - createWriteStream fails with EISDIR,
+ // and the cleanup rm of a directory would fail too - it must not mask the save error
+ const second = new Resource('http://example.com/about', 'about');
+ second.setContentStream(Readable.from('other'));
+
+ try {
+ await actions.saveResource({resource: second});
+ throw new Error('expected saveResource to reject');
+ } catch (err) {
+ err.code.should.be.eql('EISDIR');
+ }
+ });
+ });
+
+ describe('error', () => {
+ it('should remove directory if resources were saved', async () => {
+ const actions = applyPlugin(new SaveResourceToFileSystemPlugin());
+ actions.beforeStart({options: {directory: testDirname}});
+
+ const resource = new Resource('http://example.com/', 'index.html');
+ resource.setContentStream(Readable.from('content'));
+ await actions.saveResource({resource});
+ fs.existsSync(testDirname).should.be.eql(true);
+
+ await actions.error();
+
+ fs.existsSync(testDirname).should.be.eql(false);
+ });
+
+ it('should not throw if nothing was saved', async () => {
+ const actions = applyPlugin(new SaveResourceToFileSystemPlugin());
+ actions.beforeStart({options: {directory: testDirname}});
+
+ await actions.error();
+
+ fs.existsSync(testDirname).should.be.eql(false);
+ });
+
+ it('should remove directories created by a failed save even when no resource was saved', async () => {
+ const actions = applyPlugin(new SaveResourceToFileSystemPlugin());
+ actions.beforeStart({options: {directory: testDirname}});
+
+ const resource = new Resource('http://example.com/a.png', 'images/a.png');
+ resource.setContentStream(new Readable({
+ read () {
+ this.destroy(new Error('mid-stream failure'));
+ }
+ }));
+ await actions.saveResource({resource}).catch(() => {});
+ fs.existsSync(testDirname).should.be.eql(true); // directories created before the stream failed
+
+ await actions.error();
+
+ fs.existsSync(testDirname).should.be.eql(false);
+ });
+
+ it('should not remove a pre-existing directory when beforeStart failed', async () => {
+ fs.mkdirSync(testDirname, {recursive: true});
+ const actions = applyPlugin(new SaveResourceToFileSystemPlugin());
+ (() => actions.beforeStart({options: {directory: testDirname}})).should.throw(/Directory (.*?) exists/);
+
+ await actions.error();
+
+ fs.existsSync(testDirname).should.be.eql(true);
+ });
+ });
+});
diff --git a/test/unit/request-test.js b/test/unit/request-test.js
index 750a7ecb..e99ec690 100644
--- a/test/unit/request-test.js
+++ b/test/unit/request-test.js
@@ -1,7 +1,11 @@
import * as chai from 'chai';
-chai.should();
+const should = chai.should();
import sinon from 'sinon';
import nock from 'nock';
+import http from 'http';
+import { pipeline } from 'stream/promises';
+import { Writable } from 'stream';
+import { text as readStreamText, buffer as readStreamBuffer } from 'stream/consumers';
import request from '../../lib/request.js';
describe('request', () => {
@@ -47,266 +51,337 @@ describe('request', () => {
});
});
- it('should call afterResponse with correct params', () => {
- const url = 'http://example.com';
- const scope = nock(url).get('/').reply(200, 'TEST BODY');
- let handlerStub = sinon.stub().resolves('');
+ it('should return object with url, statusCode, mimeType, encoding and stream with the body', async () => {
+ const url = 'http://www.google.com';
+ nock(url).get('/').reply(200, 'Hello from Google!', {
+ 'content-type': 'text/html; charset=utf-8'
+ });
- return request.get({url, afterResponse: handlerStub}).then(() => {
- scope.isDone().should.eql(true);
- handlerStub.calledOnce.should.eql(true);
- const afterResponseArgs = handlerStub.getCall(0).args[0];
- afterResponseArgs.response.body.should.eql('TEST BODY');
- afterResponseArgs.response.headers.should.eql({});
+ const data = await request.get({url});
+ data.should.have.property('url');
+ data.should.have.property('statusCode');
+ data.should.have.property('mimeType');
+ data.should.have.property('stream');
+ data.url.should.eql('http://www.google.com/');
+ data.statusCode.should.eql(200);
+ data.mimeType.should.eql('text/html');
+ data.encoding.should.eql('utf8');
+ (data.metadata === null).should.be.true;
+
+ const body = await readStreamText(data.stream);
+ body.should.eql('Hello from Google!');
+ });
+
+ it('should return utf8 encoding when charset in content-type header is uppercase', async () => {
+ const url = 'http://www.google.com';
+ nock(url).get('/').reply(200, 'Hello from Google!', {
+ 'content-type': 'text/html; charset=UTF-8'
});
+
+ const data = await request.get({url});
+ data.encoding.should.eql('utf8');
});
- it('should pass response body to afterResponse as Buffer', () => {
- const url = 'http://example.com';
- nock(url).get('/').reply(200, 'TEST BODY');
- const handlerStub = sinon.stub().resolves('');
+ it('should return mimeType = null and binary encoding if content-type header was not found in response', async () => {
+ const url = 'http://www.google.com';
+ nock(url).get('/').reply(200, 'Hello from Google!', {});
- return request.get({url, options: {responseType: 'buffer'}, afterResponse: handlerStub}).then(() => {
- const { response } = handlerStub.getCall(0).args[0];
- Buffer.isBuffer(response.body).should.eql(true);
- response.body.toString().should.eql('TEST BODY');
- });
+ const data = await request.get({url});
+ data.url.should.eql('http://www.google.com/');
+ data.encoding.should.eql('binary');
+ data.should.have.property('mimeType', null);
});
- describe('transformResult from afterResponse', () => {
- it('should return object with body and metadata properties', () => {
- const url = 'http://example.com';
- nock(url).get('/').reply(200, 'TEST BODY');
- const handlerStub = sinon.stub().resolves({
- body: 'a',
- metadata: 'b',
- encoding: 'utf8'
- });
+ it('should retry transient network errors before response', async () => {
+ const url = 'http://retry.example.com';
+ nock(url).get('/').replyWithError(Object.assign(new Error('connection reset'), {code: 'ECONNRESET'}));
+ nock(url).get('/').reply(200, 'recovered');
- return request.get({url, afterResponse: handlerStub}).then((data) => {
- data.body.should.eql('a');
- data.metadata.should.eql('b');
- data.encoding.should.eql('utf8');
- });
- });
+ const data = await request.get({url, options: {retry: {limit: 1}}});
+ data.statusCode.should.eql(200);
+ const body = await readStreamText(data.stream);
+ body.should.eql('recovered');
+ });
- it('should return with metadata == null if metadata is not defined', () => {
- const url = 'http://example.com';
- nock(url).get('/').reply(200, 'TEST BODY');
- const handlerStub = sinon.stub().resolves({
- body: 'a'
- });
+ it('should reject when request fails without retries left', async () => {
+ const url = 'http://fail.example.com';
+ nock(url).get('/').replyWithError(Object.assign(new Error('connection reset'), {code: 'ECONNRESET'}));
+
+ try {
+ await request.get({url, options: {retry: {limit: 0}}});
+ throw new Error('expected request.get to reject');
+ } catch (err) {
+ err.code.should.eql('ECONNRESET');
+ }
+ });
- return request.get({url, afterResponse: handlerStub}).then((data) => {
- data.body.should.eql('a');
- (data.metadata === null).should.be.true;
- data.encoding.should.eql('binary');
+ // nock cannot simulate a connection dying mid-body, so these tests use a real local server
+ describe('mid-body errors', () => {
+ let server, url, attempts;
+
+ beforeEach((done) => {
+ attempts = 0;
+ server = http.createServer((req, res) => {
+ attempts++;
+ res.writeHead(200, {'content-type': 'text/html'});
+ if (attempts === 1) {
+ res.write('partial');
+ setTimeout(() => res.destroy(), 10);
+ } else {
+ res.end('full body');
+ }
+ });
+ server.listen(0, '127.0.0.1', () => {
+ url = `http://127.0.0.1:${server.address().port}/`;
+ done();
});
});
- it('should transform string result', () => {
- const url = 'http://example.com';
- nock(url).get('/').reply(200, 'TEST BODY');
- const handlerStub = sinon.stub().resolves('test body');
+ afterEach((done) => server.close(done));
- return request.get({url, afterResponse: handlerStub}).then((data) => {
- data.body.should.eql('test body');
- (data.metadata === null).should.be.true;
- });
+ it('should retry and return full body from readBody', async () => {
+ const data = await request.get({url, options: {retry: {limit: 1}}});
+ const body = await data.readBody();
+ body.toString().should.eql('full body');
+ attempts.should.eql(2);
});
- it('should be rejected if wrong result (no string nor object) returned', () => {
- const url = 'http://example.com';
- nock(url).get('/').reply(200, 'TEST BODY');
- const handlerStub = sinon.stub().resolves(['1', '2']);
-
- return request.get({url, afterResponse: handlerStub}).then(() => {
- true.should.eql(false);
- }).catch((e) => {
- e.should.be.instanceOf(Error);
- e.message.should.match(/Wrong response handler result. Expected string or object, but received/);
- });
+ it('should reject readBody with the request error when no retries are left', async () => {
+ const data = await request.get({url, options: {retry: {limit: 0}}});
+ try {
+ await data.readBody();
+ throw new Error('expected readBody to reject');
+ } catch (err) {
+ err.name.should.eql('ReadError');
+ }
+ attempts.should.eql(1);
});
- });
- it('should return object with url, body, mimeType properties', () => {
- const url = 'http://www.google.com';
- nock(url).get('/').reply(200, 'Hello from Google!', {
- 'content-type': 'text/html; charset=utf-8'
+ it('should not retry when the body is consumed as a stream', async () => {
+ const data = await request.get({url, options: {retry: {limit: 1}}});
+ try {
+ await pipeline(data.stream, new Writable({write (chunk, enc, cb) { cb(); }}));
+ throw new Error('expected pipeline to reject');
+ } catch (err) {
+ err.name.should.eql('ReadError');
+ }
+ attempts.should.eql(1);
});
- return request.get({url}).then((data) => {
- data.should.have.property('url');
- data.should.have.property('body');
- data.should.have.property('mimeType');
- data.url.should.eql('http://www.google.com/');
- data.body.should.eql('Hello from Google!');
- data.mimeType.should.eql('text/html');
- data.encoding.should.eql('utf8');
+ it('should retry inside afterResponse getBody', async () => {
+ let receivedBody;
+ const afterResponse = async ({response}) => {
+ receivedBody = await response.getBody();
+ return {};
+ };
+ const data = await request.get({url, options: {retry: {limit: 1}}, afterResponse});
+ receivedBody.toString().should.eql('full body');
+ (await readStreamText(data.stream)).should.eql('full body');
+ attempts.should.eql(2);
});
});
- it('should return mimeType = null if content-type header was not found in response', () => {
- let url = 'http://www.google.com';
- nock(url).get('/').reply(200, 'Hello from Google!', {});
+ describe('afterResponse', () => {
+ it('should call afterResponse with url, statusCode, headers and getBody', async () => {
+ const url = 'http://example.com';
+ const scope = nock(url).get('/').reply(200, 'TEST BODY', {'content-type': 'text/html'});
+ const handlerStub = sinon.stub().resolves({});
+
+ await request.get({url, afterResponse: handlerStub});
+ scope.isDone().should.eql(true);
+ handlerStub.calledOnce.should.eql(true);
- return request.get({url}).then((data) => {
- data.should.include.all.keys(['url', 'body', 'mimeType', 'encoding']);
- data.url.should.eql('http://www.google.com/');
- data.body.should.eql('Hello from Google!');
- data.encoding.should.eql('binary');
- data.should.have.property('mimeType', null);
+ const {response} = handlerStub.getCall(0).args[0];
+ response.url.should.eql('http://example.com/');
+ response.statusCode.should.eql(200);
+ response.headers.should.have.property('content-type', 'text/html');
+ response.getBody.should.be.a('function');
});
- });
-});
-describe('get encoding', () => {
- it('should return binary by default', () => {
- const result = request.getEncoding(null);
+ it('should provide body via getBody', async () => {
+ const url = 'http://example.com';
+ nock(url).get('/').reply(200, 'TEST BODY');
+
+ let receivedBody;
+ const afterResponse = async ({response}) => {
+ receivedBody = await response.getBody();
+ return {};
+ };
- result.should.eql('binary');
- });
+ const data = await request.get({url, afterResponse});
+ receivedBody.should.be.instanceOf(Buffer);
+ receivedBody.toString().should.eql('TEST BODY');
- it('should return binary when no content-type header supplies', () => {
- const result = request.getEncoding({
- headers: {}
+ // original stream was consumed by getBody - body must still be readable from result
+ const body = await readStreamText(data.stream);
+ body.should.eql('TEST BODY');
});
- result.should.eql('binary');
- });
+ it('should memoize getBody', async () => {
+ const url = 'http://example.com';
+ nock(url).get('/').reply(200, 'TEST BODY');
- it('should return binary when content type header doesn\'t include utf-8', () => {
- const result = request.getEncoding({
- headers: {}
+ const afterResponse = async ({response}) => {
+ const first = await response.getBody();
+ const second = await response.getBody();
+ first.should.be.equal(second);
+ return {};
+ };
+
+ await request.get({url, afterResponse});
});
- result.should.eql('binary');
- });
+ it('should return full body to concurrent getBody calls', async () => {
+ const url = 'http://example.com';
+ nock(url).get('/').reply(200, 'TEST BODY');
- it('should return binary when content type header doesn\'t include utf-8', () => {
- const result = request.getEncoding({
- headers: {
- 'content-type': 'text/html'
- }
+ const afterResponse = async ({response}) => {
+ const [first, second] = await Promise.all([response.getBody(), response.getBody()]);
+ first.toString().should.eql('TEST BODY');
+ second.should.be.equal(first);
+ return {};
+ };
+
+ const data = await request.get({url, afterResponse});
+ const body = await readStreamText(data.stream);
+ body.should.eql('TEST BODY');
});
- result.should.eql('binary');
- });
+ it('should reject getBody called after the action has returned', async () => {
+ const url = 'http://example.com';
+ nock(url).get('/').reply(200, 'TEST BODY');
- it('should return utf8 when content type includes utf-8', () => {
- const result = request.getEncoding({
- headers: {
- 'content-type': 'text/html; charset=utf-8'
+ let savedResponse;
+ const data = await request.get({url, afterResponse: ({response}) => {
+ savedResponse = response;
+ return {};
+ }});
+
+ try {
+ await savedResponse.getBody();
+ throw new Error('expected getBody to reject');
+ } catch (err) {
+ err.message.should.contain('getBody() can only be called while the afterResponse action is running');
}
+ // the stream handed over to the caller is untouched
+ (await readStreamText(data.stream)).should.eql('TEST BODY');
});
- result.should.eql('utf8');
- });
+ it('should return null and skip resource when action returns null', async () => {
+ const url = 'http://example.com';
+ nock(url).get('/').reply(200, 'TEST BODY');
+ const handlerStub = sinon.stub().resolves(null);
- it('should return utf8 response object includes it', () => {
- const result = request.getEncoding({
- encoding: 'utf8'
+ const data = await request.get({url, afterResponse: handlerStub});
+ should.not.exist(data);
});
- result.should.eql('utf8');
- });
-});
+ it('should replace body when action returns object with body', async () => {
+ const url = 'http://example.com';
+ nock(url).get('/').reply(200, 'TEST BODY');
+ const handlerStub = sinon.stub().resolves({
+ body: 'a',
+ metadata: 'b',
+ encoding: 'utf8'
+ });
-describe('transformResult', () => {
- it('should throw with weird shaped response', () => {
- try {
- request.transformResult([1,2,3]);
+ const data = await request.get({url, afterResponse: handlerStub});
+ data.metadata.should.eql('b');
+ data.encoding.should.eql('utf8');
+ const body = await readStreamText(data.stream);
+ body.should.eql('a');
+ });
- // We shouldn't get here.
- true.should.eql(false);
- } catch (e) {
- e.should.be.instanceOf(Error);
- e.message.should.eql('Wrong response handler result. Expected string or object, but received array');
- }
- });
+ it('should return null and skip resource when action returns object with null body', async () => {
+ const url = 'http://example.com';
+ nock(url).get('/').reply(200, 'TEST BODY');
+ const handlerStub = sinon.stub().resolves({body: null});
- it('should pass through error', () => {
- try {
- request.transformResult(new Error('Oh no'));
+ const data = await request.get({url, afterResponse: handlerStub});
+ should.not.exist(data);
+ });
- // We shouldn't get here.
- true.should.eql(false);
- } catch (e) {
- e.should.be.instanceOf(Error);
- e.message.should.eql('Oh no');
- }
- });
+ it('should decode string body with header-derived encoding when no encoding returned', async () => {
+ const url = 'http://example.com';
+ nock(url).get('/').reply(200, 'TEST BODY', {'content-type': 'text/html; charset=utf-8'});
+ const handlerStub = sinon.stub().resolves({body: 'café'});
- it('should throw with boolean input', () => {
- try {
- request.transformResult(true);
+ const data = await request.get({url, afterResponse: handlerStub});
+ data.encoding.should.eql('utf8');
+ const body = await readStreamBuffer(data.stream);
+ body.should.eql(Buffer.from('café', 'utf8'));
+ });
- // We shouldn't get here.
- true.should.eql(false);
- } catch (e) {
- e.should.be.instanceOf(Error);
- e.message.should.eql('Wrong response handler result. Expected string or object, but received boolean');
- }
- });
+ it('should support Buffer body', async () => {
+ const url = 'http://example.com';
+ nock(url).get('/').reply(200, 'TEST BODY');
+ const bodyBuffer = Buffer.from([0xff, 0x00, 0xab]);
+ const handlerStub = sinon.stub().resolves({body: bodyBuffer});
- it('should handle object', () => {
- const result = request.transformResult({
- body: 'SOME BODY',
- encoding: 'utf8',
- metadata: { foo: 'bar' }
+ const data = await request.get({url, afterResponse: handlerStub});
+ (data.metadata === null).should.be.true;
+ const body = await readStreamBuffer(data.stream);
+ body.should.eql(bodyBuffer);
});
- result.should.have.property('body', 'SOME BODY');
- result.should.have.property('encoding', 'utf8');
- result.should.have.property('metadata').that.eql({ foo: 'bar' });
- });
+ it('should support Uint8Array body', async () => {
+ const url = 'http://example.com';
+ nock(url).get('/').reply(200, 'TEST BODY');
+ const handlerStub = sinon.stub().resolves({body: new TextEncoder().encode('SOME BODY')});
- it('should handle object with empty body string', () => {
- const result = request.transformResult({
- body: '',
- encoding: 'utf8',
+ const data = await request.get({url, afterResponse: handlerStub});
+ const body = await readStreamText(data.stream);
+ body.should.eql('SOME BODY');
});
- result.should.have.property('body', '');
- result.should.have.property('encoding', 'utf8');
- result.should.have.property('metadata', null);
- });
+ it('should keep original body streaming when action returns metadata only', async () => {
+ const url = 'http://example.com';
+ nock(url).get('/').reply(200, 'TEST BODY');
+ const handlerStub = sinon.stub().resolves({metadata: {foo: 'bar'}});
- it('should handle object with defaults and buffer body', () => {
- const result = request.transformResult({
- body: Buffer.from('SOME BODY'),
+ const data = await request.get({url, afterResponse: handlerStub});
+ data.metadata.should.eql({foo: 'bar'});
+ const body = await readStreamText(data.stream);
+ body.should.eql('TEST BODY');
});
- result.should.have.property('body', 'SOME BODY');
- result.should.have.property('encoding', 'binary');
- result.should.have.property('metadata', null);
- });
+ it('should be rejected if wrong result (no null nor object) returned', async () => {
+ const url = 'http://example.com';
+ nock(url).get('/').reply(200, 'TEST BODY');
+ const handlerStub = sinon.stub().resolves(['1', '2']);
- it('should handle object with Uint8Array body', () => {
- const result = request.transformResult({
- body: new TextEncoder().encode('SOME BODY'),
+ try {
+ await request.get({url, afterResponse: handlerStub});
+ throw new Error('expected request.get to reject');
+ } catch (e) {
+ e.message.should.match(/Wrong afterResponse result. Expected null or object.*but received array/);
+ }
});
- result.should.have.property('body', 'SOME BODY');
- result.should.have.property('encoding', 'binary');
- result.should.have.property('metadata', null);
- });
-
- it('should handle raw string input', () => {
- const result = request.transformResult('SOME BODY');
+ it('should be rejected if string returned (removed in v7)', async () => {
+ const url = 'http://example.com';
+ nock(url).get('/').reply(200, 'TEST BODY');
+ const handlerStub = sinon.stub().resolves('test body');
- result.should.have.property('body', 'SOME BODY');
- result.should.have.property('encoding', 'binary');
- result.should.have.property('metadata', null);
- });
+ try {
+ await request.get({url, afterResponse: handlerStub});
+ throw new Error('expected request.get to reject');
+ } catch (e) {
+ e.message.should.match(/Wrong afterResponse result.*but received string/);
+ }
+ });
- it('should handle null input', () => {
- const result = request.transformResult(null);
- (result === null).should.be.true;
- });
+ it('should be rejected with error thrown by action', async () => {
+ const url = 'http://example.com';
+ nock(url).get('/').reply(200, 'TEST BODY');
+ const handlerStub = sinon.stub().rejects(new Error('action failed'));
- it('should handle undefined input', () => {
- const result = request.transformResult(undefined);
- (result === null).should.be.true;
+ try {
+ await request.get({url, afterResponse: handlerStub});
+ throw new Error('expected request.get to reject');
+ } catch (e) {
+ e.message.should.eql('action failed');
+ }
+ });
});
});
diff --git a/test/unit/resource-test.js b/test/unit/resource-test.js
index 545b0f71..476c451d 100644
--- a/test/unit/resource-test.js
+++ b/test/unit/resource-test.js
@@ -1,5 +1,7 @@
import * as chai from 'chai';
-chai.should();
+import { Readable } from 'stream';
+import { text as readStreamText, buffer as readStreamBuffer } from 'stream/consumers';
+const should = chai.should();
import '../../test/utils/assertions.js';
import Resource from '../../lib/resource.js';
@@ -33,4 +35,81 @@ describe('Resource', function() {
childOfChild.depth.should.eql(2);
});
});
+
+ describe('#getContentStream', function () {
+ it('should return null when resource has no content', function () {
+ const resource = new Resource('http://example.com');
+ should.not.exist(resource.getContentStream());
+ });
+
+ it('should return stream set with setContentStream', function () {
+ const resource = new Resource('http://example.com');
+ const stream = Readable.from(Buffer.from('binary data'));
+ resource.setContentStream(stream);
+ resource.getContentStream().should.be.equal(stream);
+ });
+
+ it('should throw when a live stream is requested a second time', function () {
+ const resource = new Resource('http://example.com/a.png');
+ resource.setContentStream(Readable.from(Buffer.from('binary data')));
+ resource.getContentStream();
+
+ (() => resource.getContentStream()).should.throw(/was already taken/);
+ });
+
+ it('should hand out a newly set live stream again', function () {
+ const resource = new Resource('http://example.com/a.png');
+ resource.setContentStream(Readable.from(Buffer.from('first')));
+ resource.getContentStream();
+
+ const second = Readable.from(Buffer.from('second'));
+ resource.setContentStream(second);
+ resource.getContentStream().should.be.equal(second);
+ });
+
+ it('should return fresh stream with text content on each call', async function () {
+ const resource = new Resource('http://example.com');
+ resource.setEncoding('utf8');
+ resource.setText('some text');
+
+ const firstText = await readStreamText(resource.getContentStream());
+ const secondText = await readStreamText(resource.getContentStream());
+ firstText.should.be.eql('some text');
+ secondText.should.be.eql('some text');
+ });
+
+ it('should encode text with resource encoding', async function () {
+ const resource = new Resource('http://example.com');
+ const originalBytes = Buffer.from([0xff, 0x00, 0xab, 0x10]);
+ resource.setText(originalBytes.toString('binary')); // default encoding is binary
+
+ const streamedBytes = await readStreamBuffer(resource.getContentStream());
+ streamedBytes.should.be.eql(originalBytes);
+ });
+
+ it('should return fresh stream with content buffer on each call', async function () {
+ const resource = new Resource('http://example.com');
+ const contentBuffer = Buffer.from([0xff, 0x00, 0xab, 0x10]);
+ resource.setContentBuffer(contentBuffer);
+
+ const firstBytes = await readStreamBuffer(resource.getContentStream());
+ const secondBytes = await readStreamBuffer(resource.getContentStream());
+ firstBytes.should.be.eql(contentBuffer);
+ secondBytes.should.be.eql(contentBuffer);
+ });
+ });
+
+ describe('#clearContent', function () {
+ it('should clear text, content stream and content buffer', function () {
+ const resource = new Resource('http://example.com');
+ resource.setText('some text');
+ resource.setContentStream(Readable.from('stream'));
+ resource.setContentBuffer(Buffer.from('buffer'));
+
+ resource.clearContent();
+
+ should.not.exist(resource.getText());
+ should.not.exist(resource.getContentStream());
+ });
+ });
});
diff --git a/test/unit/scraper-init-test.js b/test/unit/scraper-init-test.js
index d74605d8..1efc89b9 100644
--- a/test/unit/scraper-init-test.js
+++ b/test/unit/scraper-init-test.js
@@ -123,7 +123,6 @@ describe('Scraper initialization', function () {
s.options.request.should.deep.include({
throwHttpErrors: false,
- responseType: 'buffer',
decompress: true,
https: {
rejectUnauthorized: false
@@ -145,7 +144,6 @@ describe('Scraper initialization', function () {
s.options.request.should.deep.include({
throwHttpErrors: true,
- responseType: 'buffer',
decompress: true,
https: {
rejectUnauthorized: false
diff --git a/test/unit/scraper-test.js b/test/unit/scraper-test.js
index 2b9477b0..62a568d0 100644
--- a/test/unit/scraper-test.js
+++ b/test/unit/scraper-test.js
@@ -1,5 +1,5 @@
import * as chai from 'chai';
-chai.should();
+const should = chai.should();
import sinon from 'sinon';
import nock from 'nock';
import fs from 'fs';
@@ -96,6 +96,7 @@ describe('Scraper', () => {
urlFilter: () => { return true; },
plugins: [ new GenerateFilenamePlugin() ]
});
+ sinon.stub(s, 'persistResource').resolves();
const r = new Resource('http://example.com/a.png');
r.getDepth = sinon.stub().returns(2);
@@ -105,7 +106,8 @@ describe('Scraper', () => {
rr.should.be.eql(r);
rr.getUrl().should.be.eql('http://example.com/a.png');
rr.getFilename().should.not.be.empty;
- rr.getText().should.not.be.empty;
+ should.exist(rr.getContentStream());
+ s.persistResource.calledOnce.should.be.eql(true);
});
it('should return null if the urlFilter returns false', async () =>{
@@ -131,6 +133,7 @@ describe('Scraper', () => {
urlFilter: () => false,
plugins: [ new GenerateFilenamePlugin() ]
});
+ sinon.stub(s, 'persistResource').resolves();
const r = new Resource('http://example.com');
r.getDepth = sinon.stub().returns(0);
@@ -140,7 +143,7 @@ describe('Scraper', () => {
rr.should.be.eql(r);
rr.getUrl().should.be.eql('http://example.com');
rr.getFilename().should.not.be.empty;
- rr.getText().should.not.be.empty;
+ should.exist(rr.getContentStream());
});
});
@@ -153,6 +156,7 @@ describe('Scraper', () => {
directory: testDirname,
plugins: [ new GenerateFilenamePlugin() ]
});
+ sinon.stub(s, 'persistResource').resolves();
const r = new Resource('http://example.com/a.png');
r.getDepth = sinon.stub().returns(212);
@@ -162,7 +166,7 @@ describe('Scraper', () => {
rr.should.be.eql(r);
rr.getUrl().should.be.eql('http://example.com/a.png');
rr.getFilename().should.not.be.empty;
- rr.getText().should.not.be.empty;
+ should.exist(rr.getContentStream());
});
it('should request the resource if maxDepth is set and resource depth is less than maxDept', async () =>{
@@ -174,6 +178,7 @@ describe('Scraper', () => {
maxDepth: 3,
plugins: [ new GenerateFilenamePlugin() ]
});
+ sinon.stub(s, 'persistResource').resolves();
const r = new Resource('http://example.com/a.png');
r.getDepth = sinon.stub().returns(2);
@@ -183,7 +188,7 @@ describe('Scraper', () => {
rr.should.be.eql(r);
rr.getUrl().should.be.eql('http://example.com/a.png');
rr.getFilename().should.not.be.empty;
- rr.getText().should.not.be.empty;
+ should.exist(rr.getContentStream());
});
it('should request the resource if maxDepth is set and resource depth is equal to maxDept', async () =>{
@@ -195,6 +200,7 @@ describe('Scraper', () => {
maxDepth: 3,
plugins: [ new GenerateFilenamePlugin() ]
});
+ sinon.stub(s, 'persistResource').resolves();
const r = new Resource('http://example.com/a.png');
r.getDepth = sinon.stub().returns(3);
@@ -203,7 +209,7 @@ describe('Scraper', () => {
rr.should.be.eql(r);
rr.getUrl().should.be.eql('http://example.com/a.png');
rr.getFilename().should.not.be.empty;
- rr.getText().should.not.be.empty;
+ should.exist(rr.getContentStream());
});
it('should return null if maxDepth is set and resource depth is greater than maxDepth', async () =>{