test(readable): port setEncoding consume matrix

test/client-request.js

@@ -1092,6 +1092,43 @@ test('request multibyte text with setEncoding', async (t) => {
   await t.completed
 })
 
+test('#5611 - setEncoding() then .text() keeps a body received in several chunks', async (t) => {
+  t = tspl(t, { plan: 2 })
+
+  const text = 'abc傳def'
+  const buf = Buffer.from(text)
+  const sendRest = new EE.EventEmitter()
+
+  const server = createServer(async (req, res) => {
+    res.writeHead(200, { 'content-type': 'text/plain; charset=utf-8' })
+    res.write(buf.subarray(0, 4))
+    await EE.once(sendRest, 'go')
+    res.end(buf.subarray(4))
+  })
+  after(() => {
+    server.closeAllConnections?.()
+    server.close()
+  })
+
+  server.listen(0, async () => {
+    const client = new Client(`http://localhost:${server.address().port}`)
+    after(() => client.destroy())
+
+    const { body } = await client.request({ path: '/', method: 'GET' })
+    body.setEncoding('utf8')
+
+    const completed = EE.once(client, 'drain')
+    sendRest.emit('go')
+    await completed
+
+    const result = await body.text()
+    t.strictEqual(result, text)
+    t.strictEqual(Buffer.byteLength(result), buf.length)
+  })
+
+  await t.completed
+})
+
 test('#3736 - Aborted Response (without consuming body)', async (t) => {
   const plan = tspl(t, { plan: 1 })
 

test/readable.js

@@ -425,4 +425,100 @@ describe('Readable', () => {
 
     t.strictEqual(text, '')
   })
+
+  test('setEncoding() then .text() keeps chunks pushed after setEncoding()', async function (t) {
+    t = tspl(t, { plan: 2 })
+
+    function resume () {}
+    function abort () {}
+    const r = new Readable({ resume, abort })
+
+    // '傳' is 3 bytes in UTF-8, so cutting every 2 bytes splits each of them
+    // across a chunk boundary.
+    const expected = 'a傳b傳c傳d'
+    const buf = Buffer.from(expected)
+    const chunks = []
+    for (let n = 0; n < buf.length; n += 2) {
+      chunks.push(buf.subarray(n, n + 2))
+    }
+
+    r.push(chunks[0])
+    r.push(chunks[1])
+    r.setEncoding('utf8')
+    r.push(chunks[2])
+    r.push(chunks[3])
+
+    const promise = r.text()
+    r.push(chunks[4])
+
+    setImmediate(() => {
+      r.push(chunks[5])
+      r.push(chunks[6])
+      r.push(null)
+    })
+
+    const text = await promise
+    t.strictEqual(text, expected)
+    t.strictEqual(Buffer.byteLength(text), buf.length)
+  })
+
+  test('setEncoding() with only a partial character buffered', async function (t) {
+    t = tspl(t, { plan: 1 })
+
+    function resume () {}
+    function abort () {}
+    const r = new Readable({ resume, abort })
+    const buf = Buffer.from('傳')
+
+    r.push(buf.subarray(0, 1))
+    r.setEncoding('utf8')
+    r.push(buf.subarray(1))
+    process.nextTick(() => r.push(null))
+
+    t.strictEqual(await r.text(), '傳')
+  })
+
+  for (const encoding of ['utf8', 'hex', 'base64', 'latin1']) {
+    test(`setEncoding('${encoding}') before any chunk arrives`, async function (t) {
+      t = tspl(t, { plan: 5 })
+
+      function body () {
+        const r = new Readable({ resume () {}, abort () {} })
+        r.setEncoding(encoding)
+        const buf = Buffer.from('hello 傳 world')
+        r.push(buf.subarray(0, 4))
+        r.push(buf.subarray(4, 8))
+        process.nextTick(() => {
+          r.push(buf.subarray(8))
+          r.push(null)
+        })
+        return r
+      }
+
+      const buf = Buffer.from('hello 傳 world')
+      t.deepStrictEqual(await body().bytes(), new Uint8Array(buf))
+      t.deepStrictEqual(new Uint8Array(await body().arrayBuffer()), new Uint8Array(buf))
+
+      const blob = await body().blob()
+      t.strictEqual(blob.size, buf.length)
+      t.deepStrictEqual(Buffer.from(await blob.arrayBuffer()), buf)
+      t.strictEqual(await body().text(), buf.toString(encoding))
+    })
+  }
+
+  test('setEncoding() then .json()', async function (t) {
+    t = tspl(t, { plan: 1 })
+
+    const r = new Readable({ resume () {}, abort () {} })
+    const buf = Buffer.from(JSON.stringify({ hello: '傳' }))
+
+    r.setEncoding('utf8')
+    r.push(buf.subarray(0, 12))
+    process.nextTick(() => {
+      r.push(buf.subarray(12))
+      r.push(null)
+    })
+
+    t.deepStrictEqual(await r.json(), { hello: '傳' })
+  })
 })