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 e7697d4d..ba971aef 100644 --- a/MIGRATION.md +++ b/MIGRATION.md @@ -1,5 +1,60 @@ # Migration Guide +## From version 6 to 7 + +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. + +```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 +``` + +#### 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). + +```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, but only until response headers are received. A connection which dies mid-body is not retried. +* `responseType` in the `request` option is ignored (responses are streamed). + ## From version 4 to 5 #### ESM module diff --git a/README.md b/README.md index d4b89c99..9ce229bb 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 + * `encoding` (`binary` or `utf8`) is used to decode a string `body`, binary is used by default. * `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. + +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/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..894c8d26 100644 --- a/lib/plugins/save-resource-to-fs-plugin.js +++ b/lib/plugins/save-resource-to-fs-plugin.js @@ -1,6 +1,6 @@ import path from 'path'; import fs from 'fs'; -import { outputFile } from '../utils/fs.js'; +import { outputFileStream } from '../utils/fs.js'; class SaveResourceToFileSystemPlugin { apply (registerAction) { @@ -20,8 +20,12 @@ class SaveResourceToFileSystemPlugin { registerAction('saveResource', async ({resource}) => { const filename = path.join(absoluteDirectoryPath, resource.getFilename()); - const text = resource.getText(); - await outputFile(filename, text, resource.getEncoding()); + try { + await outputFileStream(filename, resource.getContentStream()); + } catch (err) { + await fs.promises.rm(filename, {force: true}); // don't leave partially written files + throw err; + } loadedResources.push(resource); }); diff --git a/lib/request.js b/lib/request.js index f66ce29c..9002c00e 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,13 @@ 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'; -} - -function throwTypeError (result) { +function throwAfterResponseTypeError (result) { let type = typeof result; if (result instanceof Error) { @@ -37,72 +23,143 @@ 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 getData (result) { - let data = result; - if (result && typeof result === 'object' && 'body' in result) { - data = result.body; - } - - return data; +/* + * Requests url with got.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. + * Mid-body errors after the stream was handed over to the consumer cannot be retried. + */ +function requestWithRetry (requestOptions) { + 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 ${requestOptions.url}, attempt ${retryCount}, reason: ${error.message}`); + attempt(createRetryStream()); + }; + const onResponse = (response) => { + stream.off('retry', onRetry); + stream.off('error', onError); + 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(got.stream(requestOptions)); + }); } -function transformResult (result) { - const encoding = getEncoding(result); - const data = getData(result); +async function applyAfterResponse ({ responseData, headers, afterResponse }) { + const originalStream = responseData.stream; + let bufferedBody = null; + + const getBody = async () => { + if (bufferedBody === null) { + bufferedBody = await readStreamToBuffer(originalStream); + } + return bufferedBody; + }; + + let result; + try { + result = await afterResponse({ + response: { + url: responseData.url, + statusCode: responseData.statusCode, + headers, + getBody + } + }); + } catch (err) { + originalStream.destroy(); + throw err; + } - // Check for no data - if (data === null || data === undefined) { + if (result === null || result === undefined) { + originalStream.destroy(); return null; } - // Then stringify it. - let body = null; - if (data instanceof Buffer) { - body = data.toString(encoding); - } else if (typeof data === 'string') { - body = data; - } else { - throwTypeError(result); + if (typeof result !== 'object' || Array.isArray(result) || result instanceof Error) { + originalStream.destroy(); + throwAfterResponseTypeError(result); } - return { - body, - encoding, - metadata: result.metadata || data.metadata || null - }; + const encoding = result.encoding || responseData.encoding; + const metadata = result.metadata || null; + + if ('body' in result) { + let bodyBuffer; + if (Buffer.isBuffer(result.body)) { + bodyBuffer = result.body; + } else if (typeof result.body === 'string') { + bodyBuffer = Buffer.from(result.body, encoding); + } else { + originalStream.destroy(); + throwAfterResponseTypeError(result.body); + } + if (bufferedBody === null) { + originalStream.destroy(); // original body replaced without being read + } + return extend(responseData, { stream: Readable.from(bodyBuffer), encoding, metadata }); + } + + if (bufferedBody !== null) { + // body was read by the action - original stream is consumed, serve the buffered copy + return extend(responseData, { stream: Readable.from(bufferedBody), encoding, metadata }); + } + + return extend(responseData, { encoding, metadata }); } -async function getRequest ({url, referer, options = {}, afterResponse = defaultResponseHandler}) { - const requestOptions = extend(options, {url}); +/* + * Resolves when response headers arrive with + * { url, statusCode, mimeType, encoding, metadata, stream } + * The body is NOT buffered - it is available as a Readable in stream. + * Returns null if the resource should be skipped (afterResponse returned null). + */ +async function getRequest ({ url, referer, options = {}, afterResponse }) { + const requestOptions = extend(options, { url }); + 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(requestOptions); + const { response, stream } = await requestWithRetry(requestOptions); logger.debug(`[request] received response for ${response.url}, statusCode ${response.statusCode}`); - const responseHandlerResult = transformResult(await afterResponse({response})); - if (!responseHandlerResult) { - return null; - } - return { + // 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}`)); + + 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 }; + + 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..b6a94aa9 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,31 @@ class Resource { this.text = text; } + setContentStream (stream) { + this.contentStream = stream; + } + + /** + * Returns resource content as a Readable stream. + * For buffered (html/css) resources a fresh stream is created on each call. + * For streamed resources the live response stream is returned - it can be consumed only once. + * Returns null when content was already cleared (see clearContent). + */ + getContentStream () { + if (this.contentStream) { + return this.contentStream; + } + if (typeof this.text === 'string') { + return Readable.from(Buffer.from(this.text, this.encoding)); + } + return null; + } + + clearContent () { + this.text = null; + this.contentStream = null; + } + getDepth () { return this.depth; } diff --git a/lib/scraper.js b/lib/scraper.js index ff2d810d..591f09c8 100644 --- a/lib/scraper.js +++ b/lib/scraper.js @@ -14,6 +14,7 @@ import { import * as utils from './utils/index.js'; const { extend, union, urlsEqual, getTypeByMime, getTypeByFilename, series } = utils; +import { readAll } from './utils/stream.js'; import NormalizedUrlMap from './utils/normalized-url-map.js'; const actionNames = [ @@ -122,15 +123,38 @@ 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 + // so each saveResource action can read it + resource.setText(await readAll(resource.contentStream, 'binary')); + resource.setEncoding('binary'); + 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}); + } 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}); + } + + async saveResource (resource) { + try { + await this.persistResource(resource); } catch (err) { logger.warn('failed to save resource ' + resource); await this.handleError(err, resource); @@ -141,54 +165,75 @@ 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); - } + if (!urlsEqual(responseData.url, url)) { // Url may be changed in redirects + logger.debug('url changed. old url = ' + url + ', new url = ' + responseData.url); - resource.setType(getTypeByMime(responseData.mimeType)); + 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}; + } - const { filename } = await self.runActions('generateFilename', { resource, responseData }); - resource.setFilename(filename); + resource.setUrl(responseData.url); + self.requestedResourcePromises.set(responseData.url, requestPromise); + } - // 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.setType(getTypeByMime(responseData.mimeType)); - if (responseData.metadata) { - resource.setMetadata(responseData.metadata); - } + const { stream, ...responseInfo } = responseData; + const { filename } = await self.runActions('generateFilename', { resource, responseData: responseInfo }); + resource.setFilename(filename); - resource.setEncoding(responseData.encoding); - resource.setText(responseData.body); + // 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); + } - self.loadResource(resource); // Add resource to list for future downloading, see Scraper.waitForLoad - return resource; + 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 + resource.setText(await readAll(stream, responseData.encoding)); + self.loadResource(resource); + } else { + // other types need no modification - stream them to storage immediately + resource.setContentStream(stream); + self.loadResource(resource); + await self.persistResource(resource); + } + return resource; + }); + }).then(function resolveRedirectDuplicate (result) { + if (result && result.redirectedTo) { + return self.requestedResourcePromises.get(result.redirectedTo); + } + return result; }).catch(function handleError (err) { logger.error('failed to request resource ' + resource); return self.handleError(err, resource); diff --git a/lib/utils/fs.js b/lib/utils/fs.js index 38618024..d03f97e0 100644 --- a/lib/utils/fs.js +++ b/lib/utils/fs.js @@ -1,5 +1,7 @@ import path from 'path'; import fs from 'fs/promises'; +import { createWriteStream } from 'fs'; +import { pipeline } from 'stream/promises'; async function outputFile (file, data, encoding) { const dir = path.dirname(file); @@ -8,6 +10,14 @@ async function outputFile (file, data, encoding) { return fs.writeFile(file, data, { encoding: encoding }); } +async function outputFileStream (file, stream) { + const dir = path.dirname(file); + await fs.mkdir(dir, { recursive: true}); + + return pipeline(stream, createWriteStream(file)); +} + export { - outputFile + outputFile, + outputFileStream }; diff --git a/lib/utils/stream.js b/lib/utils/stream.js new file mode 100644 index 00000000..07880483 --- /dev/null +++ b/lib/utils/stream.js @@ -0,0 +1,11 @@ +async function readAll (stream, encoding) { + const chunks = []; + for await (const chunk of stream) { + chunks.push(chunk); + } + return Buffer.concat(chunks).toString(encoding); +} + +export { + readAll +}; diff --git a/package.json b/package.json index 9652be77..30e612a8 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/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..ad4498c9 --- /dev/null +++ b/test/functional/streaming/streaming.test.js @@ -0,0 +1,243 @@ +import * as chai from 'chai'; +const should = chai.should(); +import '../../utils/assertions.js'; +import nock from 'nock'; +import fs from 'fs'; +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']); + }); + + // 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 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); + }); + }); + + 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..60540ee7 100644 --- a/test/unit/plugins.test.js +++ b/test/unit/plugins.test.js @@ -0,0 +1,108 @@ +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); + }); + }); + + 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); + }); + }); +}); diff --git a/test/unit/request-test.js b/test/unit/request-test.js index b2ee7256..071f023b 100644 --- a/test/unit/request-test.js +++ b/test/unit/request-test.js @@ -1,7 +1,8 @@ import * as chai from 'chai'; -chai.should(); +const should = chai.should(); import sinon from 'sinon'; import nock from 'nock'; +import { text as readStreamText, buffer as readStreamBuffer } from 'stream/consumers'; import request from '../../lib/request.js'; describe('request', () => { @@ -47,244 +48,195 @@ 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(''); - - 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({}); - }); - }); - - 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' - }); - - return request.get({url, afterResponse: handlerStub}).then((data) => { - data.body.should.eql('a'); - data.metadata.should.eql('b'); - data.encoding.should.eql('utf8'); - }); - }); - - 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' - }); - - return request.get({url, afterResponse: handlerStub}).then((data) => { - data.body.should.eql('a'); - (data.metadata === null).should.be.true; - data.encoding.should.eql('binary'); - }); - }); - - it('should transform string result', () => { - const url = 'http://example.com'; - nock(url).get('/').reply(200, 'TEST BODY'); - const handlerStub = sinon.stub().resolves('test body'); - - return request.get({url, afterResponse: handlerStub}).then((data) => { - data.body.should.eql('test body'); - (data.metadata === null).should.be.true; - }); - }); - - 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 return object with url, body, mimeType properties', () => { + 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}).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'); - }); + 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 mimeType = null if content-type header was not found in response', () => { - let url = 'http://www.google.com'; + 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}).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 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('get encoding', () => { - it('should return binary by default', () => { - const result = request.getEncoding(null); + 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'); - result.should.eql('binary'); + 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 binary when no content-type header supplies', () => { - const result = request.getEncoding({ - headers: {} - }); + 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'})); - result.should.eql('binary'); + try { + await request.get({url, options: {retry: {limit: 0}}}); + throw new Error('expected request.get to reject'); + } catch (err) { + err.code.should.eql('ECONNRESET'); + } }); - it('should return binary when content type header doesn\'t include utf-8', () => { - const result = request.getEncoding({ - headers: {} - }); + 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({}); - result.should.eql('binary'); - }); + await request.get({url, afterResponse: handlerStub}); + scope.isDone().should.eql(true); + handlerStub.calledOnce.should.eql(true); - it('should return binary when content type header doesn\'t include utf-8', () => { - const result = request.getEncoding({ - headers: { - 'content-type': 'text/html' - } + 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'); }); - result.should.eql('binary'); - }); + it('should provide body via getBody', 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 receivedBody; + const afterResponse = async ({response}) => { + receivedBody = await response.getBody(); + return {}; + }; - result.should.eql('utf8'); - }); + const data = await request.get({url, afterResponse}); + receivedBody.should.be.instanceOf(Buffer); + receivedBody.toString().should.eql('TEST BODY'); - it('should return utf8 response object includes it', () => { - const result = request.getEncoding({ - encoding: 'utf8' + // 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('utf8'); - }); -}); - -describe('transformResult', () => { - it('should throw with weird shaped response', () => { - try { - request.transformResult([1,2,3]); + it('should memoize getBody', async () => { + const url = 'http://example.com'; + nock(url).get('/').reply(200, 'TEST BODY'); - // 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'); - } - }); + const afterResponse = async ({response}) => { + const first = await response.getBody(); + const second = await response.getBody(); + first.should.be.equal(second); + return {}; + }; - it('should pass through error', () => { - try { - request.transformResult(new Error('Oh no')); + await request.get({url, afterResponse}); + }); - // We shouldn't get here. - true.should.eql(false); - } catch (e) { - e.should.be.instanceOf(Error); - e.message.should.eql('Oh no'); - } - }); + 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 throw with boolean input', () => { - try { - request.transformResult(true); + 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('Wrong response handler result. Expected string or object, but received boolean'); - } - }); + 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' + }); - 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.should.eql('b'); + data.encoding.should.eql('utf8'); + const body = await readStreamText(data.stream); + body.should.eql('a'); }); - 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 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 with empty body string', () => { - const result = request.transformResult({ - body: '', - encoding: 'utf8', + 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', ''); - 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 raw string input', () => { - const result = request.transformResult('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 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'); - it('should handle null input', () => { - const result = request.transformResult(null); - (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.match(/Wrong afterResponse result.*but received string/); + } + }); - it('should handle undefined input', () => { - const result = request.transformResult(undefined); - (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')); + + 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..d7875b48 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,51 @@ 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 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); + }); + }); + + describe('#clearContent', function () { + it('should clear text and content stream', function () { + const resource = new Resource('http://example.com'); + resource.setText('some text'); + resource.setContentStream(Readable.from('stream')); + + 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 () =>{ diff --git a/test/unit/utils/stream-test.js b/test/unit/utils/stream-test.js new file mode 100644 index 00000000..c2c84111 --- /dev/null +++ b/test/unit/utils/stream-test.js @@ -0,0 +1,34 @@ +import * as chai from 'chai'; +import { Readable } from 'stream'; +import { readAll } from '../../../lib/utils/stream.js'; +chai.should(); + +describe('utils/stream', function () { + describe('#readAll', function () { + it('should read whole stream to string with given encoding', async function () { + const stream = Readable.from([Buffer.from('hello '), Buffer.from('world')]); + const text = await readAll(stream, 'utf8'); + text.should.be.eql('hello world'); + }); + + it('should keep bytes intact with binary encoding', async function () { + const bytes = Buffer.from([0xff, 0x00, 0xab, 0x10]); + const text = await readAll(Readable.from(bytes), 'binary'); + Buffer.from(text, 'binary').should.be.eql(bytes); + }); + + it('should reject when stream emits error', async function () { + const stream = new Readable({ + read () { + this.destroy(new Error('read error')); + } + }); + try { + await readAll(stream, 'utf8'); + throw new Error('expected readAll to reject'); + } catch (err) { + err.message.should.be.eql('read error'); + } + }); + }); +});