WIP - use streams

Author: s0ph1e

lib/request.js

@@ -39,7 +39,19 @@ async function getRequest ({url, referer, options = {}, afterResponse = defaultR
 
 	logger.debug(`[request] sending request for url ${url}, referer ${referer}`);
 
-	const response = await got(requestOptions);
+	const responseStream = got.stream(requestOptions);
+
+	responseStream.on('data', (chunk) => {
+		console.log(`Received ${chunk.length} bytes of data.`);
+		//console.log(chunk.toString());
+	});
+
+	responseStream.on('response', (response) => {
+		console.log(`Received response ${JSON.stringify(response.headers)} `);
+	});
+
+	return responseStream;
+
 	logger.debug(`[request] received response for ${response.url}, statusCode ${response.statusCode}`);
 	const responseHandlerResult = transformResult(await afterResponse({response}));
 
@@ -55,5 +67,6 @@ async function getRequest ({url, referer, options = {}, afterResponse = defaultR
 }
 
 export default {
-	get: getRequest
+	get: getRequest,
+	getMimeType: (headers) => getMimeType(headers['content-type'])
 };

lib/scraper.js

@@ -1,10 +1,14 @@
+import stream, { Transform, Writable } from 'stream';
+import fs from 'fs';
+import {promisify} from 'util';
 import PromiseQueue from 'p-queue';
 import logger from './logger.js';
 import defaults from './config/defaults.js';
 import recursiveSources from './config/recursive-sources.js';
 import Resource from './resource.js';
 import request from './request.js';
 import ResourceHandler from './resource-handler/index.js';
+
 import {
 	SaveResourceToFileSystemPlugin,
 	GenerateFilenameBySiteStructurePlugin,
@@ -36,6 +40,44 @@ const filenameGeneratorPlugins = {
 	bySiteStructure: GenerateFilenameBySiteStructurePlugin
 };
 
+class AfterResponseTransformStream extends Transform {
+	constructor (options) {
+		super(options);
+	}
+
+	_transform (chunk, encoding, callback) {
+		//console.log(chunk.toString().toUpperCase());
+		callback(null, chunk.toString().toUpperCase());
+	}
+}
+
+class HandleChildrenResourcesStream extends Transform {
+	constructor(resource) {
+		super();
+		this.resouce = resource;
+	}
+
+	_transform(chunk, encoding, callback) {
+
+	}
+}
+
+class SaveResourceWriteStream extends Writable {
+	constructor (resource, dir) {
+		super();
+		this.stream = null;
+		this.resource = resource;
+		this.directory = dir;
+	}
+
+	_write (chunk, encoding, callback) {
+		if (!this.stream) {
+			this.stream = fs.createWriteStream(this.directory + '/' + this.resource.getFilename());
+		}
+		this.stream._write(chunk, encoding, callback);
+	}
+}
+
 class Scraper {
 	constructor (options) {
 		this.normalizeOptions(options);
@@ -49,6 +91,7 @@ class Scraper {
 		});
 		this.resources = this.options.urls.map(({url, filename}) => new Resource(url, filename));
 
+		this.requestedResourcesStream = new NormalizedUrlMap(); // Map url -> stream
 		this.requestedResourcePromises = new NormalizedUrlMap(); // Map url -> request promise
 		this.loadedResources = new NormalizedUrlMap(); // Map url -> resource
 		this.requestQueue = new PromiseQueue({concurrency: this.options.requestConcurrency});
@@ -137,10 +180,60 @@ class Scraper {
 		}
 	}
 
-	createNewRequest (resource) {
+	async getRequestStream (resource) {
+		const url = resource.getUrl();
+		const referer = resource.parent ? resource.parent.getUrl() : null;
+
+		const {requestOptions} = await this.runActions('beforeRequest', {resource, requestOptions: this.options.request});
+		return request.get({
+			url,
+			referer,
+			options: requestOptions,
+			//afterResponse: this.actions.afterResponse.length ? this.runActions.bind(this, 'afterResponse') : undefined
+		});
+	}
+
+	getSaveStream (resource) {
+
+	}
+
+	async createNewRequest (resource) {
 		const self = this;
 		const url = resource.getUrl();
 
+		// read stream stream
+		const requestStream = await this.getRequestStream(resource);
+		requestStream.on('response', async (response) => {
+			const mimeType = request.getMimeType(response.headers);
+			resource.setType(getTypeByMime(mimeType));
+
+			const { filename } = await self.runActions('generateFilename', { resource });
+			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));
+			}
+		});
+
+		// transformers
+		const afterResponseTransformStream = new AfterResponseTransformStream();
+
+		// write stream
+		fs.mkdirSync(this.options.directory);
+		const saveResourceStream = new SaveResourceWriteStream(resource, this.options.directory);
+
+		const pipeline = promisify(stream.pipeline);
+
+		await pipeline(
+			requestStream,
+			afterResponseTransformStream,
+			// handleChildResourcesStream,
+			saveResourceStream
+		);
+
+		//return this.requestQueue.add(requestStream);
+
 		const requestPromise = Promise.resolve()
 			.then(async () => {
 				const referer = resource.parent ? resource.parent.getUrl() : null;