Last minute fixes.

Author: mcollina

lib/commands.js

@@ -28,14 +28,21 @@ function load (kv) {
     process.nextTick(function () {
       tracker.on('replica', sendData)
     })
-    tracker.on('moved', function () {
+    tracker.on('move', function (peer) {
+      if (peer) {
+        sendData(peer)
+      }
       entry.hasTracker = false
       const rs = streams.get(key) || []
-      rs.forEach((stream) => stream.end())
+      rs.forEach((stream) => {
+        stream.push(null)
+      })
       streams.delete(key)
-      if (!entry.hasReplicator) {
-        db.delete(key)
-      }
+      setTimeout(function () {
+        if (!entry.hasReplicator) {
+          db.delete(key)
+        }
+      }, 30000).unref()
     })
 
     if (dest) {
@@ -67,6 +74,7 @@ function load (kv) {
 
   function put (req, reply) {
     const key = req.key
+    const sendData = genSendData(key)
     var needReply = true
     var entry = db.get(key) || new Entry(key)
 
@@ -94,18 +102,36 @@ function load (kv) {
     if (array) {
       array.forEach((stream) => stream.push(req.value))
     }
+  }
+
+  function genSendData (key) {
+    return sendData
 
     function sendData (peer, cb) {
       if (typeof cb !== 'function') {
         cb = bigError
       }
+
+      const entry = db.get(key)
+      if (!entry) {
+        cb()
+        return
+      }
+
+      const req = {
+        ns: 'kv',
+        cmd: 'put',
+        key: entry.key,
+        value: entry.value
+      }
+
       upring.peerConn(peer).request(req, function (err) {
         if (err) {
           cb(err)
           return
         }
 
-        logger.info({ key }, 'replicated key')
+        logger.debug({ key, value: entry.value }, 'replicated key')
 
         cb()
       })

package.json

@@ -44,7 +44,7 @@
     "pino-http": "^1.3.0",
     "pump": "^1.0.1",
     "readable-stream": "^2.1.5",
-    "upring": "^0.14.0",
+    "upring": "^0.14.1",
     "url-format-lax": "^1.0.0"
   }
 }