refactor: API changes and switch to async iterators

[vasco-santos]

BREAKING CHANGE: all the callbacks in the provided API were removed and each function uses aync/await

Needs:

• suite of tests

Closes #2 #26 #28

[vasco-santos]

@jacobheun @alanshaw @dirkmc can I have your opinion on this proposal? Meanwhile, I will be working on adding a suite of tests and a draft PR for one of the repos using this Interface

[dirkmc]

Thanks for putting this together @vasco-santos

I just made a couple of small suggestions.

[vasco-santos]

I have been thinking about the interface's current implementation and there are several things that do not make sense to me. Moreover, as we have an ongoing effort for the libp2p introspection, there is a lot of data missing in js-libp2p to make it possible.

According to the libp2p docs, a connection consists of a "bundle" of streams. However, in this interface-connection a connection is a raw connection, from where we can read and write data (it is a stream). All this naming made me really confused and after checking the interface for go-libp2p-net, I realized that we should leverage this refactor to improve the current logic, as well as to enable the libp2p introspection.

In the interface connection from the go implementation, it is provided a way to open streams and it stores the streams associated with the connection. Afterward, there is also an interface for streams, which is in reality what we were trying to achieve here.

I have discussed this with @jacobheun and we agreed that being able to implement an echo protocol as follows, would be a big improvement:

// An echo protocol
const echo = (stream) => { /* Runs the protocol over this stream */ }

// A new connection and echo
const connection = await libp2p.connect(peerInfo)
const stream = connection.newStream()
await echo(stream)

// An existing connection
const connection = await libp2p.getConnection(peerInfo)
const stream = connection.newStream()
await echo(stream)

In this context, I created a draft for the interface-connection and interface-stream that I would like feedback.

[vasco-santos]

@jacobheun @alanshaw @dirkmc

Updated this PR with my new proposal. I will work on a PR on top of libp2p/js-libp2p-tcp#102 to integrate this and see how it goes. Meanwhile, I would like some feedback

[dirkmc]

This looks like a big step in the right direction, thanks for putting this together @vasco-santos 🎉

[vasco-santos]

cc @jacobheun @dirkmc @vmx

The problem that I described during the JS Core Sync call was basically the distinction between a socket in the "connection" created and the connection of the interface-connection.

For example, in libp2p/js-libp2p-tcp#102 we have the following codebase:

class TCP {
  async dial (ma, options) {
    const cOpts = ma.toOptions()
    const rawSocket = await this._connect(cOpts, options)
    return new Libp2pSocket(rawSocket, ma, options)
  }
}

The rawSocket is a node stream, which consists of the underlying raw duplex connection between two nodes. It is not a stream within the connection, according to what we consider a stream.

My main concern was regarding the representation of this socket within our connection but luckily @jacobheun provided me a really nice approach to this.

Stating what I just added to the Readme:

Connection

Before creating a connection from a transport compatible with libp2p it is important to understand some concepts:

• socket: the underlying raw duplex connection between two nodes. It is created by the transports during a dial/listen.
• connection: the abstract libp2p duplex connection between two nodes. The transport must pipe the socket through the connection.
• stream: a single duplex channel of the connection. Each connection may have many streams.

A connection stands for the libp2p communication duplex layer between two nodes. It is not the underlying raw transport duplex layer (socket), such as a TCP socket, but an abstracted duplex layer that sits on top of the raw socket.

When a libp2p transport creates its socket, a new connection instance should be created and the transport should pipe both input and output of the socket through this connection, i.e. pipe(socket, connection, socket).

The transport must handle the translation of cleanup from the socket to the connection. That is, the errors, resets or closes on the socket must be passed to the connection. In the same way, the transport must map these actions from the connection to the socket. This helps ensuring that the transport is responsible for socket management, while also allowing the application layer to handle the connection management.

const pipe = require('it-pipe')
const { Connection } = require('interface-connection')

class Transport {
  async dial () {
    // ...

    // create the raw socket and the connection
    const socket = await this._connect()
    const conn = new Connection(remotePeerInfo, remoteMa)

    // pipe the socket through the connection (if necessary include a conversion to iterable duplex streams)
    pipe(socket, conn, socket)

    // bind the necessary handlers to update state changes (error, close, reset, ...)
    // ...

    return conn
  }

  _connect () {}
}

Let me know what you guys think

[dirkmc]

Very nice explanation, I like it a lot 👍

[vasco-santos]

@jacobheun @dirkmc

I made a few changes on the approach. Basically, if we want to go with the pipe approach, we need to have the sink and source functions in the interface-connection class, which by design does not look good to me. The interface-transport expects an iterable connection.

I changed the approach for the transports to extend the connection using their raw socket as follows:

const abortable = require('abortable-iterator')

const { Connection } = require('interface-connection')

class Libp2pSocket extends Connection {
  constructor (rawSocket, ma, opts = {}) {
    super(ma, opts)

    this._rawSocket = rawSocket

    this.sink = this._sink(opts)
    this.source = opts.signal ? abortable(rawSocket.source, opts.signal) : rawSocket.source
  }

  _sink (opts) {
    return async (source) => {
      try {
        await this._rawSocket.sink(abortable(source, opts.signal))
      } catch (err) {
        // Re-throw non-aborted errors
        if (err.type !== 'aborted') throw err
        // Otherwise, this is fine...
        await this._rawSocket.close()
      }
    }
  }
}

module.exports = Libp2pSocket

const Libp2pSocket = require('./socket')

class Transport {
  async dial () {
    // ...

    // create the raw socket and the connection
    const socket = await this._connect()
    const conn = new libp2pSocket(socket, remoteMa, {})

    return conn
  }

  _connect () {}
}

I got inspiration on the libp2p-tcp PR for async await, as it was creating this Libp2pSocket wrapper. With this approach in mind, I created 2 PRs, one for libp2p-tcp and one for libp2p-websockets with this version of the interface:

libp2p/js-libp2p-tcp#109
libp2p/js-libp2p-websockets#88

There are some tests failing. The majority is related to the adapter tests for interface-transport, which expect the older interface-connection API.

I think that this approach is cleaner and more appropriate to the transport implementation. This approach also helps ensuring that the transport is responsible for socket management, while also allowing the application layer to handle the connection management.

Let me know your thoughts

[dirkmc]

This does seem like a better approach to me 👍

[jacobheun]

I think the thing we just need to do for this to get to a final decisions is to actually exchange data over a muxed connection with this, the async tcp transport, and the mplex iterator. It will force us to make sure the mplex async version is done and we'll catch usability issues for this. i think it's going to be hard to nail down a version until that's done. That or we may make big changes to this after the fact.

[vasco-santos]

I agree! I will work on that experiment and let you know!

[jacobheun]

Some small changes otherwise I think this gets us what we need and should simplify things. I'll work on a PR for interface-stream-muxer to include the timeline, onStreamEnd and a way to get all streams.

[jacobheun]

Some minor things after the latest changes. I'm leveraging this in the latest version of the switch dial flow I am working on and everything is looking pretty good. After this round I think this should be good to merge and we can do any additional changes in followup PRs.

[jacobheun]

This looks good! Let's merge this monster!