Expose inflightRange opt (#493)

* Add inflightRange opt

* Fix: correct inflightRange-name in readme

* Simplify

* Pass inflightRange as arg to new Peer()

Author: HDegroote

README.md

@@ -74,7 +74,8 @@ Note that `tree`, `data`, and `bitfield` are normally heavily sparse files.
   encryptionKey: k, // optionally pass an encryption key to enable block encryption
   onwait: () => {}, // hook that is called if gets are waiting for download
   timeout: 0, // wait at max some milliseconds (0 means no timeout)
-  writable: true // disable appends and truncates
+  writable: true, // disable appends and truncates
+  inflightRange: null // Advanced option. Set to [minInflight, maxInflight] to change the min and max inflight blocks per peer when downloading.
 }
 ```
 

index.js

@@ -60,6 +60,7 @@ module.exports = class Hypercore extends EventEmitter {
     this.crypto = opts.crypto || hypercoreCrypto
     this.core = null
     this.replicator = null
+    this.inflightRange = opts.inflightRange || null
     this.encryption = null
     this.extensions = new Map()
     this.cache = createCache(opts.cache)
@@ -404,6 +405,7 @@ module.exports = class Hypercore extends EventEmitter {
       eagerUpgrade: true,
       notDownloadingLinger: opts.notDownloadingLinger,
       allowFork: opts.allowFork !== false,
+      inflightRange: this.inflightRange,
       onpeerupdate: this._onpeerupdate.bind(this),
       onupload: this._onupload.bind(this),
       oninvalid: this._oninvalid.bind(this)

lib/replicator.js

@@ -268,14 +268,15 @@ class BlockTracker {
 }
 
 class Peer {
-  constructor (replicator, protomux, channel, session) {
+  constructor (replicator, protomux, channel, session, inflightRange) {
     this.tracer = createTracer(this, { parent: replicator.core.tracer })
     this.core = replicator.core
     this.replicator = replicator
     this.stream = protomux.stream
     this.protomux = protomux
     this.remotePublicKey = this.stream.remotePublicKey
     this.remoteSupportsSeeks = false
+    this.inflightRange = inflightRange
 
     this.paused = false
 
@@ -299,7 +300,6 @@ class Peer {
     this.receiverBusy = false
 
     this.inflight = 0
-    this.inflightRange = DEFAULT_MAX_INFLIGHT
     this.dataProcessing = 0
 
     this.canUpgrade = true
@@ -1221,6 +1221,7 @@ module.exports = class Replicator {
     notDownloadingLinger = NOT_DOWNLOADING_SLACK,
     eagerUpgrade = true,
     allowFork = true,
+    inflightRange = null,
     onpeerupdate = noop,
     onupload = noop,
     oninvalid = noop
@@ -1241,6 +1242,8 @@ module.exports = class Replicator {
     this.downloading = false
     this.activeSessions = 0
 
+    this.inflightRange = inflightRange || DEFAULT_MAX_INFLIGHT
+
     this._attached = new Set()
     this._inflight = new InflightTracker()
     this._blocks = new BlockTracker()
@@ -2090,7 +2093,7 @@ module.exports = class Replicator {
 
     if (channel === null) return onnochannel()
 
-    const peer = new Peer(replicator, protomux, channel, session)
+    const peer = new Peer(replicator, protomux, channel, session, this.inflightRange)
     const stream = protomux.stream
 
     peer.channel.open({

test/replicate.js

@@ -1486,6 +1486,26 @@ test('replication updates on core copy', async function (t) {
   await t.execution(promise)
 })
 
+test('can define default max-inflight blocks for replicator peers', async function (t) {
+  const a = new Hypercore(RAM, { inflightRange: [123, 123] })
+  await a.append('some block')
+
+  const b = await create(a.key)
+  replicate(a, b, t)
+  await b.get(0)
+
+  t.alike(
+    a.replicator.peers[0].inflightRange,
+    [123, 123],
+    'Uses the custom inflight range'
+  )
+  t.alike(
+    b.replicator.peers[0].inflightRange,
+    [32, 512],
+    'Uses default if no inflight range specified'
+  )
+})
+
 async function waitForRequestBlock (core) {
   while (true) {
     const reqBlock = core.replicator._inflight._requests.find(req => req && req.block)