bare-stream - npm Package Compare versions

Comparing version 2.11.0 to 2.12.0

index.d.ts

@@ -5,2 +5,4 @@ import EventEmitter, { EventMap } from 'bare-events'
 
+import { ReadableStream, WritableStream, CustomQueuingStrategy } from './web'
+
 type StreamEncoding = BufferEncoding | 'buffer'
@@ -53,2 +55,11 @@
 
+interface ReadableFromWebOptions {
+  encoding?: BufferEncoding
+  signal?: AbortSignal
+}
+
+interface ReadableToWebOptions {
+  strategy?: CustomQueuingStrategy
+}
+
 interface Readable<M extends ReadableEvents = ReadableEvents>
@@ -81,2 +92,6 @@ extends Stream<M>, AsyncIterable<unknown> {
   static isPaused(rs: Readable): boolean
+
+  static fromWeb(readableStream: ReadableStream, opts?: ReadableFromWebOptions): Readable
+
+  static toWeb(readable: Readable, opts?: ReadableToWebOptions): ReadableStream
 }
@@ -96,2 +111,6 @@
 
+interface WritableFromWebOptions {
+  signal?: AbortSignal
+}
+
 interface Writable<M extends WritableEvents = WritableEvents> extends Stream<M> {
@@ -122,2 +141,6 @@ _write(data: unknown, encoding: StreamEncoding, cb: StreamCallback): void
   static drained(ws: Writable): Promise<boolean>
+
+  static fromWeb(writableStream: WritableStream, opts?: WritableFromWebOptions): Writable
+
+  static toWeb(writable: Writable): WritableStream
 }
@@ -129,2 +152,4 @@
 
+interface DuplexFromWebOptions extends ReadableFromWebOptions, WritableFromWebOptions {}
+
 interface Duplex<M extends DuplexEvents = DuplexEvents> extends Readable<M>, Writable<M> {}
@@ -134,2 +159,9 @@
   constructor(opts?: DuplexOptions)
+
+  static fromWeb(
+    { readable: ReadableStream, writable: Writable },
+    opts?: DuplexFromWebOptions
+  ): Readable
+
+  static toWeb(readable: Readable, opts?: ReadableToWebOptions): ReadableStream
 }
@@ -136,0 +168,0 @@

index.js

 const stream = require('streamx')
+const { ReadableStream, WritableStream } = require('./web')
 
@@ -88,2 +89,15 @@ const defaultEncoding = 'utf8'
 
+  static fromWeb(readableStream, opts = {}) {
+    const stream = readableStream._stream
+
+    if (opts.encoding) stream.setEncoding(opts.encoding)
+    if (opts.signal) exports.addAbortSignal(opts.signal, stream)
+
+    return stream
+  }
+
+  static toWeb(readable, opts = {}) {
+    return new ReadableStream(readable, opts.strategy)
+  }
+
   async [Symbol.asyncDispose]() {
@@ -169,2 +183,14 @@ if (!this.destroyed) this.destroy()
 
+  static fromWeb(writableStream, opts = {}) {
+    const stream = writableStream._stream
+
+    if (opts.signal) exports.addAbortSignal(opts.signal, stream)
+
+    return stream
+  }
+
+  static toWeb(writable) {
+    return new WritableStream(writable)
+  }
+
   async [Symbol.asyncDispose]() {
@@ -263,2 +289,29 @@ if (!this.destroyed) this.destroy()
   }
+
+  static fromWeb({ readable: readableStream, writable: writableStream }, opts) {
+    const readable = exports.Readable.fromWeb(readableStream, opts)
+    const writable = exports.Readable.fromWeb(writableStream, opts)
+
+    const duplex = new exports.Duplex({
+      write(data, encoding, cb) {
+        writable.write(data, encoding, cb)
+      }
+    })
+
+    readable
+      .on('data', (data) => duplex.push(data))
+      .on('end', () => duplex.push(null))
+      .on('error', (err) => duplex.destroy(err))
+
+    writable.on('finish', () => duplex.end()).on('error', (err) => duplex.destroy(err))
+
+    return duplex
+  }
+
+  static toWeb(duplex) {
+    const readableStream = exports.Readable.toWeb(duplex)
+    const writableStream = exports.Writable.toWeb(duplex)
+
+    return { readable: readableStream, writable: writableStream }
+  }
 }
@@ -265,0 +318,0 @@

package.json

 {
   "name": "bare-stream",
-  "version": "2.11.0",
+  "version": "2.12.0",
   "description": "Streaming data for JavaScript",
@@ -27,3 +27,5 @@ "exports": {
   "scripts": {
-    "test": "prettier . --check && bare test.js"
+    "format": "prettier --write . && lunte --fix",
+    "lint": "prettier --check . && lunte",
+    "test": "brittle-bare --coverage test.js"
   },
@@ -49,2 +51,3 @@ "repository": {
     "brittle": "^3.5.2",
+    "lunte": "^1.6.0",
     "prettier": "^3.3.3",
@@ -51,0 +54,0 @@ "prettier-config-holepunch": "^2.0.0"

web.d.ts

@@ -120,1 +120,34 @@ export interface ReadableStreamDefaultReader {
 export function isWritableStream(value: unknown): value is WritableStream
+
+export interface TransformStreamDefaultController {
+  readonly desiredSize: number
+
+  enqueue(data: unknown): void
+  error(error?: unknown): void
+  terminate(): void
+}
+
+export class TransformStreamDefaultController {
+  constructor(stream: TransformStream)
+}
+
+export interface Transformer<S extends TransformStream = TransformStream> {
+  start?(this: S, controller: TransformStreamDefaultController): void
+  transform?(this: S, chunk: unknown, controller: TransformStreamDefaultController): void
+  flush?(this: S, controller: TransformStreamDefaultController): void
+}
+
+export interface TransformStream {
+  readonly writable: WritableStream
+  readonly readable: ReadableStream
+}
+
+export class TransformStream {
+  constructor(
+    transformer?: Transformer,
+    writableStrategy?: CustomQueuingStrategy,
+    readableStrategy?: CustomQueuingStrategy
+  )
+}
+
+export function isTransformStream(value: unknown): value is TransformStream

web.js

@@ -1,2 +0,2 @@
-const { Readable, Writable, getStreamError, isStreamx, isDisturbed } = require('streamx')
+const { Readable, Writable, Transform, getStreamError, isStreamx, isDisturbed } = require('streamx')
 const tee = require('teex')
@@ -6,2 +6,3 @@
 const writableKind = Symbol.for('bare.stream.writable.kind')
+const transformKind = Symbol.for('bare.stream.transform.kind')
 
@@ -211,19 +212,19 @@ // https://streams.spec.whatwg.org/#readablestreamdefaultreader
   async _open(starting, cb) {
+    let err = null
     try {
       await starting
-
-      cb(null)
-    } catch (err) {
-      cb(err)
+    } catch (e) {
+      err = e
     }
+    cb(err)
   }
 
   async _read(pull, cb) {
+    let err = null
     try {
       await pull()
-
-      cb(null)
-    } catch (err) {
-      cb(err)
+    } catch (e) {
+      err = e
     }
+    cb(err)
   }
@@ -327,4 +328,3 @@ }
 
-    const err = getStreamError(stream)
-
+    let err = getStreamError(stream)
     if (err) return Promise.reject(err)
@@ -335,2 +335,5 @@
     await Writable.drained(stream)
+
+    err = getStreamError(stream)
+    if (err) return Promise.reject(err)
   }
@@ -450,29 +453,29 @@
   async _open(starting, cb) {
+    let err = null
     try {
       await starting
-
-      cb(null)
-    } catch (err) {
-      cb(err)
+    } catch (e) {
+      err = e
     }
+    cb(err)
   }
 
   async _write(write, data, cb) {
+    let err = null
     try {
       await write(data, this._controller)
-
-      cb(null)
-    } catch (err) {
-      cb(err)
+    } catch (e) {
+      err = e
     }
+    cb(err)
   }
 
   async _destroy(closing, cb) {
+    let err = null
     try {
       await closing
-
-      cb(null)
-    } catch (err) {
-      cb(err)
+    } catch (e) {
+      err = e
     }
+    cb(err)
   }
@@ -493,2 +496,114 @@ }
 
+// https://streams.spec.whatwg.org/#transformstreamdefaultcontroller
+exports.TransformStreamDefaultController = class TransformStreamDefaultController {
+  constructor(stream) {
+    this._stream = stream
+  }
+
+  get desiredSize() {
+    const stream = this._stream._stream
+
+    return stream._readableState.highWaterMark - stream._readableState.buffered
+  }
+
+  enqueue(data) {
+    this._stream._stream.push(data)
+  }
+
+  error(err) {
+    this._stream._stream.destroy(err)
+  }
+
+  terminate() {
+    const stream = this._stream._stream
+
+    stream.push(null)
+    stream.destroy(new TypeError('Stream has been terminated'))
+  }
+}
+
+// https://streams.spec.whatwg.org/#transformstream
+class TransformStream {
+  static get [transformKind]() {
+    return 0 // Compatibility version
+  }
+
+  constructor(transformer = {}, writableStrategy = {}, readableStrategy = {}) {
+    const { start, transform, flush } = transformer
+
+    this._stream = new Transform({ ...writableStrategy, ...readableStrategy })
+
+    this._writable = new WritableStream(this._stream)
+    this._readable = new ReadableStream(this._stream)
+
+    this._controller = new exports.TransformStreamDefaultController(this)
+
+    if (start) {
+      this._stream._open = this._open.bind(this, start.call(this, this._controller))
+    }
+
+    if (transform) {
+      this._stream._write = this._transform.bind(this, transform)
+    }
+
+    if (flush) {
+      this._stream._flush = this._flush.bind(this, flush.call(this, this._controller))
+    }
+  }
+
+  get [transformKind]() {
+    return TransformStream[transformKind]
+  }
+
+  get writable() {
+    return this._writable
+  }
+
+  get readable() {
+    return this._readable
+  }
+
+  async _open(starting, cb) {
+    let err = null
+    try {
+      await starting
+    } catch (e) {
+      err = e
+    }
+    cb(err)
+  }
+
+  async _transform(transform, data, cb) {
+    let err = null
+    try {
+      await transform(data, this._controller)
+    } catch (e) {
+      err = e
+    }
+    cb(err)
+  }
+
+  async _flush(flush, cb) {
+    let err = null
+    try {
+      await flush
+    } catch (e) {
+      err = e
+    }
+    cb(err)
+  }
+}
+
+exports.TransformStream = TransformStream
+
+exports.isTransformStream = function isTransformStream(value) {
+  if (value instanceof TransformStream) return true
+
+  return (
+    typeof value === 'object' &&
+    value !== null &&
+    value[transformKind] === TransformStream[transformKind]
+  )
+}
+
 function noop() {}

No alert changes

Improved metrics

Total package byte prevSize

49254

13.94%

Lines of code

1122

19.23%

Worsened metrics

Dev dependency count

7

16.67%

No dependency changes