Adds Streams API support for networking task of PDF.js project.

network.js file moved to main thread and `PDFNetworkStream` implemented
at worker thread, that is used to ask for data whenever worker needs.

src/core/worker.js

@@ -196,6 +196,115 @@ IPDFStreamRangeReader.prototype = {
 };
 }
 
+/** @implements {IPDFStream} */
+function PDFNetworkStream(data, msgHandler) {
+  this.data = data;
+  let source = data.source;
+  this._contentLength = source.length;
+  this.msgHandler = msgHandler;
+  this._fullRequestReader = null;
+  this._rangeRequestReaders = [];
+}
+
+PDFNetworkStream.prototype = {
+  getFullReader() {
+    assert(!this._fullRequestReader);
+    this._fullRequestReader = new PDFNetworkStreamReader(this);
+    return this._fullRequestReader;
+  },
+
+  getRangeReader(begin, end) {
+    let reader = new PDFNetworkStreamRangeReader(begin, end, this);
+    this._rangeRequestReaders.push(reader);
+    return reader;
+  },
+
+  cancelAllRequests(reason) {
+    if (this._fullRequestReader) {
+      this._fullRequestReader.cancel(reason);
+    }
+    let readers = this._rangeRequestReaders.slice(0);
+    readers.forEach(function (reader) {
+      reader.cancel(reason);
+    });
+  },
+};
+
+/** @implements {IPDFStreamReader} */
+function PDFNetworkStreamReader(networkStream) {
+  this._msgHandler = networkStream.msgHandler;
+  this._data = networkStream.data;
+  this.onProgress = null;
+
+  this._contentLength = null;
+  this._isRangeSupported = false;
+  this._isStreamingSupported = false;
+
+  let readableStream = this._msgHandler.sendWithStream('GetReader',
+                         this._data);
+
+  this._reader = readableStream.getReader();
+
+  this._headersReady = this._msgHandler.sendWithPromise('ReaderHeadersReady').
+    then((data) => {
+    this._isStreamingSupported = data.isStreamingSupported;
+    this._isRangeSupported = data.isRangeSupported;
+    this._contentLength = data.contentLength;
+  });
+}
+
+PDFNetworkStreamReader.prototype = {
+  get headersReady() {
+    return this._headersReady;
+  },
+
+  get contentLength() {
+    return this._contentLength;
+  },
+
+  get isStreamingSupported() {
+    return this._isStreamingSupported;
+  },
+
+  get isRangeSupported() {
+    return this._isRangeSupported;
+  },
+
+  read() {
+    return this._reader.read();
+  },
+
+  cancel(reason) {
+    this._reader.cancel(reason);
+  },
+};
+
+/** @implements {IPDFStreamRangeReader} */
+function PDFNetworkStreamRangeReader(begin, end, networkStream) {
+  this._msgHandler = networkStream.msgHandler;
+  this._data = networkStream.data;
+  this.onProgress = null;
+
+  let readableStream = this._msgHandler.sendWithStream('GetRangeReader',
+                        { data: this._data, begin, end, });
+
+  this._reader = readableStream.getReader();
+}
+
+PDFNetworkStreamRangeReader.prototype = {
+  get isStreamingSupported() {
+    return false;
+  },
+
+  read() {
+    return this._reader.read();
+  },
+
+  cancel(reason) {
+    this._reader.cancel(reason);
+  },
+};
+
 /** @implements {IPDFStream} */
 var PDFWorkerStream = (function PDFWorkerStreamClosure() {
   function PDFWorkerStream(params, msgHandler) {
@@ -399,17 +508,6 @@ var PDFWorkerStream = (function PDFWorkerStreamClosure() {
   return PDFWorkerStream;
 })();
 
-/** @type IPDFStream */
-var PDFNetworkStream;
-
-/**
- * Sets PDFNetworkStream class to be used as alternative PDF data transport.
- * @param {IPDFStream} cls - the PDF data transport.
- */
-function setPDFNetworkStreamClass(cls) {
-  PDFNetworkStream = cls;
-}
-
 var WorkerMessageHandler = {
   setup(handler, port) {
     var testMessageProcessed = false;
@@ -541,8 +639,7 @@ var WorkerMessageHandler = {
         if (source.chunkedViewerLoading) {
           pdfStream = new PDFWorkerStream(source, handler);
         } else {
-          assert(PDFNetworkStream, './network module is not loaded');
-          pdfStream = new PDFNetworkStream(data);
+          pdfStream = new PDFNetworkStream(data, handler);
         }
       } catch (ex) {
         pdfManagerCapability.reject(ex);
@@ -963,7 +1060,6 @@ if (typeof window === 'undefined' && !isNodeJS() &&
 }
 
 export {
-  setPDFNetworkStreamClass,
   WorkerTask,
   WorkerMessageHandler,
 };

src/display/api.js

@@ -28,6 +28,7 @@ import {
 import { FontFaceObject, FontLoader } from './font_loader';
 import { CanvasGraphics } from './canvas';
 import { Metadata } from './metadata';
+import { PDFNetworkStream } from './network';
 
 var DEFAULT_RANGE_CHUNK_SIZE = 65536; // 2^16 = 65536
 
@@ -1522,6 +1523,10 @@ var WorkerTransport = (function WorkerTransportClosure() {
     this.destroyCapability = null;
     this._passwordCapability = null;
 
+    this._PDFNetworkStream = null;
+    this._fullReader = null;
+    this._rangeReader = null;
+
     this.pageCache = [];
     this.pagePromises = [];
     this.downloadInfoCapability = createPromiseCapability();
@@ -1601,6 +1606,62 @@ var WorkerTransport = (function WorkerTransportClosure() {
           }, this);
       }
 
+      messageHandler.on('GetReader', function(data, sink) {
+        this._PDFNetworkStream = new PDFNetworkStream(data);
+        this._fullReader = this._PDFNetworkStream.getFullReader();
+
+        sink.onPull = () => {
+          this._fullReader.read().then(function(result) {
+            if (result.done) {
+              sink.close();
+              return;
+            }
+            sink.enqueue(result.value);
+          }).catch(function(reason) {
+            sink.error(reason);
+          });
+        };
+
+        sink.onCancel = (reason) => {
+          this._fullReader.cancel(reason);
+        };
+      }, this);
+
+      messageHandler.on('ReaderHeadersReady', function(data) {
+        let headersCapability = createPromiseCapability();
+        this._fullReader.headersReady.then(() => {
+          headersCapability.resolve({
+            isStreamingSupported: this._fullReader.isStreamingSupported,
+            isRangeSupported: this._fullReader.isRangeSupported,
+            contentLength: this._fullReader.contentLength,
+          });
+        });
+
+        return headersCapability.promise;
+      }, this);
+
+      messageHandler.on('GetRangeReader', function(data, sink) {
+        this._PDFNetworkStream = new PDFNetworkStream(data.data);
+        this._rangeReader =
+          this._PDFNetworkStream.getRangeReader(data.begin, data.end);
+
+        sink.onPull = () => {
+          this._rangeReader.read().then(function(result) {
+            if (result.done) {
+              sink.close();
+              return;
+            }
+            sink.enqueue(result.value);
+          }).catch(function(reason) {
+            sink.error(reason);
+          });
+        };
+
+        sink.onCancel = (reason) => {
+          this._rangeReader.cancel(reason);
+        };
+      }, this);
+
       messageHandler.on('GetDoc', function transportDoc(data) {
         var pdfInfo = data.pdfInfo;
         this.numPages = data.pdfInfo.numPages;

src/core/network.js → src/display/network.js

@@ -17,7 +17,6 @@ import {
   assert, createPromiseCapability, globalScope, isInt, MissingPDFException,
   UnexpectedResponseException
 } from '../shared/util';
-import { setPDFNetworkStreamClass } from './worker';
 
 if (typeof PDFJSDev !== 'undefined' && PDFJSDev.test('FIREFOX || MOZCENTRAL')) {
   throw new Error('Module "./network" shall not ' +
@@ -593,7 +592,6 @@ PDFNetworkStreamRangeRequestReader.prototype = {
   },
 };
 
-setPDFNetworkStreamClass(PDFNetworkStream);
 
 export {
   PDFNetworkStream,

src/pdf.worker.js

@@ -21,9 +21,4 @@ var pdfjsBuild = PDFJSDev.eval('BUNDLE_BUILD');
 
 var pdfjsCoreWorker = require('./core/worker.js');
 
-if (typeof PDFJSDev === 'undefined' ||
-    !PDFJSDev.test('FIREFOX || MOZCENTRAL')) {
-  require('./core/network.js');
-}
-
 exports.WorkerMessageHandler = pdfjsCoreWorker.WorkerMessageHandler;

src/worker_loader.js

@@ -28,7 +28,6 @@ importScripts('./shared/compatibility.js');
 importScripts('../node_modules/systemjs/dist/system.js');
 importScripts('../systemjs.config.js');
 
-Promise.all([SystemJS.import('pdfjs/core/network'),
-             SystemJS.import('pdfjs/core/worker')]).then(function () {
+Promise.resolve(SystemJS.import('pdfjs/core/worker')).then(function () {
   // Worker is loaded at this point.
 });

test/unit/network_spec.js

@@ -13,7 +13,7 @@
  * limitations under the License.
  */
 
-import { PDFNetworkStream } from '../../src/core/network';
+import { PDFNetworkStream } from '../../src/display/network';
 
 describe('network', function() {
   var pdf1 = new URL('../pdfs/tracemonkey.pdf', window.location).href;