Merge pull request #78 from neon-bindings/kv/gzip-stream

feat: Add streaming gzip compression example

Author: kjvalencik

Cargo.lock

@@ -1,6 +1,10 @@
 [workspace]
 members = [
 	"examples/async-sqlite",
-	"examples/hello-world",
 	"examples/cpu-count",
+	"examples/gzip-stream",
+	"examples/hello-world",
 ]
+
+[profile.release]
+lto = true

Cargo.toml

@@ -16,10 +16,12 @@ All examples are for [`napi-backend`][napi-migration]. For examples using `legac
 | ------------------------------ | ------------------------------------------ |
 | [`async-sqlite`][async-sqlite] | Async interface to a SQLite database       |
 | [`cpu-count`][cpu-count]       | Return the number of CPUs                  |
+| [`gzip-stream`][gzip-stream]   | Asynchronously compress a stream of data   |
 | [`hello-world`][hello-world]   | Return a JS String with a greeting message |
 
 [async-sqlite]: examples/async-sqlite
 [cpu-count]: examples/cpu-count
+[gzip-stream]: examples/gzip-stream
 [hello-world]: examples/hello-world
 
 ## Contributing

README.md

@@ -0,0 +1,18 @@
+[package]
+name = "gzip-stream"
+version = "0.1.0"
+description = "Neon Stream Example"
+license = "MIT"
+edition = "2018"
+exclude = ["index.node"]
+
+[lib]
+crate-type = ["cdylib"]
+
+[dependencies]
+flate2 = "1"
+
+[dependencies.neon]
+version = "0.10.0-alpha.2"
+default-features = false
+features = ["napi-6", "promise-api", "task-api"]

examples/gzip-stream/Cargo.toml

@@ -0,0 +1,106 @@
+# Async gzip compress
+
+The gzip compression example demonstrates building a ["through" stream](https://nodejs.dev/learn/nodejs-streams) with Neon. The stream is both readable and writeable and CPU intensive processing occurs on the Node worker thread pool.
+
+## Design
+
+A small amount of JavaScript glue code can often simplify Neon modules. In this case, Node.js already provides a [`Transform`](https://nodejs.org/api/stream.html#stream_duplex_and_transform_streams) stream class. The `Transform` stream provides important features that could be complex to implement:
+
+* Writeable
+* Readable
+* Backpressure
+* Asynchronous
+
+Two methods must be implemented.
+
+### `transform(chunk, encoding, callback)`
+
+The `transform` method accepts a chunk of data and its encoding, as well a callback. The callback should be called when processing the `chunk` has completed.
+
+### `flush(callback)`
+
+The `flush` method allows any internally buffered data to be processed before completion. The `callback` is identical to the callback in `transform`.
+
+## Glue
+
+```js
+function compress() {
+    const compressor = compressNew();
+
+    return new Transform({
+        transform(chunk, encoding, callback) {
+            compressChunk(compressor, encoding, chunk)
+                .then(data => callback(null, data))
+                .catch(callback);
+        },
+
+        flush(callback) {
+            compressFinish(compressor)
+                .then(data => callback(null, data))
+                .catch(callback);
+        }
+    });
+}
+```
+
+The glue code exports a single function `compress` that creates a `Transform` stream delegating the implementation to Neon functions. Since these functions return promises, they are adapted to the `callback` style continuation that `Transform` expects.
+
+## Neon
+
+The Neon module exports three functions:
+
+* [`compressNew`](#compressnew)
+* [`compressChunk`](#compresschunkcompressstream-chunk-encoding-callback)
+* [`compressFinish`](#compressfinishcompressstream-callback)
+
+### `compressNew()`
+
+```rust
+fn compress_new(mut cx: FunctionContext) -> JsResult<JsBox<CompressStream>> {
+    let stream = CompressStream::new(Compression::best());
+
+    Ok(cx.boxed(stream))
+}
+```
+
+`compressNew` creates an instance of the stateful Rust struct, `CompressStream`, and returns it wrapped in a [`JsBox`](https://docs.rs/neon/latest/neon/types/struct.JsBox.html). Each of the other two methods expects `CompressStream` as the first argument. This pattern is similar to using [`Function.prototype.call`](https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/Function/call) on a class method to manually bind `this`.
+
+### `compressChunk(compressStream, chunk, encoding, callback)`
+
+```rust
+fn compress_chunk(mut cx: FunctionContext) -> JsResult<JsPromise> {
+    let stream = (&**cx.argument::<JsBox<CompressStream>>(0)?).clone();
+    let chunk = cx.argument::<JsTypedArray<u8>>(2)?
+        .as_slice(&cx)
+        .to_vec();
+
+    let promise = cx
+        .task(move || stream.write(chunk))
+        .promise(CompressStream::and_buffer);
+
+    Ok(promise)
+}
+```
+
+`compressChunk` accepts the instance of the `CompressStream` struct and the other arguments to the [`transform`](#transformchunk-encoding-callback) function. The chunk is cloned to a `Vec<u8>` and passed to a task to execute on the Node worker pool. The asynchronous task compresses the data and passes the compressed data to the `.promise(|cx, result| { ... })` callback. The callback to `promise` is executed on the JavaScript main thread and converts the compressed `Vec<u8>` to a `JsBuffer` and resolves the promise.
+
+`CompressChunk::and_buffer` is used to create a `Buffer`. `ArrayBuffer` cannot be used because stream chunks are required to be an instance of `Uint8Array`. `Buffer` is a subclass of `Uint8Array`.
+
+### `compressFinish(compressStream, callback)`
+
+fn compress_finish(mut cx: FunctionContext) -> JsResult<JsPromise> {
+let stream = (&**cx.argument::<JsBox<CompressStream>>(0)?).clone();
+
+```rust
+fn compress_finish(mut cx: FunctionContext) -> JsResult<JsPromise> {
+    let stream = (&**cx.argument::<JsBox<CompressStream>>(0)?).clone();
+
+    let promise = cx
+        .task(move || stream.finish())
+        .promise(CompressStream::and_buffer);
+
+    Ok(promise)
+}
+```
+
+`compressFinish` works very similar to [`compressChunkl`](#compresschunkcompressstream-chunk-encoding-callback), except it is provided the arguments to [`flush`](#flushcallback) which does not include any data. Instead, the remaining buffered data is compressed, a CRC is calculated, and the compressed gzip data is completed.

examples/gzip-stream/README.md

@@ -0,0 +1,23 @@
+#!/usr/bin/env node
+
+"use strict";
+
+const { pipeline } = require("stream");
+const { Gzip, constants: { Z_BEST_COMPRESSION } } = require("zlib");
+
+const neonCompress = require("..");
+const builtInCompress = () => new Gzip({ level: Z_BEST_COMPRESSION });
+
+const compress = process.argv[2] === "built-in" ? builtInCompress : neonCompress;
+
+pipeline(
+    process.stdin,
+    compress(),
+    process.stdout,
+    (err) => {
+        if (err) {
+            console.error(err);
+            process.exitCode = -1;
+        }
+    }
+);

examples/gzip-stream/bin/compress.js

@@ -0,0 +1,31 @@
+"use strict";
+
+// Transform stream reduces the boilerplate of a stream that reads bytes from a
+// source and produces new bytes for a destination sink.
+const { Transform } = require("stream");
+
+const { compressNew, compressChunk, compressFinish } = require("./index.node");
+
+// Creates a gzip compression transform stream, implemented asynchronously in Rust
+function compress() {
+    // Create a native streaming gzip compressing with Neon
+    const compressor = compressNew();
+
+    return new Transform({
+        // Compress a chunk of data by delegating to `compressChunk`
+        transform(chunk, encoding, callback) {
+            compressChunk(compressor, encoding, chunk)
+                .then(data => callback(null, data))
+                .catch(callback);
+        },
+
+        // Complete the compression by delegating to `compressFinish`
+        flush(callback) {
+            compressFinish(compressor)
+                .then(data => callback(null, data))
+                .catch(callback);
+        }
+    });
+}
+
+module.exports = compress;