Update MongoDB connector

lib/db.js

@@ -33,85 +33,39 @@ let tasksCollection,
   operationsCollection;
 
 function connect(callback) {
-  let callbackCounter = 9;
   mongodb.MongoClient.connect(
     config.get("MONGODB_CONNECTION_URL"),
-    (err, db) => {
+    { useNewUrlParser: true },
+    (err, client) => {
       if (err) return void callback(err);
+      exports.client = client;
+      const db = client.db();
 
-      exports.mongoDb = db;
-      db.collection("tasks", (err, collection) => {
-        exports.tasksCollection = tasksCollection = collection;
-        if (collection != null)
-          collection.ensureIndex({ device: 1, timestamp: 1 });
+      exports.tasksCollection = tasksCollection = db.collection("tasks");
+      tasksCollection.createIndex({ device: 1, timestamp: 1 });
 
-        if (--callbackCounter === 0 || err) {
-          callbackCounter = 0;
-          callback(err);
-        }
-      });
-      db.collection("devices", (err, collection) => {
-        exports.devicesCollection = devicesCollection = collection;
-        if (--callbackCounter === 0 || err) {
-          callbackCounter = 0;
-          callback(err);
-        }
-      });
-      db.collection("presets", (err, collection) => {
-        exports.presetsCollection = presetsCollection = collection;
-        if (--callbackCounter === 0 || err) {
-          callbackCounter = 0;
-          callback(err);
-        }
-      });
-      db.collection("objects", (err, collection) => {
-        exports.objectsCollection = objectsCollection = collection;
-        if (--callbackCounter === 0 || err) {
-          callbackCounter = 0;
-          callback(err);
-        }
-      });
-      db.collection("fs.files", (err, collection) => {
-        exports.filesCollection = filesCollection = collection;
-        if (--callbackCounter === 0 || err) {
-          callbackCounter = 0;
-          callback(err);
-        }
-      });
-      db.collection("provisions", (err, collection) => {
-        exports.provisionsCollection = provisionsCollection = collection;
-        if (--callbackCounter === 0 || err) {
-          callbackCounter = 0;
-          callback(err);
-        }
-      });
-      db.collection("virtualParameters", (err, collection) => {
-        exports.virtualParametersCollection = virtualParametersCollection = collection;
-        if (--callbackCounter === 0 || err) {
-          callbackCounter = 0;
-          callback(err);
-        }
-      });
-      db.collection("faults", (err, collection) => {
-        exports.faultsCollection = faultsCollection = collection;
-        if (--callbackCounter === 0 || err) {
-          callbackCounter = 0;
-          callback(err);
-        }
-      });
-      db.collection("operations", (err, collection) => {
-        exports.operationsCollection = operationsCollection = collection;
-        if (--callbackCounter === 0 || err) {
-          callbackCounter = 0;
-          callback(err);
-        }
-      });
+      exports.devicesCollection = devicesCollection = db.collection("devices");
+      exports.presetsCollection = presetsCollection = db.collection("presets");
+      exports.objectsCollection = objectsCollection = db.collection("objects");
+      exports.filesCollection = filesCollection = db.collection("fs.files");
+      exports.provisionsCollection = provisionsCollection = db.collection(
+        "provisions"
+      );
+      exports.virtualParametersCollection = virtualParametersCollection = db.collection(
+        "virtualParameters"
+      );
+      exports.faultsCollection = faultsCollection = db.collection("faults");
+      exports.operationsCollection = operationsCollection = db.collection(
+        "operations"
+      );
+
+      callback();
     }
   );
 }
 
 function disconnect() {
-  if (exports.mongoDb) exports.mongoDb.close();
+  if (exports.client) exports.client.close();
 }
 
 // Optimize projection by removing overlaps
@@ -279,12 +233,13 @@ function fetchDevice(id, timestamp, patterns, callback) {
     }
   }
 
-  let proj;
-  if (projection[""]) proj = {};
-  else if (!Object.keys(projection).length) proj = { _id: 1 };
-  else proj = projection;
+  const options = {};
+  if (!projection[""]) {
+    if (!Object.keys(projection).length) options.projection = { _id: 1 };
+    else options.projection = projection;
+  }
 
-  devicesCollection.findOne({ _id: id }, proj, (err, device) => {
+  devicesCollection.findOne({ _id: id }, options, (err, device) => {
     if (err || !device) return void callback(err);
 
     function storeParams(obj, path, ts, descendantsFetched, projTree) {
@@ -358,11 +313,13 @@ function fetchDevice(id, timestamp, patterns, callback) {
       }
 
       if (!descendantsFetched) {
-        for (const k of Object.keys(projTree)) {
-          const p = path.concat(k);
-          loaded.push([p, ((1 << path.length) - 1) ^ ((1 << MAX_DEPTH) - 1)]);
-          if ((obj["_object"] || !path.length) && obj["_timestamp"])
-            res.push([p, obj["_timestamp"]]);
+        if (projTree) {
+          for (const k of Object.keys(projTree)) {
+            const p = path.concat(k);
+            loaded.push([p, ((1 << path.length) - 1) ^ ((1 << MAX_DEPTH) - 1)]);
+            if ((obj["_object"] || !path.length) && obj["_timestamp"])
+              res.push([p, obj["_timestamp"]]);
+          }
         }
       } else if ((obj["_object"] || !path.length) && obj["_timestamp"]) {
         res.push([path.concat("*"), obj["_timestamp"]]);
@@ -740,26 +697,26 @@ function saveDevice(deviceId, deviceData, isNew, sessionTimestamp, callback) {
     delete update["$pull"];
   }
 
-  devicesCollection.update(
+  devicesCollection.updateOne(
     { _id: deviceId },
     update,
     { upsert: isNew },
     (err, result) => {
-      if (!err && result.result.n !== 1) {
+      if (err || result.result.n !== 1) {
         return void callback(
-          new Error(`Device ${deviceId} not found in database`)
+          err || new Error(`Device ${deviceId} not found in database`)
         );
       }
 
       if (update2) {
-        return void devicesCollection.update(
+        return void devicesCollection.updateOne(
           { _id: deviceId },
           update2,
           callback
         );
       }
 
-      callback(err);
+      callback();
     }
   );
 }
@@ -786,17 +743,18 @@ function getFaults(deviceId, callback) {
 }
 
 function saveFault(deviceId, channel, fault, callback) {
+  const id = `${deviceId}:${channel}`;
   fault = Object.assign({}, fault);
-  fault._id = `${deviceId}:${channel}`;
+  fault._id = id;
   fault.device = deviceId;
   fault.channel = channel;
   fault.timestamp = new Date(fault.timestamp);
   fault.provisions = JSON.stringify(fault.provisions);
-  faultsCollection.save(fault, callback);
+  faultsCollection.replaceOne({ _id: id }, fault, { upsert: true }, callback);
 }
 
 function deleteFault(deviceId, channel, callback) {
-  faultsCollection.remove({ _id: `${deviceId}:${channel}` }, callback);
+  faultsCollection.deleteOne({ _id: `${deviceId}:${channel}` }, callback);
 }
 
 function getDueTasks(deviceId, timestamp, callback) {
@@ -804,7 +762,7 @@ function getDueTasks(deviceId, timestamp, callback) {
   const tasks = [];
 
   let f;
-  cur.nextObject(
+  cur.next(
     (f = (err, task) => {
       if (err) return void callback(err);
 
@@ -835,17 +793,17 @@ function getDueTasks(deviceId, timestamp, callback) {
               task.fileName = res[0].filename || res[0]._id.toString();
           }
 
-          cur.nextObject(f);
+          cur.next(f);
         });
       } else {
-        cur.nextObject(f);
+        cur.next(f);
       }
     })
   );
 }
 
 function clearTasks(deviceId, taskIds, callback) {
-  tasksCollection.remove(
+  tasksCollection.deleteMany(
     { _id: { $in: taskIds.map(id => new mongodb.ObjectID(id)) } },
     callback
   );
@@ -873,17 +831,26 @@ function getOperations(deviceId, callback) {
 }
 
 function saveOperation(deviceId, commandKey, operation, callback) {
+  const id = `${deviceId}:${commandKey}`;
   operation = Object.assign({}, operation);
-  operation._id = `${deviceId}:${commandKey}`;
+  operation._id = id;
   operation.timestamp = new Date(operation.timestamp);
   operation.provisions = JSON.stringify(operation.provisions);
   operation.retries = JSON.stringify(operation.retries);
   operation.args = JSON.stringify(operation.args);
-  operationsCollection.save(operation, callback);
+  operationsCollection.replaceOne(
+    { _id: id },
+    operation,
+    { upsert: true },
+    callback
+  );
 }
 
 function deleteOperation(deviceId, commandKey, callback) {
-  operationsCollection.remove({ _id: `${deviceId}:${commandKey}` }, callback);
+  operationsCollection.deleteOne(
+    { _id: `${deviceId}:${commandKey}` },
+    callback
+  );
 }
 
 function getPresets(callback) {