stream: preserve object mode in compose

Fixes: https://github.com/nodejs/node/issues/46829

PR-URL: https://github.com/nodejs/node/pull/47413
Reviewed-By: Robert Nagy <ronagy@icloud.com>
Reviewed-By: Benjamin Gruenbaum <benjamingr@gmail.com>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>

lib/internal/streams/compose.js

@@ -115,7 +115,7 @@ module.exports = function compose(...streams) {
   d = new Duplex({
     // TODO (ronag): highWaterMark?
     writableObjectMode: !!head?.writableObjectMode,
-    readableObjectMode: !!tail?.writableObjectMode,
+    readableObjectMode: !!tail?.readableObjectMode,
     writable,
     readable,
   });

test/parallel/test-stream-compose.js

@@ -420,3 +420,77 @@ const assert = require('assert');
     assert.strictEqual(buf, 'HELLOWORLD');
   }));
 }
+
+{
+  // In the new stream than should use the writeable of the first stream and readable of the last stream
+  // #46829
+  (async () => {
+    const newStream = compose(
+      new PassThrough({
+        // reading FROM you in object mode or not
+        readableObjectMode: false,
+
+        // writing TO you in object mode or not
+        writableObjectMode: false,
+      }),
+      new Transform({
+        // reading FROM you in object mode or not
+        readableObjectMode: true,
+
+        // writing TO you in object mode or not
+        writableObjectMode: false,
+        transform: (chunk, encoding, callback) => {
+          callback(null, {
+            value: chunk.toString()
+          });
+        }
+      })
+    );
+
+    assert.strictEqual(newStream.writableObjectMode, false);
+    assert.strictEqual(newStream.readableObjectMode, true);
+
+    newStream.write('Steve Rogers');
+    newStream.write('On your left');
+
+    newStream.end();
+
+    assert.deepStrictEqual(await newStream.toArray(), [{ value: 'Steve Rogers' }, { value: 'On your left' }]);
+  })().then(common.mustCall());
+}
+
+{
+  // In the new stream than should use the writeable of the first stream and readable of the last stream
+  // #46829
+  (async () => {
+    const newStream = compose(
+      new PassThrough({
+        // reading FROM you in object mode or not
+        readableObjectMode: true,
+
+        // writing TO you in object mode or not
+        writableObjectMode: true,
+      }),
+      new Transform({
+        // reading FROM you in object mode or not
+        readableObjectMode: false,
+
+        // writing TO you in object mode or not
+        writableObjectMode: true,
+        transform: (chunk, encoding, callback) => {
+          callback(null, chunk.value);
+        }
+      })
+    );
+
+    assert.strictEqual(newStream.writableObjectMode, true);
+    assert.strictEqual(newStream.readableObjectMode, false);
+
+    newStream.write({ value: 'Steve Rogers' });
+    newStream.write({ value: 'On your left' });
+
+    newStream.end();
+
+    assert.deepStrictEqual(await newStream.toArray(), [Buffer.from('Steve RogersOn your left')]);
+  })().then(common.mustCall());
+}