feat: onPut and onRemove events (#205)

Author: HexaField

src/index.js

@@ -54,6 +54,8 @@ class KadDHT extends EventEmitter {
    * @param {object} props.validators validators object with namespace as keys and function(key, record, callback)
    * @param {object} props.selectors selectors object with namespace as keys and function(key, records)
    * @param {randomWalkOptions} options.randomWalk randomWalk options
+   * @param {function(record: Record, peerId: PeerId)} [props.onPut = () => {}] Called when an entry is added to or changed in the datastore
+   * @param {function(record: Record)} [props.onRemove = () => {}] Called when an entry is removed from the datastore
    */
   constructor ({
     libp2p,
@@ -67,7 +69,9 @@ class KadDHT extends EventEmitter {
     concurrency = c.ALPHA,
     validators = {},
     selectors = {},
-    randomWalk = {}
+    randomWalk = {},
+    onPut = () => {},
+    onRemove = () => {}
   }) {
     super()
 
@@ -182,6 +186,10 @@ class KadDHT extends EventEmitter {
     this.contentFetching = contentFetching(this)
     this.contentRouting = contentRouting(this)
     this.peerRouting = peerRouting(this)
+
+    // datastore events
+    this.onPut = onPut
+    this.onRemove = onRemove
   }
 
   /**
@@ -411,6 +419,7 @@ class KadDHT extends EventEmitter {
       utils.now() - record.timeReceived > c.MAX_RECORD_AGE) {
       // If record is bad delete it and return
       await this.datastore.delete(dsKey)
+      this.onRemove(record)
       return undefined
     }
 

src/rpc/handlers/put-value.js

@@ -32,6 +32,8 @@ module.exports = (dht) => {
     const recordKey = utils.bufferToKey(record.key)
     await dht.datastore.put(recordKey, record.serialize())
 
+    dht.onPut(record, peerId)
+
     return msg
   }
 }

test/kad-dht.spec.js

@@ -837,9 +837,15 @@ describe('KadDHT', () => {
         const lookup = await dht.datastore.get(kadUtils.bufferToKey(record.key))
         expect(lookup).to.exist('Record should be in the local datastore')
 
+        let eventResponse
+        dht.onRemove = (record) => {
+          eventResponse = { record }
+        }
+
         const rec = await dht._checkLocalDatastore(record.key)
         expect(rec).to.not.exist('Record should have expired')
 
+        expect(eventResponse).to.have.property('record').eql(record)
         // TODO
         // const lookup2 = await dht.datastore.get(kadUtils.bufferToKey(record.key))
         // expect(lookup2).to.not.exist('Record should be removed from datastore')

test/rpc/handlers/put-value.spec.js

@@ -56,9 +56,17 @@ describe('rpc - handlers - PutValue', () => {
     )
     msg.record = record
 
+    let eventResponse
+    dht.onPut = (record, peerId) => {
+      eventResponse = { record, peerId }
+    }
+
     const response = await handler(dht)(peerIds[1], msg)
     expect(response).to.be.eql(msg)
 
+    expect(eventResponse).to.have.property('record').eql(record)
+    expect(eventResponse).to.have.property('peerId').eql(peerIds[1])
+
     const key = utils.bufferToKey(uint8ArrayFromString('hello'))
     const res = await dht.datastore.get(key)