Revert "chore: less async await (#2463)"

This reverts commit e7ab79a.

Author: mcollina

lib/agent.js

@@ -118,7 +118,7 @@ class Agent extends DispatcherBase {
     return dispatcher.dispatch(opts, handler)
   }
 
-  [kClose] () {
+  async [kClose] () {
     const closePromises = []
     for (const ref of this[kClients].values()) {
       const client = ref.deref()
@@ -128,10 +128,10 @@ class Agent extends DispatcherBase {
       }
     }
 
-    return Promise.all(closePromises)
+    await Promise.all(closePromises)
   }
 
-  [kDestroy] (err) {
+  async [kDestroy] (err) {
     const destroyPromises = []
     for (const ref of this[kClients].values()) {
       const client = ref.deref()
@@ -141,7 +141,7 @@ class Agent extends DispatcherBase {
       }
     }
 
-    return Promise.all(destroyPromises)
+    await Promise.all(destroyPromises)
   }
 }
 

lib/api/readable.js

@@ -209,14 +209,12 @@ function isUnusable (self) {
   return util.isDisturbed(self) || isLocked(self)
 }
 
-function consume (stream, type) {
+async function consume (stream, type) {
   if (isUnusable(stream)) {
-    return Promise.reject(new TypeError('unusable'))
+    throw new TypeError('unusable')
   }
 
-  if (stream[kConsume]) {
-    return Promise.reject(new assert.AssertionError('null != true'))
-  }
+  assert(!stream[kConsume])
 
   return new Promise((resolve, reject) => {
     stream[kConsume] = {

lib/client.js

@@ -380,7 +380,7 @@ class Client extends DispatcherBase {
     return this[kNeedDrain] < 2
   }
 
-  [kClose] () {
+  async [kClose] () {
     // TODO: for H2 we need to gracefully flush the remaining enqueued
     // request and close each stream.
     return new Promise((resolve) => {
@@ -392,7 +392,7 @@ class Client extends DispatcherBase {
     })
   }
 
-  [kDestroy] (err) {
+  async [kDestroy] (err) {
     return new Promise((resolve) => {
       const requests = this[kQueue].splice(this[kPendingIdx])
       for (let i = 0; i < requests.length; i++) {

lib/fetch/body.js

@@ -234,8 +234,8 @@ function extractBody (object, keepalive = false) {
         }
         return controller.desiredSize > 0
       },
-      cancel (reason) {
-        return iterator.return()
+      async cancel (reason) {
+        await iterator.return()
       },
       type: undefined
     })

lib/fetch/index.js

@@ -1808,10 +1808,10 @@ async function httpNetworkFetch (
         fetchParams.controller.controller = controller
       },
       async pull (controller) {
-        pullAlgorithm(controller)
+        await pullAlgorithm(controller)
       },
       async cancel (reason) {
-        cancelAlgorithm(reason)
+        await cancelAlgorithm(reason)
       }
     },
     {

lib/pool-base.js

@@ -111,7 +111,7 @@ class PoolBase extends DispatcherBase {
     return this[kStats]
   }
 
-  [kClose] () {
+  async [kClose] () {
     if (this[kQueue].isEmpty()) {
       return Promise.all(this[kClients].map(c => c.close()))
     } else {
@@ -121,7 +121,7 @@ class PoolBase extends DispatcherBase {
     }
   }
 
-  [kDestroy] (err) {
+  async [kDestroy] (err) {
     while (true) {
       const item = this[kQueue].shift()
       if (!item) {

lib/proxy-agent.js

@@ -82,38 +82,40 @@ class ProxyAgent extends DispatcherBase {
     this[kClient] = clientFactory(resolvedUrl, { connect })
     this[kAgent] = new Agent({
       ...opts,
-      connect: (opts, callback) => {
+      connect: async (opts, callback) => {
         let requestedHost = opts.host
         if (!opts.port) {
           requestedHost += `:${defaultProtocolPort(opts.protocol)}`
         }
-        this[kClient].connect({
-          origin,
-          port,
-          path: requestedHost,
-          signal: opts.signal,
-          headers: {
-            ...this[kProxyHeaders],
-            host
-          }
-        }).catch(callback)
-          .then(({ socket, statusCode }) => {
-            if (statusCode !== 200) {
-              socket.on('error', () => { }).destroy()
-              callback(new RequestAbortedError('Proxy response !== 200 when HTTP Tunneling'))
-            }
-            if (opts.protocol !== 'https:') {
-              callback(null, socket)
-              return
+        try {
+          const { socket, statusCode } = await this[kClient].connect({
+            origin,
+            port,
+            path: requestedHost,
+            signal: opts.signal,
+            headers: {
+              ...this[kProxyHeaders],
+              host
             }
-            let servername
-            if (this[kRequestTls]) {
-              servername = this[kRequestTls].servername
-            } else {
-              servername = opts.servername
-            }
-            this[kConnectEndpoint]({ ...opts, servername, httpSocket: socket }, callback)
-          }).catch(callback)
+          })
+          if (statusCode !== 200) {
+            socket.on('error', () => {}).destroy()
+            callback(new RequestAbortedError('Proxy response !== 200 when HTTP Tunneling'))
+          }
+          if (opts.protocol !== 'https:') {
+            callback(null, socket)
+            return
+          }
+          let servername
+          if (this[kRequestTls]) {
+            servername = this[kRequestTls].servername
+          } else {
+            servername = opts.servername
+          }
+          this[kConnectEndpoint]({ ...opts, servername, httpSocket: socket }, callback)
+        } catch (err) {
+          callback(err)
+        }
       }
     })
   }
@@ -134,12 +136,14 @@ class ProxyAgent extends DispatcherBase {
     )
   }
 
-  [kClose] () {
-    return Promise.all([this[kAgent].close, this[kClient].close])
+  async [kClose] () {
+    await this[kAgent].close()
+    await this[kClient].close()
   }
 
-  [kDestroy] () {
-    return Promise.all([this[kAgent].destroy, this[kClient].destroy])
+  async [kDestroy] () {
+    await this[kAgent].destroy()
+    await this[kClient].destroy()
   }
 }