js-libp2p-tcp/src/socket-to-conn.js

'use strict'

const abortable = require('abortable-iterator')
const log = require('debug')('libp2p:tcp:socket')
const toIterable = require('stream-to-it')
const toMultiaddr = require('libp2p-utils/src/ip-port-to-multiaddr')
const { CLOSE_TIMEOUT } = require('./constants')

// Convert a socket into a MultiaddrConnection
// https://github.com/libp2p/interface-transport#multiaddrconnection
module.exports = (socket, options) => {
  options = options || {}

  // Check if we are connected on a unix path
  if (options.listeningAddr && options.listeningAddr.getPath()) {
    options.remoteAddr = options.listeningAddr
  }

  if (options.remoteAddr && options.remoteAddr.getPath()) {
    options.localAddr = options.remoteAddr
  }

  const { sink, source } = toIterable.duplex(socket)
  const maConn = {
    async sink (source) {
      if (options.signal) {
        source = abortable(source, options.signal)
      }

      try {
        await sink((async function * () {
          for await (const chunk of source) {
            // Convert BufferList to Buffer
            yield Buffer.isBuffer(chunk) ? chunk : chunk.slice()
          }
        })())
      } catch (err) {
        // If aborted we can safely ignore
        if (err.type !== 'aborted') {
          // If the source errored the socket will already have been destroyed by
          // toIterable.duplex(). If the socket errored it will already be
          // destroyed. There's nothing to do here except log the error & return.
          log(err)
        }
      }
    },

    source: options.signal ? abortable(source, options.signal) : source,

    conn: socket,

    localAddr: options.localAddr || toMultiaddr(socket.localAddress, socket.localPort),

    // If the remote address was passed, use it - it may have the peer ID encapsulated
    remoteAddr: options.remoteAddr || toMultiaddr(socket.remoteAddress, socket.remotePort),

    timeline: { open: Date.now() },

    close () {
      if (socket.destroyed) return

      return new Promise((resolve, reject) => {
        const start = Date.now()

        // Attempt to end the socket. If it takes longer to close than the
        // timeout, destroy it manually.
        const timeout = setTimeout(() => {
          const { host, port } = maConn.remoteAddr.toOptions()
          log('timeout closing socket to %s:%s after %dms, destroying it manually',
            host, port, Date.now() - start)

          if (socket.destroyed) {
            log('%s:%s is already destroyed', host, port)
          } else {
            socket.destroy()
          }

          resolve()
        }, CLOSE_TIMEOUT)

        socket.once('close', () => clearTimeout(timeout))
        socket.end(err => {
          maConn.timeline.close = Date.now()
          if (err) return reject(err)
          resolve()
        })
      })
    }
  }

  socket.once('close', () => {
    // In instances where `close` was not explicitly called,
    // such as an iterable stream ending, ensure we have set the close
    // timeline
    if (!maConn.timeline.close) {
      maConn.timeline.close = Date.now()
    }
  })

  return maConn
}