feat: allow passing a http.Agent to the grpc client

packages/ipfs-client/README.md

@@ -28,6 +28,7 @@ An optional object which may have the following keys:
 | ---- | ---- | ------- | ----------- |
 | grpc | `Multiaddr` or `string` or `URL` | `undefined` | The address of a [ipfs-grpc-server][] to connect to |
 | http | `Multiaddr` or `string` or `URL` | `undefined` | The address of a [ipfs-http-server][] to connect to |
+| agent | [http.Agent](https://nodejs.org/api/http.html#http_class_http_agent) | `undefined` | A http.Agent used to control HTTP client behaviour (node.js only) |
 
 ### Returns
 

packages/ipfs-grpc-client/README.md

@@ -35,6 +35,7 @@ An optional object which may have the following keys:
 | Name | Type | Default | Description |
 | ---- | ---- | ------- | ----------- |
 | url | `Multiaddr` or `string` or `URL` | `undefined` | The address of a [ipfs-grpc-server][] to connect to |
+| agent | [http.Agent](https://nodejs.org/api/http.html#http_class_http_agent) | `undefined` | A http.Agent used to control HTTP client behaviour (node.js only) |
 
 ### Returns
 

packages/ipfs-grpc-client/src/core-api/add-all.js

@@ -79,7 +79,8 @@ module.exports = function grpcAddAll (grpc, service, opts = {}) {
     } = bidiToDuplex(grpc, service, {
       host: opts.url,
       debug: Boolean(process.env.DEBUG),
-      metadata: options
+      metadata: options,
+      agent: opts.agent
     })
 
     sendFiles(stream, sink)

packages/ipfs-grpc-client/src/core-api/files/ls.js

@@ -13,7 +13,8 @@ module.exports = function grpcMfsLs (grpc, service, opts = {}) {
     for await (const result of serverStreamToIterator(grpc, service, request, {
       host: opts.url,
       debug: Boolean(process.env.DEBUG),
-      metadata: options
+      metadata: options,
+      agent: opts.agent
     })) {
       yield {
         name: result.name,

packages/ipfs-grpc-client/src/core-api/files/write.js

@@ -38,7 +38,8 @@ module.exports = function grpcMfsWrite (grpc, service, opts = {}) {
     await clientStreamToPromise(grpc, service, stream(path, content), {
       host: opts.url,
       debug: Boolean(process.env.DEBUG),
-      metadata: options
+      metadata: options,
+      agent: opts.agent
     })
   }
 

packages/ipfs-grpc-client/src/core-api/id.js

@@ -11,7 +11,8 @@ module.exports = function grpcId (grpc, service, opts = {}) {
 
     const res = await unaryToPromise(grpc, service, request, {
       host: opts.url,
-      metadata: toHeaders(options)
+      metadata: toHeaders(options),
+      agent: opts.agent
     })
 
     return {

packages/ipfs-grpc-client/src/grpc/transport.node.js

@@ -13,12 +13,32 @@ const WebsocketSignal = {
 
 const finishSendFrame = new Uint8Array([1])
 
-function WebsocketTransport () {
-  return (opts) => {
-    return websocketRequest(opts)
+/**
+ * @param {object} options
+ * @param {import('http').Agent} [options.agent] - http.Agent used to control HTTP client behaviour
+ */
+function WebsocketTransport (options) {
+  /**
+   * @param {import('@improbable-eng/grpc-web').grpc.TransportOptions} opts
+   */
+  const websocketTransportFactory = (opts) => {
+    return websocketRequest({
+      ...options,
+      ...opts
+    })
   }
+
+  return websocketTransportFactory
 }
 
+/**
+ * @typedef {object} NodeTransportOptions
+ * @property {import('http').Agent} [options.agent]
+ *
+ * @typedef {NodeTransportOptions & import('@improbable-eng/grpc-web').grpc.TransportOptions} WebSocketTransportOptions
+ *
+ * @param {WebSocketTransportOptions} options
+ */
 function websocketRequest (options) {
   const webSocketAddress = constructWebSocketAddress(options.url)
 
@@ -54,7 +74,7 @@ function websocketRequest (options) {
       }
     },
     start: (metadata) => {
-      ws = new WebSocket(webSocketAddress, ['grpc-websockets'])
+      ws = new WebSocket(webSocketAddress, ['grpc-websockets'], options)
       ws.binaryType = 'arraybuffer'
       ws.onopen = function () {
         options.debug && debug('websocketRequest.onopen')
@@ -93,7 +113,8 @@ function constructWebSocketAddress (url) {
   } else if (url.substr(0, 7) === 'http://') {
     return `ws://${url.substr(7)}`
   }
-  throw new Error('Websocket transport constructed with non-https:// or http:// host.')
+
+  throw new Error('Websocket transport url must start with ws:// or wss:// or http:// or https://')
 }
 
 function headersToBytes (headers) {

packages/ipfs-grpc-client/src/index.js

@@ -1,10 +1,8 @@
 'use strict'
 
-const transport = require('./grpc/transport')
 const toUrlString = require('ipfs-core-utils/src/to-url-string')
 const loadServices = require('./utils/load-services')
 const { grpc } = require('@improbable-eng/grpc-web')
-grpc.setDefaultTransport(transport())
 
 const service = loadServices()
 
@@ -21,7 +19,12 @@ function normaliseUrls (opts) {
   })
 }
 
-module.exports = function createClient (opts = {}) {
+/**
+ * @param {object} opts
+ * @param {string} opts.url - The URL to connect to as a URL or Multiaddr
+ * @param {import('http').Agent} [opts.agent] - http.Agent used to control HTTP client behaviour (node.js only)
+ */
+module.exports = function createClient (opts = { url: '' }) {
   opts.url = toUrlString(opts.url)
 
   // @improbable-eng/grpc-web requires http:// protocol URLs, not ws://

packages/ipfs-grpc-client/src/utils/bidi-to-duplex.js

@@ -3,6 +3,7 @@
 const pushable = require('it-pushable')
 const errCode = require('err-code')
 const toHeaders = require('./to-headers')
+const transport = require('../grpc/transport')
 
 async function sendMessages (service, client, source) {
   for await (const obj of source) {
@@ -23,6 +24,7 @@ async function sendMessages (service, client, source) {
  * @param {string} options.host - The remote host
  * @param {boolean} [options.debug] - Whether to print debug messages
  * @param {object} [options.metadata] - Metadata sent as headers
+ * @param {import('http').Agent} [options.agent] - http.Agent used to control HTTP client behaviour (node.js only)
  * @returns {{ source: AsyncIterable<object>, sink: { push: Function, end: Function } }}
  **/
 module.exports = function bidiToDuplex (grpc, service, options) {
@@ -32,7 +34,12 @@ module.exports = function bidiToDuplex (grpc, service, options) {
   // @ts-ignore
   const sink = pushable()
 
-  const client = grpc.client(service, options)
+  const client = grpc.client(service, {
+    ...options,
+    transport: transport({
+      agent: options.agent
+    })
+  })
   client.onMessage(message => {
     sink.push(message)
   })

packages/ipfs-grpc-client/src/utils/client-stream-to-promise.js

@@ -15,6 +15,7 @@ const bidiToDuplex = require('./bidi-to-duplex')
  * @param {string} options.host - The remote host
  * @param {boolean} [options.debug] - Whether to print debug messages
  * @param {object} [options.metadata] - Metadata sent as headers
+ * @param {import('http').Agent} [options.agent] - http.Agent used to control HTTP client behaviour (node.js only)
  * @returns {Promise<Object>} - A promise that resolves to a response object
  **/
 module.exports = async function clientStreamToPromise (grpc, service, source, options) {

packages/ipfs-grpc-client/src/utils/server-stream-to-iterator.js

@@ -14,6 +14,7 @@ const bidiToDuplex = require('./bidi-to-duplex')
  * @param {string} options.host - The remote host
  * @param {boolean} [options.debug] - Whether to print debug messages
  * @param {object} [options.metadata] - Metadata sent as headers
+ * @param {import('http').Agent} [options.agent] - http.Agent used to control HTTP client behaviour (node.js only)
  * @returns {AsyncIterable<object>}
  **/
 module.exports = function serverStreamToIterator (grpc, service, request, options) {

packages/ipfs-grpc-client/src/utils/unary-to-promise.js

@@ -15,6 +15,7 @@ const bidiToDuplex = require('./bidi-to-duplex')
  * @param {string} options.host - The remote host
  * @param {boolean} [options.debug] - Whether to print debug messages
  * @param {object} [options.metadata] - Metadata sent as headers
+ * @param {import('http').Agent} [options.agent] - http.Agent used to control HTTP client behaviour (node.js only)
  * @returns {Promise<Object>} - A promise that resolves to a response object
  **/
 module.exports = function unaryToPromise (grpc, service, request, options) {

packages/ipfs-grpc-client/test/agent.js

@@ -0,0 +1,49 @@
+/* eslint-env mocha */
+'use strict'
+
+const grpcClient = require('../src')
+const WebSocket = require('ws')
+
+function startServer () {
+  return new Promise((resolve) => {
+    const wss = new WebSocket.Server({ port: 0 })
+
+    wss.on('listening', () => {
+      resolve({
+        port: wss.address().port,
+        close: () => wss.close()
+      })
+    })
+
+    wss.on('connection', (ws) => {
+      ws.once('message', () => {
+        ws.send('')
+        ws.end()
+      })
+    })
+  })
+}
+
+describe('agent', function () {
+  it('uses the passed agent', async () => {
+    const server = await startServer()
+
+    try {
+      await new Promise((resolve) => {
+        const ipfs = grpcClient({
+          url: `http://localhost:${server.port}`,
+          agent: {
+            addRequest () {
+              // an agent method was invoked
+              resolve()
+            }
+          }
+        })
+
+        ipfs.id().catch(() => {})
+      })
+    } finally {
+      server.close()
+    }
+  })
+})

packages/ipfs-grpc-client/test/node.js

@@ -0,0 +1,3 @@
+'use strict'
+
+require('./agent')

packages/ipfs-http-client/test/node/agent.js

@@ -82,6 +82,11 @@ describe('agent', function () {
 
         break
       }
+
+      if (i === 4) {
+        // should have first two responses by now
+        expect(responses).to.have.lengthOf(2)
+      }
     }
 
     // wait for the final request to arrive