feat: add XZ compression and decompression support

index.d.ts

@@ -53,6 +53,37 @@ export namespace gzip {
 
 }
 
+export namespace xz {
+
+  function compressFile(source: sourceType, dest: destType, opts?: any): Promise<void>
+
+  function uncompress(source: sourceType, dest: destType, opts?: any): Promise<void>
+
+  function decompress(source: sourceType, dest: destType, opts?: any): Promise<void>
+
+  export class FileStream extends ReadStream {
+
+    constructor(opts?: {
+      preset?: number,
+      threads?: number,
+      source?: sourceType
+    });
+
+  }
+
+  export class UncompressStream extends WriteStream {
+
+    constructor(opts?: {
+      source?: sourceType
+    });
+
+    on(event: string, listener: (...args: any[]) => void): this
+    on(event: 'error', listener: (err: Error) => void): this
+
+  }
+
+}
+
 export namespace tar {
 
   function compressFile(source: sourceType, dest: destType, opts?: any): Promise<void>

index.js

@@ -4,3 +4,4 @@ exports.zip = require('./lib/zip');
 exports.gzip = require('./lib/gzip');
 exports.tar = require('./lib/tar');
 exports.tgz = require('./lib/tgz');
+exports.xz = require('./lib/xz');

lib/xz/file_stream.js

@@ -0,0 +1,42 @@
+'use strict';
+
+const fs = require('fs');
+const lzma = require('lzma-native');
+const utils = require('../utils');
+const streamifier = require('streamifier');
+
+class XzFileStream extends lzma.Compressor {
+  constructor(opts) {
+    opts = opts || {};
+    const lzmaOpts = {
+      preset: opts.preset || 6,
+      threads: opts.threads || 0,
+    };
+    super(lzmaOpts);
+
+    const sourceType = utils.sourceType(opts.source);
+
+    if (sourceType === 'file') {
+      const stream = fs.createReadStream(opts.source, opts.fs);
+      stream.on('error', err => this.emit('error', err));
+      stream.pipe(this);
+      return;
+    }
+
+    if (sourceType === 'buffer') {
+      const stream = streamifier.createReadStream(opts.source, opts.streamifier);
+      stream.on('error', err => this.emit('error', err));
+      stream.pipe(this);
+      return;
+    }
+
+    if (sourceType === 'stream') {
+      opts.source.on('error', err => this.emit('error', err));
+      opts.source.pipe(this);
+    }
+
+    // else undefined: do nothing
+  }
+}
+
+module.exports = XzFileStream;

lib/xz/index.js

@@ -0,0 +1,11 @@
+'use strict';
+
+const utils = require('../utils');
+const XzFileStream = require('./file_stream');
+const XzUncompressStream = require('./uncompress_stream');
+
+exports.FileStream = XzFileStream;
+exports.UncompressStream = XzUncompressStream;
+exports.compressFile = utils.makeFileProcessFn(XzFileStream);
+exports.uncompress = utils.makeFileProcessFn(XzUncompressStream);
+exports.decompress = utils.makeFileProcessFn(XzUncompressStream);

lib/xz/uncompress_stream.js

@@ -0,0 +1,38 @@
+'use strict';
+
+const fs = require('fs');
+const lzma = require('lzma-native');
+const utils = require('../utils');
+const streamifier = require('streamifier');
+
+class XzUncompressStream extends lzma.Decompressor {
+  constructor(opts) {
+    opts = opts || {};
+    super();
+
+    const sourceType = utils.sourceType(opts.source);
+
+    if (sourceType === 'file') {
+      const stream = fs.createReadStream(opts.source, opts.fs);
+      stream.on('error', err => this.emit('error', err));
+      stream.pipe(this);
+      return;
+    }
+
+    if (sourceType === 'buffer') {
+      const stream = streamifier.createReadStream(opts.source, opts.streamifier);
+      stream.on('error', err => this.emit('error', err));
+      stream.pipe(this);
+      return;
+    }
+
+    if (sourceType === 'stream') {
+      opts.source.on('error', err => this.emit('error', err));
+      opts.source.pipe(this);
+    }
+
+    // else: waiting to be piped
+  }
+}
+
+module.exports = XzUncompressStream;

package.json

@@ -43,6 +43,7 @@
     "flushwritable": "^1.0.0",
     "get-ready": "^1.0.0",
     "iconv-lite": "^0.5.0",
+    "lzma-native": "^8.0.6",
     "streamifier": "^0.1.1",
     "tar-stream": "^1.5.2",
     "yazl": "^2.4.2"

test/xz/file_stream.test.js

@@ -0,0 +1,113 @@
+const fs = require('fs');
+const os = require('os');
+const path = require('path');
+const uuid = require('uuid');
+const { pipeline: pump } = require('stream');
+const compressing = require('../..');
+const assert = require('assert');
+
+describe('test/xz/file_stream.test.js', () => {
+  it('should be a transform stream', done => {
+    const sourceFile = path.join(__dirname, '..', 'fixtures', 'xx.log');
+    const sourceStream = fs.createReadStream(sourceFile);
+    const destFile = path.join(os.tmpdir(), uuid.v4() + '.log.xz');
+    // console.log('destFile', destFile);
+    const xzStream = new compressing.xz.FileStream();
+    const destStream = fs.createWriteStream(destFile);
+    pump(sourceStream, xzStream, destStream, err => {
+      assert(!err);
+      assert(fs.existsSync(destFile));
+      done();
+    });
+  });
+
+  it('should compress according to file path', done => {
+    const sourceFile = path.join(__dirname, '..', 'fixtures', 'xx.log');
+    const destFile = path.join(os.tmpdir(), uuid.v4() + '.log.xz');
+    // console.log('destFile', destFile);
+    const xzStream = new compressing.xz.FileStream({ source: sourceFile });
+    const destStream = fs.createWriteStream(destFile);
+    pump(xzStream, destStream, err => {
+      assert(!err);
+      assert(fs.existsSync(destFile));
+      done();
+    });
+  });
+
+  it('should compress file into Buffer', async () => {
+    const sourceFile = path.join(__dirname, '..', 'fixtures', 'xx.log');
+    const xzStream = new compressing.xz.FileStream({ source: sourceFile });
+    const xzChunks = [];
+    for await (const chunk of xzStream) {
+      xzChunks.push(chunk);
+    }
+
+    const destFile = path.join(os.tmpdir(), uuid.v4() + '.log.xz');
+    await fs.promises.writeFile(destFile, Buffer.concat(xzChunks));
+    // console.log(destFile);
+  });
+
+  it('should compress buffer', done => {
+    const sourceFile = path.join(__dirname, '..', 'fixtures', 'xx.log');
+    const sourceBuffer = fs.readFileSync(sourceFile);
+    const destFile = path.join(os.tmpdir(), uuid.v4() + '.log.xz');
+    // console.log('destFile', destFile);
+    const destStream = fs.createWriteStream(destFile);
+    const xzStream = new compressing.xz.FileStream({ source: sourceBuffer });
+    pump(xzStream, destStream, err => {
+      assert(!err);
+      assert(fs.existsSync(destFile));
+      done();
+    });
+
+  });
+
+  it('should compress stream', done => {
+    const sourceFile = path.join(__dirname, '..', 'fixtures', 'xx.log');
+    const sourceStream = fs.createReadStream(sourceFile);
+    const destFile = path.join(os.tmpdir(), uuid.v4() + '.log.xz');
+    // console.log('destFile', destFile);
+    const destStream = fs.createWriteStream(destFile);
+    const xzStream = new compressing.xz.FileStream({ source: sourceStream });
+    pump(xzStream, destStream, err => {
+      assert(!err);
+      assert(fs.existsSync(destFile));
+      done();
+    });
+  });
+
+  it('should compress with custom level', done => {
+    const sourceFile = path.join(__dirname, '..', 'fixtures', 'xx.log');
+    const destFile = path.join(os.tmpdir(), uuid.v4() + '.log.xz');
+    const xzStream = new compressing.xz.FileStream({
+      source: sourceFile,
+      level: 6,
+    });
+    const destStream = fs.createWriteStream(destFile);
+    pump(xzStream, destStream, err => {
+      assert(!err);
+      assert(fs.existsSync(destFile));
+      done();
+    });
+  });
+
+  it('should emit error if sourceFile does not exit', done => {
+    const sourceFile = 'file-not-exist';
+    const xzStream = new compressing.xz.FileStream({ source: sourceFile });
+    xzStream.on('error', err => {
+      assert(err);
+      done();
+    });
+  });
+
+  it('should emit error if sourceStream emit error', done => {
+    const sourceFile = 'file-not-exist';
+    const sourceStream = fs.createReadStream(sourceFile);
+    const xzStream = new compressing.xz.FileStream({ source: sourceStream });
+    xzStream.on('error', err => {
+      assert(err && err.code === 'ENOENT');
+      done();
+    });
+  });
+
+});