Adds http support to node_stream logic

This commit is contained in:
Mukul Mishra 2017-07-30 20:28:32 +05:30
parent ed78b23ff2
commit 18ede8c65d
3 changed files with 309 additions and 115 deletions

View file

@ -15,26 +15,39 @@
/* globals __non_webpack_require__ */
let fs = __non_webpack_require__('fs');
let utils = require('../shared/util');
let http = __non_webpack_require__('http');
let https = __non_webpack_require__('https');
let url = __non_webpack_require__('url');
let assert = utils.assert;
let createPromiseCapability = utils.createPromiseCapability;
import { assert, createPromiseCapability } from '../shared/util';
import { validateRangeRequestCapabilities } from './network_utils';
class PDFNodeStream {
constructor(args) {
this._path = args.path;
constructor(options) {
this.options = options;
this.source = options.source;
this.url = url.parse(this.source.url);
this.isHttp = this.url.protocol === 'http:' ||
this.url.protocol === 'https:';
this.isFsUrl = !this.url.host;
this.httpHeaders = (this.isHttp && this.source.httpHeaders) || {};
this._fullRequest = null;
this._rangeRequestReaders = [];
}
getFullReader() {
assert(!this._fullRequest);
this._fullRequest = new PDFNodeStreamFullReader(this._path);
this._fullRequest = this.isFsUrl ?
new PDFNodeStreamFsFullReader(this) :
new PDFNodeStreamFullReader(this);
return this._fullRequest;
}
getRangeReader(begin, end) {
let rangeReader = new PDFNodeStreamRangeReader(this._path, begin, end);
getRangeReader(start, end) {
let rangeReader = this.isFsUrl ?
new PDFNodeStreamFsRangeReader(this, start, end) :
new PDFNodeStreamRangeReader(this, start, end);
this._rangeRequestReaders.push(rangeReader);
return rangeReader;
}
@ -51,44 +64,19 @@ class PDFNodeStream {
}
}
class PDFNodeStreamFullReader {
constructor(path) {
this._path = path;
class BaseFullReader {
constructor(stream) {
this._url = stream.url;
this._done = false;
this._errored = false;
this._reason = null;
this.onProgress = null;
this._length = null;
this._length = stream.source.length;
this._loaded = 0;
this._fullRequest = null;
this._readCapability = createPromiseCapability();
this._headersCapability = createPromiseCapability();
this._fullRequest = fs.createReadStream(path);
fs.lstat(this._path, (error, stat) => {
if (error) {
this._errored = true;
this._reason = error;
this._headersCapability.reject(error);
return;
}
this._length = stat.size;
this._headersCapability.resolve();
});
this._fullRequest.on('readable', () => {
this._readCapability.resolve();
});
this._fullRequest.on('end', () => {
this._done = true;
this._readCapability.resolve();
});
this._fullRequest.on('error', (reason) => {
this._errored = true;
this._reason = reason;
this._readCapability.resolve();
});
}
get headersReady() {
@ -100,11 +88,11 @@ class PDFNodeStreamFullReader {
}
get isRangeSupported() {
return true;
return this._isRangeSupported;
}
get isStreamingSupported() {
return true;
return this._isStreamingSupported;
}
read() {
@ -138,45 +126,20 @@ class PDFNodeStreamFullReader {
}
}
class PDFNodeStreamRangeReader {
constructor(path, start, end) {
this._path = path;
class BaseRangeReader {
constructor(stream) {
this._url = stream.url;
this._done = false;
this._errored = false;
this._reason = null;
this.onProgress = null;
this._length = null;
this._length = stream.source.length;
this._loaded = 0;
this._readCapability = createPromiseCapability();
this._rangeRequest = fs.createReadStream(path, { start, end, });
fs.lstat(this._path, (error, stat) => {
if (error) {
this._errored = true;
this._reason = error;
return;
}
this._length = stat.size;
});
this._rangeRequest.on('readable', () => {
this._readCapability.resolve();
});
this._rangeRequest.on('end', () => {
this._done = true;
this._readCapability.resolve();
});
this._rangeRequest.on('error', (reason) => {
this._errored = true;
this._reason = reason;
this._readCapability.resolve();
});
}
get isStreamingSupported() {
return true;
return false;
}
read() {
@ -188,7 +151,7 @@ class PDFNodeStreamRangeReader {
return Promise.reject(this._reason);
}
let chunk = this._rangeRequest.read();
let chunk = this._read();
if (chunk === null) {
this._readCapability = createPromiseCapability();
return this.read();
@ -203,11 +166,209 @@ class PDFNodeStreamRangeReader {
return Promise.resolve({ value: chunk, done: false, });
});
}
}
cancel(reason) {
this._fullRequest.cancel(reason);
this._fullRequest.destroy(reason);
class PDFNodeStreamFullReader extends BaseFullReader {
constructor(stream) {
super(stream);
this._disableRange = stream.options.disableRange || false;
this._rangeChunkSize = stream.source.rangeChunkSize;
if (!this._rangeChunkSize && !this._disableRange) {
this._disableRange = true;
}
this._isStreamingSupported = !stream.source.disableStream;
this._isRangeSupported = false;
let options = {
host: this._url.host,
path: this._url.path,
method: 'GET',
headers: stream.httpHeaders,
};
let handleResponse = (response) => {
this._headersCapability.resolve();
this._fullRequest = response;
response.on('readable', () => {
this._readCapability.resolve();
});
response.on('end', () => {
// Destroy response to minimize resource usage.
response.destroy();
this._done = true;
this._readCapability.resolve();
});
response.on('error', (reason) => {
this._errored = true;
this._reason = reason;
this._readCapability.resolve();
});
};
this._request = this._url.protocol === 'http:' ?
http.request(options, handleResponse) :
https.request(options, handleResponse);
this._request.on('error', (reason) => {
this._errored = true;
this._reason = reason;
this._headersCapability.reject(reason);
});
this._request.end();
this._headersCapability.promise.then(() => {
let { allowRangeRequests, suggestedLength, } =
validateRangeRequestCapabilities({
getResponseHeader: this.getResponseHeader.bind(this),
isHttp: stream.isHttp,
rangeChunkSize: this._rangeChunkSize,
disableRange: this._disableRange,
});
if (allowRangeRequests) {
this._isRangeSupported = true;
}
this._length = suggestedLength;
});
}
getReasponseHeader(name) {
return this._fullRequest.headers[name];
}
}
exports.PDFNodeStream = PDFNodeStream;
class PDFNodeStreamRangeReader extends BaseRangeReader {
constructor(stream, start, end) {
super(stream);
this._rangeRequest = null;
this._read = null;
let rangeStr = start + '-' + (end - 1);
stream.httpHeaders['Range'] = 'bytes=' + rangeStr;
let options = {
host: this._url.host,
path: this._url.path,
method: 'GET',
headers: stream.httpHeaders,
};
let handleResponse = (response) => {
this._rangeRequest = response;
this._read = this._rangeRequest.read;
response.on('readable', () => {
this._readCapability.resolve();
});
response.on('end', () => {
response.destroy();
this._done = true;
this._readCapability.resolve();
});
response.on('error', (reason) => {
this._errored = true;
this._reason = reason;
this._readCapability.resolve();
});
};
this._request = this._url.protocol === 'http:' ?
http.request(options, handleResponse) :
https.request(options, handleResponse);
this._request.on('error', (reason) => {
this._errored = true;
this._reason = reason;
});
this._request.end();
}
cancel(reason) {
this._rangeRequest.close(reason);
this._rangeRequest.destroy(reason);
}
}
class PDFNodeStreamFsFullReader extends BaseFullReader {
constructor(stream) {
super(stream);
this._isRangeSupported = true;
this._isStreamingSupported = true;
this._fullRequest = fs.createReadStream(this._url.path);
fs.lstat(this._url.path, (error, stat) => {
if (error) {
this._errored = true;
this._reason = error;
this._headersCapability.reject(error);
return;
}
this._length = stat.size;
this._headersCapability.resolve();
});
this._fullRequest.on('readable', () => {
this._readCapability.resolve();
});
this._fullRequest.on('end', () => {
this._fullRequest.destroy();
this._done = true;
this._readCapability.resolve();
});
this._fullRequest.on('error', (reason) => {
this._errored = true;
this._reason = reason;
this._readCapability.resolve();
});
}
}
class PDFNodeStreamFsRangeReader extends BaseRangeReader {
constructor(stream, start, end) {
super(stream);
this._rangeRequest = fs.createReadStream(this._url.path, { start, end, });
fs.lstat(this._url.path, (error, stat) => {
if (error) {
this._errored = true;
this._reason = error;
return;
}
this._length = stat.size;
});
this._read = this._rangeRequest.read;
this._rangeRequest.on('readable', () => {
this._readCapability.resolve();
});
this._rangeRequest.on('end', () => {
this._rangeRequest.destroy();
this._done = true;
this._readCapability.resolve();
});
this._rangeRequest.on('error', (reason) => {
this._errored = true;
this._reason = reason;
this._readCapability.resolve();
});
}
cancel(reason) {
this._rangeRequest.close(reason);
this._rangeRequest.destroy(reason);
}
}
export {
PDFNodeStream,
};