stream: pipeline accept Buffer as a valid first argument

change isStream to also check existence of on, so it
wont mistake buffers as Streams.

fixes: #37731

Author: Linkgoron

lib/internal/streams/add-abort-signal.js

@@ -18,13 +18,13 @@ const validateAbortSignal = (signal, name) => {
   }
 };
 
-function isStream(obj) {
+function isReadable(obj) {
   return !!(obj && typeof obj.pipe === 'function');
 }
 
 module.exports.addAbortSignal = function addAbortSignal(signal, stream) {
   validateAbortSignal(signal, 'signal');
-  if (!isStream(stream)) {
+  if (!isReadable(stream)) {
     throw new ERR_INVALID_ARG_TYPE('stream', 'stream.Stream', stream);
   }
   return module.exports.addAbortSignalNoValidate(signal, stream);

lib/internal/streams/utils.js

@@ -6,11 +6,13 @@ const {
 } = primordials;
 
 function isReadable(obj) {
-  return !!(obj && typeof obj.pipe === 'function');
+  return !!(obj && typeof obj.pipe === 'function' &&
+            typeof obj.on === 'function');
 }
 
 function isWritable(obj) {
-  return !!(obj && typeof obj.write === 'function');
+  return !!(obj && typeof obj.write === 'function' &&
+            typeof obj.on === 'function');
 }
 
 function isStream(obj) {

test/parallel/test-stream-pipeline.js

@@ -1339,3 +1339,19 @@ const net = require('net');
     assert.strictEqual(res, '123');
   }));
 }
+
+{
+  const content = 'abc';
+  pipeline(Buffer.from(content), PassThrough({ objectMode: true }),
+           common.mustSucceed(() => {}));
+
+  let res = '';
+  pipeline(Buffer.from(content), async function*(previous) {
+    for await (const val of previous) {
+      res += String.fromCharCode(val);
+      yield val;
+    }
+  }, common.mustSucceed(() => {
+    assert.strictEqual(res, content);
+  }));
+}