feat: handling max active extension messages

lib/extension.js

@@ -1,6 +1,6 @@
 const { Extension } = require('./messages')
 
-// const MAX_ACTIVE = 32
+const MAX_ACTIVE = 32
 const FLUSH_BATCH = 128
 const MAX_PASSIVE_BATCH = 2048
 const MAX_ACTIVE_BATCH = MAX_PASSIVE_BATCH + FLUSH_BATCH
@@ -53,15 +53,33 @@ class HyperbeeExtension {
     if (snapshot.limit === -1) snapshot.limit = 0
     this.outgoing.broadcast(Extension.encode({ iterator: snapshot }))
   }
+
+  // can decide whether to expose this or not, can pass as callback function also
+  finishActive(){
+    this.active--
+    if (this.active < 0) {
+      this.active = 0
+    }
+  }
 
   onmessage (buf, from) {
     // TODO: handle max active extension messages
-    // this.active++
+    // drop or ignore if too many active
+    if (this.active >= MAX_ACTIVE) {
+      return;
+    }
+    this.active++
 
     const message = decode(buf)
-    if (!message) return
+    if (!message){
+      this.active--;
+      return;
+    }
 
-    if (message.cache) this.oncache(message.cache, from)
+    if (message.cache){
+       this.oncache(message.cache, from)
+       this.finishActive();
+    }
     if (message.get) this.onget(message.get, from)
     if (message.iterator) this.oniterator(message.iterator, from)
   }
@@ -72,16 +90,21 @@ class HyperbeeExtension {
   }
 
   onget (message, from) {
-    if (!message.version || message.version > this.db.version) return
+    if (!message.version || message.version > this.db.version){
+      this.finishActive();
+      return
+    }
 
     const b = new Batch(this.outgoing, from)
     const db = this.db.checkout(message.version)
+    const self = this
 
     db.get(message.key, { extension: false, wait: false, update: false, onseq }).then(done, done)
 
     function done () {
       db.close().catch(noop)
       b.send()
+      self.finishActive();
     }
 
     function onseq (seq) {
@@ -90,7 +113,10 @@ class HyperbeeExtension {
   }
 
   async oniterator (message, from) {
-    if (!message.version || message.version > this.db.version) return
+    if (!message.version || message.version > this.db.version) {
+      this.finishActive();
+      return
+    }
 
     const b = new Batch(this.outgoing, from)
     const seqs = new Set()
@@ -126,6 +152,7 @@ class HyperbeeExtension {
       ite.close().catch(noop)
       db.close().catch(noop)
       b.send()
+      this.finishActive();
     }
   }