WARC Serialize

index.js

@@ -1,5 +1,7 @@
 require = require("esm")(module);
-module.exports = require("./main.js");
+module.exports = {
+  ...require("./main.js")
+};
 
 // ensure global Headers object is set for node
 /* istanbul ignore next */

main.js

@@ -5,6 +5,8 @@ export { StatusAndHeadersParser, StatusAndHeaders } from './src/statusandheaders
 
 export { WARCParser } from './src/warcparser';
 
+export { WARCSerializer } from './src/warcserializer';
+
 export { WARCRecord } from './src/warcrecord';
 
 export { Indexer, CDXIndexer } from './src/indexer';

package.json

@@ -1,6 +1,6 @@
 {
   "name": "warcio",
-  "version": "1.1.3",
+  "version": "1.2.0",
   "main": "index.js",
   "module": "main.js",
   "license": "Apache-2.0",
@@ -31,9 +31,12 @@
     "cli.js"
   ],
   "dependencies": {
+    "@peculiar/webcrypto": "^1.1.1",
     "esm": "^3.2.25",
+    "hi-base32": "^0.5.0",
     "node-fetch": "^2.6.0",
     "pako": "^1.0.11",
+    "uuid-random": "^1.3.0",
     "yargs": "^15.3.1"
   },
   "scripts": {

rollup.config.js

@@ -15,6 +15,6 @@ export default {
       format: 'esm',
     }
   ],
-  plugins: [commonjs(), resolve()]
+  plugins: [commonjs(), resolve({ browser: true })]
 };
 

src/indexer.js

@@ -11,8 +11,12 @@ class BaseIndexer
     this.out = out;
   }
 
+  serialize(result) {
+    return JSON.stringify(result) + "\n";
+  }
+
   write(result) {
-    this.out.write(JSON.stringify(result) + "\n");
+    this.out.write(this.serialize(result));
   }
 
   async run(files) {
@@ -39,7 +43,6 @@ class BaseIndexer
         }
       }
     }
-
   }
 
   indexRecord(record, parser, filename) {
@@ -132,11 +135,11 @@ class CDXIndexer extends Indexer
 
     switch (opts.format) {
       case "cdxj":
-        this.write = this.writeCDXJ;
+        this.serialize = this.serializeCDXJ;
         break;
 
       case "cdx":
-        this.write = this.writeCDX11;
+        this.serialize = this.serializeCDX11;
         break;
 
       case "json":
@@ -159,22 +162,22 @@ class CDXIndexer extends Indexer
     return true;
   }
 
-  writeCDXJ(result) {
+  serializeCDXJ(result) {
     const { urlkey, timestamp } = result;
     delete result.urlkey;
     delete result.timestamp;
 
-    this.out.write(`${urlkey} ${timestamp} ${JSON.stringify(result)}\n`);
+    return `${urlkey} ${timestamp} ${JSON.stringify(result)}\n`;
   }
 
-  writeCDX11(result) {
+  serializeCDX11(result) {
     const value = [];
 
     for (const field of DEFAULT_LEGACY_CDX_FIELDS) {
       value.push(result[field] != undefined ? result[field] : "-");
     }
 
-    this.out.write(value.join(" ") + "\n");
+    return value.join(" ") + "\n";
   }
 
   getField(field, record) {

src/readers.js

@@ -45,6 +45,20 @@ class BaseAsyncIterReader
     return [chunk.slice(0, inx), chunk.slice(inx)];
   }
 
+  static async readFully(iter) {
+    const chunks = [];
+    let size = 0;
+
+    for await (const chunk of iter) {
+      chunks.push(chunk);
+      size += chunk.byteLength;
+    }
+
+    return BaseAsyncIterReader.concatChunks(chunks, size);
+  }
+
+
+
   getReadableStream() {
     const streamIter = this[Symbol.asyncIterator]();
 
@@ -62,16 +76,8 @@ class BaseAsyncIterReader
     });
   }
 
-  async readFully() {
-    const chunks = [];
-    let size = 0;
-
-    for await (const chunk of this) {
-      chunks.push(chunk);
-      size += chunk.byteLength;
-    }
-
-    return BaseAsyncIterReader.concatChunks(chunks, size);
+  readFully() {
+    return BaseAsyncIterReader.readFully(this);
   }
 
   async readline(maxLength = 0) {
@@ -110,10 +116,10 @@ class AsyncIterReader extends BaseAsyncIterReader {
       }
     }
 
-    this._sourceIter = streamOrIter[Symbol.asyncIterator]();
-
     if (dechunk) {
-      this._sourceIter = this.dechunk(this._sourceIter);
+      this._sourceIter = this.dechunk(streamOrIter);
+    } else {
+      this._sourceIter = streamOrIter[Symbol.asyncIterator]();
     }
 
     this.lastValue = null;
@@ -134,7 +140,7 @@ class AsyncIterReader extends BaseAsyncIterReader {
   }
 
   async* dechunk(source) {
-    const reader = new AsyncIterReader(source, null);
+    const reader = (source instanceof AsyncIterReader) ? source : new AsyncIterReader(source, null);
 
     let size = -1;
     let first = true;
@@ -180,8 +186,12 @@ class AsyncIterReader extends BaseAsyncIterReader {
         break;
 
       } else {
-        yield chunk;
         first = false;
+        if (!chunk || size === 0) {
+          return;
+        } else {
+          yield chunk;
+        }
       }
     }
 

src/statusandheaders.js

@@ -1,3 +1,5 @@
+const CRLF = new Uint8Array([13, 10]);
+const CRLFCRLF = new Uint8Array([13, 10, 13, 10]);
 
 
 // ===========================================================================
@@ -17,6 +19,14 @@ class StatusAndHeaders {
     return buff.join('\r\n') + '\r\n';
   }
 
+  async* iterSerialize(encoder) {
+    yield encoder.encode(this.statusline);
+    yield CRLF;
+    for (const [name, value] of this.headers) {
+      yield encoder.encode(`${name}: ${value}\r\n`);
+    }
+  }
+
   _parseResponseStatusLine() {
     const parts = splitRemainder(this.statusline, " ", 2);
     this._protocol = parts[0];
@@ -135,4 +145,4 @@ function splitRemainder(str, sep, limit) {
 
 
 // ===========================================================================
-export { StatusAndHeaders, StatusAndHeadersParser };
+export { StatusAndHeaders, StatusAndHeadersParser, CRLF, CRLFCRLF };

src/warcparser.js

@@ -1,6 +1,6 @@
-import { StatusAndHeadersParser } from './statusandheaders';
+import { StatusAndHeaders, StatusAndHeadersParser } from './statusandheaders';
 import { WARCRecord } from './warcrecord';
-import { AsyncIterReader } from './readers';
+import { AsyncIterReader, LimitReader } from './readers';
 
 
 // ===========================================================================
@@ -11,7 +11,7 @@ class WARCParser
   }
 
   static iterRecords(source, options) {
-    return new WARCParser(source, options);
+    return new WARCParser(source, options)[Symbol.asyncIterator]();
   }
 
   constructor(source, {keepHeadersCase = false, parseHttp = true} = {}) {
@@ -40,6 +40,10 @@ class WARCParser
     }
   }
 
+  _initRecordReader(warcHeaders) {
+    return new LimitReader(this._reader, Number(warcHeaders.headers.get("Content-Length") || 0));
+  }
+
   async parse() {
     await this.readToNextRecord();
 
@@ -55,7 +59,7 @@ class WARCParser
 
     this._warcHeadersLength = this._reader.getReadOffset();
 
-    const record = new WARCRecord({warcHeaders, reader: this._reader});
+    const record = new WARCRecord({warcHeaders, reader: this._initRecordReader(warcHeaders)});
 
     this._atRecordBoundary = false;
     this._record = record;
@@ -96,9 +100,14 @@ class WARCParser
     this._record = null;
   }
 
-  async _addHttpHeaders(record, headersParser, reader) {
-    const httpHeaders = await headersParser.parse(reader, {headersClass: this._headersClass});
-    record._addHttpHeaders(httpHeaders, reader.getReadOffset() - this._warcHeadersLength);
+  async _addHttpHeaders(record, headersParser) {
+    const httpHeaders = await headersParser.parse(this._reader, {headersClass: this._headersClass});
+    record.httpHeaders = httpHeaders;
+
+    const headersLen = this._reader.getReadOffset() - this._warcHeadersLength;
+    if (record.reader.setLimitSkip) {
+      record.reader.setLimitSkip(record.warcContentLength - headersLen);
+    }
   }
 }