ipfs-bitswap - npm Package Compare versions

Comparing version 0.26.2 to 0.27.0

CHANGELOG.md

@@ -0,1 +1,6 @@
+<a name="0.27.0"></a>
+# [0.27.0](https://github.com/ipfs/js-ipfs-bitswap/compare/v0.26.2...v0.27.0) (2020-01-28)
+
+
+
 <a name="0.26.2"></a>
@@ -2,0 +7,0 @@ ## [0.26.2](https://github.com/ipfs/js-ipfs-bitswap/compare/v0.26.1...v0.26.2) (2019-12-22)

package.json

 {
   "name": "ipfs-bitswap",
-  "version": "0.26.2",
+  "version": "0.27.0",
   "description": "Node.js implementation of the Bitswap data exchange protocol used by IPFS",
@@ -50,9 +50,10 @@ "leadMaintainer": "Dirk McCormick <[e-mail]>",
     "chai": "^4.2.0",
+    "delay": "^4.3.0",
     "dirty-chai": "^2.0.1",
-    "ipfs-repo": "^0.28.0",
-    "libp2p": "^0.26.1",
-    "libp2p-kad-dht": "^0.16.0",
-    "libp2p-mplex": "^0.8.0",
-    "libp2p-secio": "~0.11.1",
-    "libp2p-tcp": "^0.13.0",
+    "ipfs-repo": "^0.28.2",
+    "libp2p": "^0.27.0",
+    "libp2p-kad-dht": "^0.18.3",
+    "libp2p-mplex": "^0.9.2",
+    "libp2p-secio": "^0.12.1",
+    "libp2p-tcp": "^0.14.2",
     "lodash.difference": "^4.5.0",
@@ -63,6 +64,8 @@ "lodash.flatten": "^4.4.0",
     "ncp": "^2.0.0",
+    "p-defer": "^3.0.0",
     "p-event": "^4.1.0",
+    "p-wait-for": "^3.1.0",
     "peer-book": "~0.9.0",
-    "peer-id": "^0.12.2",
-    "peer-info": "~0.15.1",
+    "peer-id": "^0.13.5",
+    "peer-info": "^0.17.0",
     "promisify-es6": "^1.0.3",
@@ -76,13 +79,13 @@ "rimraf": "^3.0.0",
     "bignumber.js": "^9.0.0",
-    "callbackify": "^1.1.0",
     "cids": "~0.7.0",
     "debug": "^4.1.0",
     "ipfs-block": "~0.8.0",
+    "it-length-prefixed": "^3.0.0",
+    "it-pipe": "^1.1.0",
     "just-debounce-it": "^1.1.0",
     "moving-average": "^1.0.0",
-    "multicodec": "~0.5.7",
+    "multicodec": "^1.0.0",
     "multihashing-async": "^0.8.0",
     "protons": "^1.0.1",
-    "pull-length-prefixed": "^1.3.1",
-    "pull-stream": "^3.6.9",
+    "streaming-iterables": "^4.1.1",
     "varint-decoder": "~0.1.1"
@@ -89,0 +92,0 @@ },

src/network.js

 'use strict'
 
-const lp = require('pull-length-prefixed')
-const pull = require('pull-stream')
-const callbackify = require('callbackify')
+const lp = require('it-length-prefixed')
+const pipe = require('it-pipe')
 
@@ -20,18 +19,21 @@ const Message = require('./types/message')
     this.bitswap = bitswap
-    this.b100Only = options.b100Only || false
+    this.protocols = [BITSWAP100]
+    if (!options.b100Only) {
+      // Latest bitswap first
+      this.protocols.unshift(BITSWAP110)
+    }
 
     this._stats = stats
     this._running = false
-  }
 
-  start () {
-    this._running = true
     // bind event listeners
     this._onPeerConnect = this._onPeerConnect.bind(this)
     this._onPeerDisconnect = this._onPeerDisconnect.bind(this)
-
     this._onConnection = this._onConnection.bind(this)
-    this.libp2p.handle(BITSWAP100, this._onConnection)
-    if (!this.b100Only) { this.libp2p.handle(BITSWAP110, this._onConnection) }
+  }
 
+  start () {
+    this._running = true
+    this.libp2p.handle(this.protocols, this._onConnection)
+
     this.libp2p.on('peer:connect', this._onPeerConnect)
@@ -41,6 +43,7 @@ this.libp2p.on('peer:disconnect', this._onPeerDisconnect)
     // All existing connections are like new ones for us
-    this.libp2p.peerBook
-      .getAllArray()
-      .filter((peer) => peer.isConnected())
-      .forEach((peer) => this._onPeerConnect((peer)))
+    for (const peer of this.libp2p.peerStore.peers.values()) {
+      if (this.libp2p.registrar.getConnection(peer)) {
+        this._onPeerConnect(peer)
+      }
+    }
   }
@@ -51,4 +54,4 @@
 
-    this.libp2p.unhandle(BITSWAP100)
-    if (!this.b100Only) { this.libp2p.unhandle(BITSWAP110) }
+    // Unhandle both, libp2p doesn't care if it's not already handled
+    this.libp2p.unhandle(this.protocols)
 
@@ -59,32 +62,37 @@ this.libp2p.removeListener('peer:connect', this._onPeerConnect)
 
-  // Handles both types of bitswap messgages
-  _onConnection (protocol, conn) {
+  /**
+   * Handles both types of incoming bitswap messages
+   * @private
+   * @param {object} param0
+   * @param {string} param0.protocol The protocol the stream is running
+   * @param {Stream} param0.stream A duplex iterable stream
+   * @param {Connection} param0.connection A libp2p Connection
+   * @returns {void}
+   */
+  async _onConnection ({ protocol, stream, connection }) {
     if (!this._running) { return }
-    this._log('incomming new bitswap connection: %s', protocol)
+    this._log('incoming new bitswap %s connection from %s', protocol, connection.remotePeer.toB58String())
 
-    pull(
-      conn,
-      lp.decode(),
-      pull.asyncMap((data, cb) => callbackify(Message.deserialize)(data, cb)),
-      pull.asyncMap((msg, cb) => {
-        conn.getPeerInfo((err, peerInfo) => {
-          if (err) {
-            return cb(err)
+    try {
+      await pipe(
+        stream,
+        lp.decode(),
+        async (source) => {
+          for await (const data of source) {
+            try {
+              const message = await Message.deserialize(data.slice())
+              this.bitswap._receiveMessage(connection.remotePeer, message)
+            } catch (err) {
+              this.bitswap._receiveError(err)
+              break
+            }
           }
-
-          callbackify(this.bitswap._receiveMessage.bind(this.bitswap))(peerInfo.id, msg, cb)
-        })
-      }),
-      pull.onEnd((err) => {
-        this._log('ending connection')
-        if (err) {
-          this.bitswap._receiveError(err)
         }
-      })
-    )
+      )
+    } catch (err) {
+      this._log(err)
+    }
   }
 
   _onPeerConnect (peerInfo) {
-    if (!this._running) { return }
-
     this.bitswap._onPeerConnected(peerInfo.id)
@@ -94,4 +102,2 @@ }
   _onPeerDisconnect (peerInfo) {
-    if (!this._running) { return }
-
     this.bitswap._onPeerDisconnected(peerInfo.id)
@@ -124,5 +130,8 @@ }
   async findAndConnect (cid) {
-    const provs = await this.findProviders(cid, CONSTANTS.maxProvidersPerRequest)
-    this._log('connecting to providers', provs.map((p) => p.id.toB58String()))
-    await Promise.all(provs.map((p) => this.connectTo(p)))
+    const connectAttempts = []
+    for await (const provider of this.findProviders(cid, CONSTANTS.maxProvidersPerRequest)) {
+      this._log('connecting to providers', provider.id.toB58String())
+      connectAttempts.push(this.connectTo(provider))
+    }
+    await Promise.all(connectAttempts)
   }
@@ -139,6 +148,6 @@
 
-    const stringId = peer.toB58String() ? peer.toB58String() : peer.id.toB58String()
+    const stringId = peer.toB58String()
     this._log('sendMessage to %s', stringId, msg)
 
-    const { conn, protocol } = await this._dialPeer(peer)
+    const { stream, protocol } = await this._dialPeer(peer)
 
@@ -158,3 +167,3 @@ let serialized
     // Note: Don't wait for writeMessage() to complete
-    writeMessage(conn, serialized, this._log)
+    writeMessage(stream, serialized, this._log)
 
@@ -179,16 +188,4 @@ this._updateSentStats(peer, msg.blocks)
   // Dial to the peer and try to use the most recent Bitswap
-  async _dialPeer (peer) {
-    try {
-      // Attempt Bitswap 1.1.0
-      return {
-        conn: await this.libp2p.dialProtocol(peer, BITSWAP110),
-        protocol: BITSWAP110
-      }
-    } catch (err) {
-      // Attempt Bitswap 1.0.0
-      return {
-        conn: await this.libp2p.dialProtocol(peer, BITSWAP100),
-        protocol: BITSWAP100
-      }
-    }
+  _dialPeer (peer) {
+    return this.libp2p.dialProtocol(peer, [BITSWAP110, BITSWAP100])
   }
@@ -206,15 +203,14 @@
 
-function writeMessage (conn, msg, log) {
-  pull(
-    pull.values([msg]),
-    lp.encode(),
-    conn.conn,
-    pull.onEnd((err) => {
-      if (err) {
-        log(err)
-      }
-    })
-  )
+async function writeMessage (stream, msg, log) {
+  try {
+    await pipe(
+      [msg],
+      lp.encode(),
+      stream
+    )
+  } catch (err) {
+    log(err)
+  }
 }
 
 module.exports = Network

src/types/message/index.js

 'use strict'
 
-const protons = require('protons')
 const Block = require('ipfs-block')
@@ -10,3 +9,3 @@ const CID = require('cids')
 const { isMapEqual } = require('../../utils')
-const pbm = protons(require('./message.proto'))
+const { Message } = require('./message.proto')
 const Entry = require('./entry')
@@ -73,3 +72,3 @@
 
-    return pbm.Message.encode(msg)
+    return Message.encode(msg)
   }
@@ -106,3 +105,3 @@
 
-    return pbm.Message.encode(msg)
+    return Message.encode(msg)
   }
@@ -129,3 +128,3 @@
 BitswapMessage.deserialize = async (raw) => {
-  const decoded = pbm.Message.decode(raw)
+  const decoded = Message.decode(raw)
 
@@ -132,0 +131,0 @@ const isFull = (decoded.wantlist && decoded.wantlist.full) || false

src/types/message/message.proto.js

 'use strict'
-
+const protons = require('protons')
 // from: https://github.com/ipfs/go-ipfs/blob/master/exchange/bitswap/message/pb/message.proto
 
-module.exports = `
+module.exports = protons(`
   message Message {
@@ -28,2 +28,2 @@ message Wantlist {
   }
-`
+`)

No alert changes

Improved metrics

Total package byte prevSize

2296609

increased by16.46%

Lines of code

18533

increased by15.37%

Worsened metrics

Dev dependency count

29

increased by11.54%

Dependency changes

• + Addedit-length-prefixed@^3.0.0

• + Addedit-pipe@^1.1.0

• + Addedstreaming-iterables@^4.1.1

• + Added@types/bl@2.1.0(transitive)

• + Added@types/node@22.13.4(transitive)

• + Addedbl@4.1.0(transitive)

• + Addedinherits@2.0.4(transitive)

• + Addedit-length-prefixed@3.1.0(transitive)

• + Addedit-pipe@1.1.0(transitive)

• + Addedreadable-stream@3.6.2(transitive)

• + Addedstreaming-iterables@4.1.2(transitive)

• + Addedstring_decoder@1.3.0(transitive)

• + Addedundici-types@6.20.0(transitive)

• + Addedutil-deprecate@1.0.2(transitive)

• - Removedcallbackify@^1.1.0

• - Removedpull-length-prefixed@^1.3.1

• - Removedpull-stream@^3.6.9

• - Removedcallbackify@1.1.0(transitive)

• - Removedmulticodec@0.5.7(transitive)

• - Removedpull-length-prefixed@1.3.3(transitive)

• - Removedpull-pushable@2.2.0(transitive)

• - Removedpull-reader@1.3.1(transitive)

• - Removedpull-stream@3.7.0(transitive)

• Updatedmulticodec@^1.0.0