feat: add framed Snappy streaming APIs (#351)

* feat: add framed Snappy streaming APIs

Add Compressor/Decompressor classes, Web Streams, and Node Duplex
factories mirroring @napi-rs/lzma. Streaming uses the Snappy frame
format (distinct from the existing raw one-shot compress/uncompress).

* fix: satisfy clippy new_without_default for Compressor

Implement Default for Compressor and apply prettier on streaming sources.

Author: Brooooooklyn

Cargo.toml

@@ -8,10 +8,18 @@ version = "0.1.0"
 crate-type = ["cdylib"]
 
 [dependencies]
-napi        = { version = "3.0.0", features = ["napi5", "serde-json"] }
-napi-derive = { version = "3.0.0" }
+napi        = { version = "3", features = ["napi5", "serde-json"] }
+napi-derive = "3"
 snap        = "1"
 
+# Web Streams transforms (`src/stream_web.rs`) need tokio via napi's `web_stream`
+# feature. Enable it ONLY for non-wasm targets so the wasm build keeps plain
+# `napi` (no tokio) and drops the streaming module. `web_stream` only pulls
+# `napi4`, but napi's ReadableStream finalizer calls `napi_add_finalizer`
+# (Node-API 5), so `napi5` is required alongside it.
+[target.'cfg(not(target_family = "wasm"))'.dependencies]
+napi = { version = "3", features = ["web_stream", "napi5"] }
+
 [target.'cfg(all(not(target_os = "linux"), not(target_family = "wasm")))'.dependencies]
 mimalloc-safe = { version = "0.1", features = ["skip_collect_on_exit"] }
 

README.md

@@ -45,25 +45,25 @@ yarn add snappy
 
 ### Targets
 
-| Rust triple                     | Platform             | CI                              |
-| ------------------------------- | -------------------- | ------------------------------- |
-| `x86_64-pc-windows-msvc`        | Windows x64          | tested — node 22, 24            |
-| `aarch64-pc-windows-msvc`       | Windows arm64        | tested — node 22, 24            |
-| `i686-pc-windows-msvc`          | Windows x32          | built, not tested               |
-| `x86_64-apple-darwin`           | macOS x64            | tested — node 22, 24            |
-| `aarch64-apple-darwin`          | macOS arm64          | tested — node 22, 24            |
-| `x86_64-unknown-linux-gnu`      | Linux x64 gnu        | tested — node 22, 24            |
-| `x86_64-unknown-linux-musl`     | Linux x64 musl       | tested — node 22, 24            |
-| `aarch64-unknown-linux-gnu`     | Linux arm64 gnu      | tested — node 22, 24            |
-| `aarch64-unknown-linux-musl`    | Linux arm64 musl     | tested — node 22, 24            |
-| `armv7-unknown-linux-gnueabihf` | Linux armv7 gnu      | tested — node 22 only           |
-| `s390x-unknown-linux-gnu`       | Linux s390x          | tested — node 22, 24            |
-| `x86_64-unknown-freebsd`        | FreeBSD x64          | built, not tested               |
-| `powerpc64le-unknown-linux-gnu` | Linux ppc64le        | built, not tested               |
-| `riscv64gc-unknown-linux-gnu`   | Linux riscv64        | built, not tested               |
-| `aarch64-linux-android`         | Android arm64        | built, not tested               |
-| `arm-linux-androideabi`         | Android armv7        | built, not tested               |
-| `aarch64-unknown-linux-ohos`    | OpenHarmony arm64    | built, not tested               |
+| Rust triple                     | Platform             | CI                                      |
+| ------------------------------- | -------------------- | --------------------------------------- |
+| `x86_64-pc-windows-msvc`        | Windows x64          | tested — node 22, 24                    |
+| `aarch64-pc-windows-msvc`       | Windows arm64        | tested — node 22, 24                    |
+| `i686-pc-windows-msvc`          | Windows x32          | built, not tested                       |
+| `x86_64-apple-darwin`           | macOS x64            | tested — node 22, 24                    |
+| `aarch64-apple-darwin`          | macOS arm64          | tested — node 22, 24                    |
+| `x86_64-unknown-linux-gnu`      | Linux x64 gnu        | tested — node 22, 24                    |
+| `x86_64-unknown-linux-musl`     | Linux x64 musl       | tested — node 22, 24                    |
+| `aarch64-unknown-linux-gnu`     | Linux arm64 gnu      | tested — node 22, 24                    |
+| `aarch64-unknown-linux-musl`    | Linux arm64 musl     | tested — node 22, 24                    |
+| `armv7-unknown-linux-gnueabihf` | Linux armv7 gnu      | tested — node 22 only                   |
+| `s390x-unknown-linux-gnu`       | Linux s390x          | tested — node 22, 24                    |
+| `x86_64-unknown-freebsd`        | FreeBSD x64          | built, not tested                       |
+| `powerpc64le-unknown-linux-gnu` | Linux ppc64le        | built, not tested                       |
+| `riscv64gc-unknown-linux-gnu`   | Linux riscv64        | built, not tested                       |
+| `aarch64-linux-android`         | Android arm64        | built, not tested                       |
+| `arm-linux-androideabi`         | Android armv7        | built, not tested                       |
+| `aarch64-unknown-linux-ohos`    | OpenHarmony arm64    | built, not tested                       |
 | `wasm32-wasip1-threads`         | wasm32-wasi, browser | tested — node 24 (`NAPI_RS_FORCE_WASI`) |
 
 Eighteen targets: eleven CI-tested, seven built but not exercised.
@@ -81,13 +81,62 @@ served with `Cross-Origin-Opener-Policy: same-origin` and
 
 ## API
 
+### One-shot (raw Snappy block format)
+
 ```ts
 export function compressSync(input: Buffer | string | ArrayBuffer | Uint8Array): Buffer
 export function compress(input: Buffer | string | ArrayBuffer | Uint8Array): Promise<Buffer>
 export function uncompressSync(compressed: Buffer): Buffer
 export function uncompress(compressed: Buffer): Promise<Buffer>
 ```
 
+### Streaming (framed Snappy format)
+
+Streaming uses the [Snappy frame format](https://github.com/google/snappy/blob/master/framing_format.txt)
+(file extension `.sz`). This is **not** the same wire format as the one-shot APIs
+above — framed output cannot be passed to `uncompress()`, and raw blocks cannot
+be passed to the stream decompressors.
+
+#### Incremental classes
+
+```js
+import { Compressor, Decompressor } from 'snappy'
+
+const compressor = new Compressor()
+const parts = [compressor.update('Hello '), compressor.update('snappy 🚀'), await compressor.finish()]
+const compressed = Buffer.concat(parts)
+
+const decompressor = new Decompressor()
+const restored = Buffer.concat([decompressor.update(compressed), await decompressor.finish()])
+console.log(restored.toString('utf8')) // Hello snappy 🚀
+```
+
+The valid stream is the concatenation of every `update()` output plus the `finish()` tail.
+
+#### Web Streams
+
+```js
+import { compressStream, uncompressStream } from 'snappy'
+
+const restored = uncompressStream(compressStream(source)) // ReadableStream<Uint8Array>
+```
+
+`input` must be a WHATWG `ReadableStream`; wrap a Node `Readable` with `Readable.toWeb()`.
+
+On wasm / browser builds the native transforms are unavailable; a buffered class-API
+polyfill is used automatically.
+
+#### Node Duplex factories
+
+```js
+import { createReadStream, createWriteStream } from 'node:fs'
+import { createCompressStream, createUncompressStream } from 'snappy'
+
+createReadStream('input.txt').pipe(createCompressStream()).pipe(createWriteStream('input.txt.sz'))
+```
+
+Requires a modern Node.js with Web Streams and `Duplex.fromWeb` (effectively Node 18+).
+
 ## Performance
 
 ### Hardware

__test__/helpers.ts

@@ -0,0 +1,106 @@
+/**
+ * Shared streaming-test helpers for framed Snappy.
+ */
+import { createRequire } from 'node:module'
+
+const requireFrom = createRequire(import.meta.url)
+
+/** True when the suite runs against the WASI-forced binding (`NAPI_RS_FORCE_WASI`). */
+export const IS_WASI = !!process.env.NAPI_RS_FORCE_WASI
+
+export const IS_SLOW_EMULATED_ARCH = ['s390x', 'ppc64', 'ppc64le'].includes(process.arch)
+
+export const MAX_EMULATED_FIXTURE_BYTES = 4 * 1024 * 1024
+
+export const runsFixtureOfSize = (byteLength: number): boolean =>
+  !IS_SLOW_EMULATED_ARCH || byteLength <= MAX_EMULATED_FIXTURE_BYTES
+
+export interface CompressorInstance {
+  update(chunk: string | Uint8Array): Buffer
+  finish(): Promise<Buffer>
+}
+
+export interface DecompressorInstance {
+  update(chunk: Uint8Array): Buffer
+  finish(): Promise<Buffer>
+}
+
+export function loadBinding() {
+  // Prefer the honest package entry (main.js) so stream factories resolve.
+  return requireFrom('..') as typeof import('..')
+}
+
+export function chunkBySize(buf: Buffer, size: number): Uint8Array[] {
+  if (buf.length === 0) {
+    return []
+  }
+  const chunks: Uint8Array[] = []
+  for (let i = 0; i < buf.length; i += size) {
+    chunks.push(buf.subarray(i, Math.min(i + size, buf.length)))
+  }
+  return chunks
+}
+
+export function chunkByByte(buf: Buffer): Uint8Array[] {
+  return chunkBySize(buf, 1)
+}
+
+/** Drive class compressor over chunks; return full framed stream. */
+export async function driveClassCompress(chunks: Array<string | Uint8Array>): Promise<Buffer> {
+  const { Compressor } = loadBinding()
+  const compressor = new Compressor()
+  const parts: Buffer[] = []
+  for (const chunk of chunks) {
+    parts.push(Buffer.from(compressor.update(chunk)))
+  }
+  parts.push(Buffer.from(await compressor.finish()))
+  return Buffer.concat(parts)
+}
+
+/** Drive class decompressor over framed chunks; return plaintext. */
+export async function driveClassUncompress(chunks: Uint8Array[]): Promise<Buffer> {
+  const { Decompressor } = loadBinding()
+  const decompressor = new Decompressor()
+  const parts: Buffer[] = []
+  for (const chunk of chunks) {
+    parts.push(Buffer.from(decompressor.update(chunk)))
+  }
+  parts.push(Buffer.from(await decompressor.finish()))
+  return Buffer.concat(parts)
+}
+
+/** Collect a Web ReadableStream into one Buffer. */
+export async function collectWebStream(stream: ReadableStream<Uint8Array>): Promise<Buffer> {
+  const reader = stream.getReader()
+  const chunks: Buffer[] = []
+  try {
+    for (;;) {
+      const { done, value } = await reader.read()
+      if (done) break
+      if (value && value.length) {
+        chunks.push(Buffer.from(value))
+      }
+    }
+  } finally {
+    try {
+      reader.releaseLock()
+    } catch {
+      // ignore
+    }
+  }
+  return Buffer.concat(chunks)
+}
+
+/** Wrap Buffer chunks as a WHATWG ReadableStream. */
+export function bufferToStream(chunks: Uint8Array[]): ReadableStream<Uint8Array> {
+  let i = 0
+  return new ReadableStream({
+    pull(controller) {
+      if (i >= chunks.length) {
+        controller.close()
+        return
+      }
+      controller.enqueue(chunks[i++])
+    },
+  })
+}

__test__/node-stream-factory.spec.ts

@@ -0,0 +1,59 @@
+import { createReadStream, createWriteStream } from 'node:fs'
+import { mkdtemp, readFile, rm, writeFile } from 'node:fs/promises'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { Readable } from 'node:stream'
+import { pipeline } from 'node:stream/promises'
+
+import test from 'ava'
+
+import { chunkBySize, driveClassCompress, loadBinding } from './helpers'
+
+const INPUT = Buffer.from('Node-stream factory 🚀 snappy bridge '.repeat(4096), 'utf8')
+
+const collect = async (readable: NodeJS.ReadableStream): Promise<Buffer> => {
+  const chunks: Buffer[] = []
+  for await (const chunk of readable) {
+    chunks.push(Buffer.from(chunk as Uint8Array))
+  }
+  return Buffer.concat(chunks)
+}
+
+test('createCompressStream → createUncompressStream round-trips (piped, multi-chunk)', async (t) => {
+  const { createCompressStream, createUncompressStream } = loadBinding()
+  const compressed = await collect(Readable.from(chunkBySize(INPUT, 64 * 1024)).pipe(createCompressStream()))
+  const restored = await collect(Readable.from(chunkBySize(compressed, 4096)).pipe(createUncompressStream()))
+  t.deepEqual(restored, INPUT)
+})
+
+test('createCompressStream output decodes via class Decompressor', async (t) => {
+  const { createCompressStream } = loadBinding()
+  const compressed = await collect(Readable.from([INPUT]).pipe(createCompressStream()))
+  const { Decompressor } = loadBinding()
+  const d = new Decompressor()
+  const head = Buffer.from(d.update(compressed))
+  const tail = Buffer.from(await d.finish())
+  t.deepEqual(Buffer.concat([head, tail]), INPUT)
+})
+
+test('createUncompressStream decodes class-compressed framed stream', async (t) => {
+  const { createUncompressStream } = loadBinding()
+  const compressed = await driveClassCompress([INPUT])
+  const restored = await collect(Readable.from(chunkBySize(compressed, 7)).pipe(createUncompressStream()))
+  t.deepEqual(restored, INPUT)
+})
+
+test('fs.createReadStream → createCompressStream → createUncompressStream → file round-trips', async (t) => {
+  const { createCompressStream, createUncompressStream } = loadBinding()
+  const dir = await mkdtemp(join(tmpdir(), 'snappy-factory-'))
+  t.teardown(() => rm(dir, { recursive: true, force: true }))
+  const srcPath = join(dir, 'input.bin')
+  const szPath = join(dir, 'output.sz')
+  const outPath = join(dir, 'restored.bin')
+  await writeFile(srcPath, INPUT)
+
+  await pipeline(createReadStream(srcPath), createCompressStream(), createWriteStream(szPath))
+  await pipeline(createReadStream(szPath), createUncompressStream(), createWriteStream(outPath))
+
+  t.deepEqual(await readFile(outPath), INPUT)
+})

__test__/streaming-class.spec.ts

@@ -0,0 +1,93 @@
+import test from 'ava'
+
+import { chunkByByte, chunkBySize, driveClassCompress, driveClassUncompress, loadBinding } from './helpers'
+
+const INPUT = Buffer.from('Hello 🚀'.repeat(500), 'utf8')
+
+const emptyAnd1Byte = (buf: Buffer): Uint8Array[] => {
+  const chunks: Uint8Array[] = [Buffer.alloc(0)]
+  for (const byte of chunkByByte(buf)) {
+    chunks.push(byte)
+    chunks.push(Buffer.alloc(0))
+  }
+  return chunks
+}
+
+const CHUNKINGS: ReadonlyArray<{ name: string; split: (buf: Buffer) => Uint8Array[] }> = [
+  { name: '1-byte', split: (buf) => chunkByByte(buf) },
+  { name: '64-byte', split: (buf) => chunkBySize(buf, 64) },
+  { name: 'single-chunk', split: (buf) => [buf] },
+  { name: 'awkward empty+1-byte', split: (buf) => emptyAnd1Byte(buf) },
+]
+
+for (const { name, split } of CHUNKINGS) {
+  test(`class compress round-trips via class uncompress (${name})`, async (t) => {
+    const compressed = await driveClassCompress(split(INPUT))
+    const restored = await driveClassUncompress(chunkBySize(compressed, 64))
+    t.deepEqual(restored, INPUT)
+  })
+}
+
+test('class compress output is byte-identical across all chunkings', async (t) => {
+  const reference = await driveClassCompress([INPUT])
+  for (const { name, split } of CHUNKINGS) {
+    const got = await driveClassCompress(split(INPUT))
+    t.true(got.equals(reference), `${name} output diverged from the single-chunk reference`)
+  }
+})
+
+test('class compress of empty input decodes back to empty', async (t) => {
+  const compressed = await driveClassCompress([])
+  const restored = await driveClassUncompress([compressed])
+  t.is(restored.length, 0)
+})
+
+const STRING_CHUNK = 'Hello 🚀 streaming string chunk — Ünïcöde'
+
+test('class compress accepts a string chunk (UTF-8) and round-trips', async (t) => {
+  const compressed = await driveClassCompress([STRING_CHUNK])
+  const restored = await driveClassUncompress([compressed])
+  t.deepEqual(restored, Buffer.from(STRING_CHUNK, 'utf8'))
+})
+
+test('a string chunk compresses byte-identically to the equivalent Uint8Array', async (t) => {
+  const fromString = await driveClassCompress([STRING_CHUNK])
+  const fromBytes = await driveClassCompress([Buffer.from(STRING_CHUNK, 'utf8')])
+  t.true(fromString.equals(fromBytes))
+})
+
+test('double finish on Compressor rejects', async (t) => {
+  const { Compressor } = loadBinding()
+  const c = new Compressor()
+  await c.finish()
+  // finish() takes the encoder synchronously, so a second call throws (not rejects).
+  t.throws(() => c.finish(), { code: 'InvalidArg' })
+})
+
+test('double finish on Decompressor rejects', async (t) => {
+  const { Decompressor } = loadBinding()
+  const d = new Decompressor()
+  await d.finish()
+  t.throws(() => d.finish(), { code: 'InvalidArg' })
+})
+
+test('framed stream output is NOT valid raw uncompress input', async (t) => {
+  const { uncompressSync } = loadBinding()
+  const framed = await driveClassCompress([INPUT])
+  t.throws(() => uncompressSync(framed), { any: true })
+})
+
+test('corrupt framed input surfaces InvalidArg', async (t) => {
+  const { Decompressor } = loadBinding()
+  const d = new Decompressor()
+  // Snappy frame magic is sNaPpY; garbage should fail at finish or update.
+  const garbage = Buffer.from('this is not framed snappy data at all!!!!!')
+  try {
+    d.update(garbage)
+    await d.finish()
+    t.fail('expected decode error')
+  } catch (err) {
+    t.true(err instanceof Error)
+    t.is((err as { code?: string }).code, 'InvalidArg')
+  }
+})