undici - npm Package Compare versions

Comparing version 0.2.0 to 0.3.0

lib/client.js

@@ -45,3 +45,3 @@ 'use strict'
   socket.on('end', () => {
-    reconnect(client, new Error('other side closed - finished'))
+    reconnect(client, new Error('other side closed - ended'))
   })
@@ -149,7 +149,12 @@
       if (typeof body === 'string' || body instanceof Uint8Array) {
-        this.socket.write(`content-length: ${Buffer.byteLength(body)}\r\n\r\n`, 'ascii')
+        if (headers && headers.hasOwnProperty('content-length')) {
+          // we have already written the content-length header
+          this.socket.write('\r\n')
+        } else {
+          this.socket.write(`content-length: ${Buffer.byteLength(body)}\r\n\r\n`, 'ascii')
+        }
         this.socket.write(body)
       } else if (body && typeof body.pipe === 'function') {
         // TODO we should pause the queue while we are piping
-        if (headers && headers['content-length']) {
+        if (headers && headers.hasOwnProperty('content-length')) {
           this.socket.write('\r\n', 'ascii')
@@ -281,3 +286,3 @@ body.pipe(this.socket, { end: false })
       process.nextTick(cb, new Error('The client is closed'))
-      return
+      return false
     }
@@ -284,0 +289,0 @@

lib/pool.js

 'use strict'
 
 const Client = require('./client')
-const next = Symbol('next')
+const current = Symbol('current')
 
@@ -22,14 +22,43 @@ class Pool {
 
-    this[next] = 0
+    for (let client of this.clients) {
+      client.on('drain', onDrain)
+    }
+
+    this.drained = []
+    this[current] = null
+
+    const that = this
+    function onDrain () {
+      // this is the client
+      that.drained.push(this)
+    }
   }
 
   request (opts, cb) {
-    // TODO use a smarter algorithm than round robin
-    const current = this[next]++
+    // needed because we need the return value from client.request
+    if (cb === undefined) {
+      return new Promise((resolve, reject) => {
+        this.request(opts, (err, data) => {
+          return err ? reject(err) : resolve(data)
+        })
+      })
+    }
 
-    if (this[next] === this.clients.length) {
-      this[next] = 0
+    if (this[current] === null) {
+      if (this.drained.length > 0) {
+        // LIFO QUEUE
+        // we use the last one that drained, because that's the one
+        // that is more probable to have an alive socket
+        this[current] = this.drained.pop()
+      } else {
+        // if no one drained recently, let's just pick one randomly
+        this[current] = this.clients[Math.floor(Math.random() * this.clients.length)]
+      }
     }
 
-    return this.clients[current].request(opts, cb)
+    const writeMore = this[current].request(opts, cb)
+
+    if (!writeMore) {
+      this[current] = null
+    }
   }
@@ -36,0 +65,0 @@

package.json

 {
   "name": "undici",
-  "version": "0.2.0",
+  "version": "0.3.0",
   "description": "An HTTP/1.1 client, written from scratch for Node.js",
@@ -22,2 +22,3 @@ "main": "index.js",
     "pre-commit": "^1.2.2",
+    "proxyquire": "^2.0.1",
     "readable-stream": "^2.3.6",
@@ -24,0 +25,0 @@ "snazzy": "^7.1.1",

test/client.js

@@ -81,2 +81,59 @@ 'use strict'
 
+test('basic POST with empty string', (t) => {
+  t.plan(6)
+
+  const server = createServer(postServer(t, ''))
+  t.tearDown(server.close.bind(server))
+
+  server.listen(0, () => {
+    const client = new Client(`http://localhost:${server.address().port}`)
+    t.tearDown(client.close.bind(client))
+
+    client.request({ path: '/', method: 'POST', body: '' }, (err, { statusCode, headers, body }) => {
+      t.error(err)
+      t.strictEqual(statusCode, 200)
+      const bufs = []
+      body.on('data', (buf) => {
+        bufs.push(buf)
+      })
+      body.on('end', () => {
+        t.strictEqual('hello', Buffer.concat(bufs).toString('utf8'))
+      })
+    })
+  })
+})
+
+test('basic POST with string and content-length', (t) => {
+  t.plan(6)
+
+  const expected = readFileSync(__filename, 'utf8')
+
+  const server = createServer(postServer(t, expected))
+  t.tearDown(server.close.bind(server))
+
+  server.listen(0, () => {
+    const client = new Client(`http://localhost:${server.address().port}`)
+    t.tearDown(client.close.bind(client))
+
+    client.request({
+      path: '/',
+      method: 'POST',
+      headers: {
+        'content-length': Buffer.byteLength(expected)
+      },
+      body: expected
+    }, (err, { statusCode, headers, body }) => {
+      t.error(err)
+      t.strictEqual(statusCode, 200)
+      const bufs = []
+      body.on('data', (buf) => {
+        bufs.push(buf)
+      })
+      body.on('end', () => {
+        t.strictEqual('hello', Buffer.concat(bufs).toString('utf8'))
+      })
+    })
+  })
+})
+
 test('basic POST with Buffer', (t) => {
@@ -83,0 +140,0 @@ t.plan(6)

test/pool.js

 'use strict'
 
+const proxyquire = require('proxyquire')
 const { test } = require('tap')
 const { Pool } = require('..')
 const { createServer } = require('http')
+const { EventEmitter } = require('events')
+const { promisify } = require('util')
+const eos = require('end-of-stream')
 
@@ -36,1 +40,89 @@ test('basic get', (t) => {
 })
+
+test('basic get with async/await', async (t) => {
+  const server = createServer((req, res) => {
+    t.strictEqual('/', req.url)
+    t.strictEqual('GET', req.method)
+    res.setHeader('content-type', 'text/plain')
+    res.end('hello')
+  })
+  t.tearDown(server.close.bind(server))
+
+  await promisify(server.listen.bind(server))(0)
+  const client = new Pool(`http://localhost:${server.address().port}`)
+  t.tearDown(client.close.bind(client))
+
+  const { statusCode, headers, body } = await client.request({ path: '/', method: 'GET' })
+  t.strictEqual(statusCode, 200)
+  t.strictEqual(headers['content-type'], 'text/plain')
+
+  body.resume()
+  await promisify(eos)(body)
+})
+
+test('backpressure algorithm', (t) => {
+  const seen = []
+  let total = 0
+  let writeMore = false
+
+  class FakeClient extends EventEmitter {
+    constructor () {
+      super()
+
+      this.id = total++
+    }
+
+    request (req, cb) {
+      seen.push({ req, cb, client: this })
+      return writeMore
+    }
+  }
+
+  const Pool = proxyquire('../lib/pool', {
+    './client': FakeClient
+  })
+
+  const pool = new Pool(`http://notanhost`)
+
+  t.strictEqual(total, 10)
+
+  writeMore = true
+
+  pool.request({}, noop)
+  pool.request({}, noop)
+
+  const d1 = seen.shift()
+  const d2 = seen.shift()
+
+  t.strictEqual(d1.client, d2.client)
+
+  writeMore = false
+  pool.request({}, noop)
+
+  writeMore = true
+  pool.request({}, noop)
+
+  const d3 = seen.shift()
+  const d4 = seen.shift()
+
+  t.strictEqual(d3.client, d2.client)
+  t.notStrictEqual(d3.client, d4.client)
+
+  d3.client.emit('drain')
+
+  writeMore = false
+  pool.request({}, noop)
+
+  writeMore = true
+  pool.request({}, noop)
+
+  const d5 = seen.shift()
+  const d6 = seen.shift()
+
+  t.strictEqual(d5.client, d4.client)
+  t.strictEqual(d3.client, d6.client)
+
+  t.end()
+})
+
+function noop () {}

No alert changes

Improved metrics

Total package byte prevSize

53818

increased by9.21%

Lines of code

1617

increased by9.85%

Worsened metrics

Dev dependency count

7

increased by16.67%

No dependency changes