Watch downloads (#2)

* Watch downloads

* Use event listener changes for (un)watching downloads.

* Remove unused argument

* Use noReply for (un)watching downloads

README.md

@@ -221,6 +221,10 @@ Emitted when a peer is removed.
 
 Emitted when the feed is appended to, either locally or remotely.
 
+#### `feed.on('download', seq)`
+
+Emitted when a block is downloaded.
+
 # License
 
 MIT

index.js

@@ -73,6 +73,11 @@ class RemoteCorestore extends EventEmitter {
         const remoteCore = this._sessions.get(id)
         if (!remoteCore) throw new Error('Invalid RemoteHypercore ID.')
         remoteCore._onwait(onWaitId, seq)
+      },
+      onDownload ({ id, seq }) {
+        const remoteCore = this._sessions.get(id)
+        if (!remoteCore) throw new Error('Invalid RemoteHypercore ID.')
+        remoteCore._ondownload({ seq })
       }
     })
     this._client.corestore.onRequest(this, {
@@ -359,6 +364,18 @@ class RemoteHypercore extends Nanoresource {
     this._extensions = new Map()
     this._onwaits = new FreeMap(1)
 
+    // Track listeners for the download event, and enable/disable download watching.
+    this.on('newListener', (event) => {
+      if (event === 'download' && !this.listenerCount(event)) {
+        this._watchDownloads()
+      }
+    })
+    this.on('removeListener', (event) => {
+      if (event === 'download' && !this.listenerCount(event)) {
+        this._unwatchDownloads()
+      }
+    })
+
     if (!this.lazy) this.ready(() => {})
   }
 
@@ -447,6 +464,11 @@ class RemoteHypercore extends Nanoresource {
     ext.onmessage(data, remotePeer)
   }
 
+  _ondownload (rsp) {
+    // TODO: Add to local bitfield?
+    this.emit('download', rsp.seq)
+  }
+
   // Private Methods
 
   _indexOfPeer (remotePublicKey) {
@@ -569,6 +591,22 @@ class RemoteHypercore extends Nanoresource {
     return rsp.bytes
   }
 
+  async _watchDownloads () {
+    try {
+      if (!this.opened) await this.open()
+      if (this.closed) return
+      this._client.hypercore.watchDownloadsNoReply({ id: this._id })
+    } catch (_) {}
+  }
+
+  async _unwatchDownloads () {
+    try {
+      if (!this.opened) await this.open()
+      if (this.closed) return
+      this._client.hypercore.unwatchDownloadsNoReply({ id: this._id })
+    } catch (_) {}
+  }
+
   // Public Methods
 
   append (blocks, cb) {