fixup: benchmrk

Author: ronag

benchmark/streams/transform-by.js

@@ -16,8 +16,14 @@ function main({ n }) {
   s.resume();
 
   bench.start();
-  for (var k = 0; k < n; ++k) {
-    s.write(String.fromCharCode(n % 64 + 64));
+
+  let k = 0;
+  function run() {
+    while (k++ < n && s.write(b));
+    if (k >= n)
+      s.end();
   }
-  s.end(() => bench.end(n));
+  s.on('drain', run);
+  s.on('finish', () => bench.end(n));
+  run();
 }

doc/api/stream.md

@@ -1646,14 +1646,14 @@ Calling `Readable.from(string)` or `Readable.from(buffer)` will not have
 the strings or buffers be iterated to match the other streams semantics
 for performance reasons.
 
-### stream.Transform.by(asyncGeneratorFunction[, options])
+### stream.Transform.by(transform[, options])
 <!-- YAML
 added: REPLACEME
 -->
 
-* `asyncGeneratorFunction` {AsyncGeneratorFunction} A mapping function which
-accepts a `source` async iterable which can be used to read incoming data, while
-transformed data is pushed to the stream with the `yield` keyword.
+* `transform` {AsyncGeneratorFunction} A mapping function which accepts a `source`
+async iterable which can be used to read incoming data, while transformed data is
+pushed to the stream with the `yield` keyword.
 * `options` {Object} Options provided to `new stream.Transform([options])`.
 By default, `Transform.by()` will set `options.objectMode` to `true`,
 unless this is explicitly opted out by setting `options.objectMode` to `false`.

lib/_stream_transform.js

@@ -71,23 +71,12 @@ const {
 
 module.exports = Transform;
 const {
-  ERR_ARG_RETURN_VALUE_NOT_ASYNC_ITERABLE,
   ERR_METHOD_NOT_IMPLEMENTED,
   ERR_MULTIPLE_CALLBACK,
   ERR_TRANSFORM_ALREADY_TRANSFORMING,
   ERR_TRANSFORM_WITH_LENGTH_0
 } = require('internal/errors').codes;
 const Duplex = require('_stream_duplex');
-const Readable = require('_stream_readable');
-const AsyncIteratorPrototype = ObjectGetPrototypeOf(
-  ObjectGetPrototypeOf(async function* () {}).prototype);
-
-const kSourceIteratorPull = Symbol('kSourceIteratorPull');
-const kSourceIteratorResolve = Symbol('kSourceIteratorResolve');
-const kSourceIteratorChunk = Symbol('kSourceIteratorChunk');
-const kSourceIteratorStream = Symbol('kSourceIteratorStream');
-const kSourceIteratorPump = Symbol('kSourceIteratorPump');
-const kSourceIteratorGrabResolve = Symbol('kSourceIteratorGrabResolve');
 
 ObjectSetPrototypeOf(Transform.prototype, Duplex.prototype);
 ObjectSetPrototypeOf(Transform, Duplex);
@@ -235,10 +224,15 @@ function done(stream, er, data) {
 
 const from = require('internal/streams/from');
 
-Transform.by = function by(asyncGeneratorFn, opts) {
+Transform.by = function by(transform, opts) {
   let _resolve;
   let _promise = new Promise((resolve) => _resolve = resolve);
-  return from(Duplex, asyncGeneratorFn(async function*() {
+
+  if (typeof transform !== 'function') {
+    throw new ERR_INVALID_ARG_TYPE('transform', ['function'], iterable);
+  }
+
+  return from(Duplex, transform(async function*() {
     while (true) {
       const { chunk, done, cb } = await _promise;
       if (done) return cb();