Repository: nodejs/undici

Problem statement:
client.request throws Error with setEncoding
## Bug Description

`client.request` throws an error when `body.json()` & `body.text()` are called with `setEncoding`.

## Reproducible By

```javascript
import { createServer } from 'http'
import { Client } from 'undici'
import { once } from 'events'
import assert from 'assert'

const data = 'a'.repeat(100)

const server = createServer((request, response) => {
  response.end(JSON.stringify({ data }))
}).listen()

await once(server, 'listening')

const client = new Client(`http://localhost:${server.address().port}`)

try {
  const { body, headers, statusCode } = await client.request({
    path: '/',
    method: 'GET'
  })
  console.log(`response received ${statusCode}`)
  console.log('headers', headers)
  body.setEncoding('utf8')
  const json = await body.json()
  assert.strictEqual(json.data, data, 'multi byte')
} catch (error) {
  console.error('error!!:', error)
} finally {
  client.close()
  server.close()
}
```

```bash
$ node index.mjs
error!!: TypeError [ERR_INVALID_ARG_TYPE]: The "list[0]" argument must be an instance of Buffer or Uint8Array. Received type string ('{"data":"aaaaaaaaaaaaaaa...)
    at new NodeError (node:internal/errors:371:5)
    at Function.concat (node:buffer:559:13)
    at consumeEnd (/Users/kohtaito/dev/tmp/test_undici/node_modules/undici/lib/api/readable.js:238:33)
    at BodyReadable.<anonymous> (/Users/kohtaito/dev/tmp/test_undici/node_modules/undici/lib/api/readable.js:220:7)
    at BodyReadable.emit (node:events:390:28)
    at BodyReadable.emit (/Users/kohtaito/dev/tmp/test_undici/node_modules/undici/lib/api/readable.js:66:18)
    at endReadableNT (node:internal/streams/readable:1343:12)
    at processTicksAndRejections (node:internal/process/task_queues:83:21) {
  code: 'ERR_INVALID_ARG_TYPE'
}
```

## Expected Behavior

<!-- A clear and concise description of what you expected to happen. -->

## Logs & Screenshots

<!-- If applicable, add screenshots to help explain your problem, or
alternatively add your console logs here. -->

## Environment

Node.js v16.13.1
undici v4.11.0

### Additional context

`chunk` is converted to a string by StringDecoder only when consumeStart is executed.

```diff
$ git diff
diff --git a/lib/api/readable.js b/lib/api/readable.js
index 3bb454e..8ea758e 100644
--- a/lib/api/readable.js
+++ b/lib/api/readable.js
@@ -94,6 +94,7 @@ module.exports = class BodyReadable extends Readable {

   push (chunk) {
     if (this[kConsume] && chunk !== null && !this[kReading]) {
+      console.log('BodyReadable: push', typeof chunk, this.readableEncoding)
       consumePush(this[kConsume], chunk)
       return true
     } else {
@@ -196,6 +197,7 @@ function consumeStart (consume) {
   const { _readableState: state } = consume.stream

   for (const chunk of state.buffer) {
+    console.log('consumeStart', typeof chunk, consume.stream.readableEncoding)
     consumePush(consume, chunk)
   }
```

```bash
$ node index.mjs
response received 200
headers {
  date: 'Mon, 06 Dec 2021 09:08:48 GMT',
  connection: 'keep-alive',
  'keep-alive': 'timeout=5',
  'content-length': '10000011'
}
consumeStart string utf8
BodyReadable: push object utf8
BodyReadable: push object utf8
BodyReadable: push object utf8
BodyReadable: push object utf8
BodyReadable: push object utf8
```

I think that this code needs to use `consume.stream._readableState.decoder.end()` in `consumeEnd` instead of `Buffer.concat(body)` if the value is in `readableEncoding`.
Is this the right way to fix it?

I also considered a method to overwrite StringDecoder so that it is not created in setEncoding.
However, it seems that it will be difficult to upgrade because it depends too much on the structure of Stream.

Ref: https://github.com/nodejs/undici/issues/1119

CC: @mcollina 


Hints:
> I think that this code needs to use consume.stream._readableState.decoder.end() in consumeEnd instead of Buffer.concat(body) if the value is in readableEncoding.
Is this the right way to fix it?

Just check if elements in body are string or Buffer and concat accordingly.
I think that mixing strings and buffers could result in data loss at chunk boundaries.
Isn't this a problem?

Then it seems that it can be made simple.
```javascript
if (typeof chunk === 'str')  {
  chunk = Buffer.from(chunk)
} 
```
I think that would work.
okey. I'll do test & create PR.
Thanks.
@ronag 
I tested this code.

```diff
diff --git a/lib/api/readable.js b/lib/api/readable.js
index 3bb454e..c1a80db 100644
--- a/lib/api/readable.js
+++ b/lib/api/readable.js
@@ -196,6 +196,10 @@ function consumeStart (consume) {
   const { _readableState: state } = consume.stream

   for (const chunk of state.buffer) {
+    if (consume.stream.readableEncoding) {
+      consumePush(consume, Buffer.from(chunk, consume.stream.readableEncoding))
+      continue
+    }
     consumePush(consume, chunk)
   }
```

The first reproduction code now works with this fix.

But this reproduction code doesn't work.

```javascript
import { createServer } from 'http'
import { Client } from 'undici'
import { once } from 'events'
import assert from 'assert'

const data = Buffer.from('あいうえお')

const server = createServer((request, response) => {
  response.write(data.slice(0, 1))
  setTimeout(() => {
    response.write(data.slice(1))
    response.end()
  }, 100)
}).listen()

await once(server, 'listening')

const client = new Client(`http://localhost:${server.address().port}`)

try {
  const { body, headers, statusCode } = await client.request({
    path: '/',
    method: 'GET'
  })
  console.log(`response received ${statusCode}`)
  console.log('headers', headers)
  body.setEncoding('utf8')
  const text = await body.text()
  console.log(Buffer.from(text))
  console.log(Buffer.from(data))
  assert.strictEqual(text, data, 'multi byte')
} catch (error) {
  console.error('error!!:', error)
} finally {
  client.close()
  server.close()
}
```

```bash
$ node test-first-byte.mjs
response received 200
headers {
  date: 'Tue, 07 Dec 2021 05:31:38 GMT',
  connection: 'keep-alive',
  'keep-alive': 'timeout=5',
  'transfer-encoding': 'chunked'
}
<Buffer ef bf bd ef bf bd e3 81 84 e3 81 86 e3 81 88 e3 81 8a>
<Buffer e3 81 82 e3 81 84 e3 81 86 e3 81 88 e3 81 8a>
error!!: AssertionError [ERR_ASSERTION]: multi byte
    at file:///tmp/test_undici/index2.mjs:31:10
    at processTicksAndRejections (node:internal/process/task_queues:96:5) {
  generatedMessage: false,
  code: 'ERR_ASSERTION',
  actual: '��いうえお',
  expected: [Buffer [Uint8Array]],
  operator: 'strictEqual'
}
```

It looks like the first few bytes are lost. So I think we need another way.
I you open a PR with the tests then we can collaborate on it.
OK. Thanks.
I'll open PR.

Patch:
diff --git a/lib/api/readable.js b/lib/api/readable.js
--- a/lib/api/readable.js
+++ b/lib/api/readable.js
@@ -201,6 +201,17 @@ class BodyReadable extends Readable {
         .resume()
     })
   }
+
+  /**
+   * @param {BufferEncoding} encoding
+   * @returns {BodyReadable}
+   */
+  setEncoding (encoding) {
+    if (Buffer.isEncoding(encoding)) {
+      this._readableState.encoding = encoding
+    }
+    return this
+  }
 }
 
 // https://streams.spec.whatwg.org/#readablestream-locked
@@ -278,10 +289,10 @@ function consumeStart (consume) {
   }
 
   if (state.endEmitted) {
-    consumeEnd(this[kConsume])
+    consumeEnd(this[kConsume], this._readableState.encoding)
   } else {
     consume.stream.on('end', function () {
-      consumeEnd(this[kConsume])
+      consumeEnd(this[kConsume], this._readableState.encoding)
     })
   }
 
@@ -295,8 +306,10 @@ function consumeStart (consume) {
 /**
  * @param {Buffer[]} chunks
  * @param {number} length
+ * @param {BufferEncoding} encoding
+ * @returns {string}
  */
-function chunksDecode (chunks, length) {
+function chunksDecode (chunks, length, encoding) {
   if (chunks.length === 0 || length === 0) {
     return ''
   }
@@ -311,7 +324,11 @@ function chunksDecode (chunks, length) {
     buffer[2] === 0xbf
       ? 3
       : 0
-  return buffer.utf8Slice(start, bufferLength)
+  if (!encoding || encoding === 'utf8' || encoding === 'utf-8') {
+    return buffer.utf8Slice(start, bufferLength)
+  } else {
+    return buffer.subarray(start, bufferLength).toString(encoding)
+  }
 }
 
 /**
@@ -339,14 +356,14 @@ function chunksConcat (chunks, length) {
   return buffer
 }
 
-function consumeEnd (consume) {
+function consumeEnd (consume, encoding) {
   const { type, body, resolve, stream, length } = consume
 
   try {
     if (type === 'text') {
-      resolve(chunksDecode(body, length))
+      resolve(chunksDecode(body, length, encoding))
     } else if (type === 'json') {
-      resolve(JSON.parse(chunksDecode(body, length)))
+      resolve(JSON.parse(chunksDecode(body, length, encoding)))
     } else if (type === 'arrayBuffer') {
       resolve(chunksConcat(body, length).buffer)
     } else if (type === 'blob') {


Test patch:
diff --git a/test/client-request.js b/test/client-request.js
--- a/test/client-request.js
+++ b/test/client-request.js
@@ -220,7 +220,7 @@ test('request body destroyed on invalid callback', async (t) => {
     after(() => client.destroy())
 
     const body = new Readable({
-      read () {}
+      read () { }
     })
     try {
       client.request({
@@ -1252,3 +1252,88 @@ test('request post body DataView', async (t) => {
 
   await t.completed
 })
+
+test('request multibyte json with setEncoding', async (t) => {
+  t = tspl(t, { plan: 1 })
+
+  const asd = Buffer.from('あいうえお')
+  const data = JSON.stringify({ asd })
+  const server = createServer((req, res) => {
+    res.write(data.slice(0, 1))
+    setTimeout(() => {
+      res.write(data.slice(1))
+      res.end()
+    }, 100)
+  })
+  after(server.close.bind(server))
+
+  server.listen(0, async () => {
+    const client = new Client(`http://localhost:${server.address().port}`)
+    after(client.destroy.bind(client))
+
+    const { body } = await client.request({
+      path: '/',
+      method: 'GET'
+    })
+    body.setEncoding('utf8')
+    t.deepStrictEqual(JSON.parse(data), await body.json())
+  })
+
+  await t.completed
+})
+
+test('request multibyte text with setEncoding', async (t) => {
+  t = tspl(t, { plan: 1 })
+
+  const data = Buffer.from('あいうえお')
+  const server = createServer((req, res) => {
+    res.write(data.slice(0, 1))
+    setTimeout(() => {
+      res.write(data.slice(1))
+      res.end()
+    }, 100)
+  })
+  after(server.close.bind(server))
+
+  server.listen(0, async () => {
+    const client = new Client(`http://localhost:${server.address().port}`)
+    after(client.destroy.bind(client))
+
+    const { body } = await client.request({
+      path: '/',
+      method: 'GET'
+    })
+    body.setEncoding('utf8')
+    t.deepStrictEqual(data.toString('utf8'), await body.text())
+  })
+
+  await t.completed
+})
+
+test('request multibyte text with setEncoding', async (t) => {
+  t = tspl(t, { plan: 1 })
+
+  const data = Buffer.from('あいうえお')
+  const server = createServer((req, res) => {
+    res.write(data.slice(0, 1))
+    setTimeout(() => {
+      res.write(data.slice(1))
+      res.end()
+    }, 100)
+  })
+  after(server.close.bind(server))
+
+  server.listen(0, async () => {
+    const client = new Client(`http://localhost:${server.address().port}`)
+    after(client.destroy.bind(client))
+
+    const { body } = await client.request({
+      path: '/',
+      method: 'GET'
+    })
+    body.setEncoding('hex')
+    t.deepStrictEqual(data.toString('hex'), await body.text())
+  })
+
+  await t.completed
+})