Implement task queue and notifications

server/api-functions.js

@@ -8,6 +8,7 @@ const querystring = require("querystring");
 const config = require("./config");
 const device = require("./device");
 const filterParser = require("../common/filter-parser");
+const Filter = require("../common/filter");
 
 const commaAscii = ",".charCodeAt(0);
 const newLineAscii = "\n".charCodeAt(0);
@@ -205,5 +206,130 @@ function query(resource, filter, limit, skip, projection, callback) {
   });
 }
 
+function deleteResource(resource, id) {
+  return new Promise((resolve, reject) => {
+    let options = url.parse(
+      `${config.get("server.nbi")}${resource}/${encodeURIComponent(id)}`
+    );
+
+    options.method = "DELETE";
+
+    let _http = options.protocol === "https:" ? https : http;
+
+    _http
+      .request(options, res => {
+        res.resume();
+        if (res.statusCode === 200) resolve(true);
+        else if (res.statusCode === 404) resolve(false);
+        else reject(new Error(`Unexpected status code ${res.statusCode}`));
+      })
+      .end();
+  });
+}
+
+function postTask(task, connectionRequest, callback) {
+  let options = url.parse(
+    `${config.get("server.nbi")}devices/${task.device}/tasks${
+      connectionRequest ? "?connection_request" : ""
+    }`
+  );
+  options.method = "POST";
+  let _http = options.protocol === "https:" ? https : http;
+
+  _http
+    .request(options, res => {
+      if (res.statusCode !== 200 && res.statusCode !== 202) {
+        callback(new Error(`Unexpected status code ${res.statusCode}`));
+        res.resume();
+        return;
+      }
+      const chunks = [];
+      let bytes = 0;
+      res.on("data", chunk => {
+        chunks.push(chunk);
+        bytes += chunk.length;
+      });
+
+      res.on("end", () => {
+        let buf = new Buffer(bytes);
+        let o = 0;
+        for (let c of chunks) {
+          c.copy(buf, o, 0, c.length);
+          o += c.length;
+        }
+        try {
+          let t = JSON.parse(buf);
+          let connectionRequestStatus;
+          if (connectionRequest) {
+            connectionRequestStatus = "OK";
+            if (
+              res.statusCode === 202 &&
+              res.statusMessage !== "Task queued but not processed" &&
+              res.statusMessage !== "Task faulte"
+            )
+              connectionRequestStatus = res.statusMessage;
+          }
+          callback(null, t, connectionRequestStatus);
+        } catch (err) {
+          callback(err);
+        }
+      });
+    })
+    .end(JSON.stringify(task));
+}
+
+function postTasks(deviceId, tasks) {
+  return new Promise((resolve, reject) => {
+    let connectionRequestStatus;
+    let promises = [];
+    for (let [idx, task] of tasks.entries()) {
+      task.device = deviceId;
+      promises.push(
+        new Promise((res, rej) => {
+          let conReq = idx === tasks.length - 1;
+          delete task._id;
+          task.expiry = 5;
+          postTask(task, conReq, (err, t, crs) => {
+            if (err) return rej(err);
+            if (conReq) connectionRequestStatus = crs;
+            res({ _id: t._id, status: "pending" });
+          });
+        })
+      );
+    }
+
+    Promise.all(promises)
+      .then(statuses => {
+        if (connectionRequestStatus !== "OK")
+          return resolve({
+            connectionRequest: connectionRequestStatus,
+            tasks: statuses
+          });
+        Promise.all(
+          statuses.map(s => query("tasks", new Filter(["=", "_id", s._id])))
+        )
+          .then(res => {
+            for (let [i, r] of res.entries()) {
+              if (!r[0]) {
+                statuses[i].status = "done";
+              } else if (r[0].fault) {
+                statuses[i].status = "fault";
+                statuses[i].fault = r[0].fault;
+              }
+              deleteResource("tasks", statuses[i]._id);
+            }
+
+            resolve({
+              connectionRequest: connectionRequestStatus,
+              tasks: statuses
+            });
+          })
+          .catch(reject);
+      })
+      .catch(reject);
+  });
+}
+
 exports.query = query;
 exports.count = count;
+exports.postTasks = postTasks;

server/api.js

@@ -12,7 +12,8 @@ const resources = {
   presets: true,
   provisions: true,
   files: true,
-  virtual_parameters: true
+  virtual_parameters: true,
+  tasks: false
 };
 
 for (let [resource, update] of Object.entries(resources)) {
@@ -60,4 +61,11 @@ for (let [resource, update] of Object.entries(resources)) {
   }
 }
 
+router.post("/devices/:id/tasks", async ctx => {
+  let res = await apiFunctions.postTasks(ctx.params.id, ctx.request.body);
+  // TODO 404 if no such device
+  ctx.set("Connection-Request", res.connectionRequest);
+  ctx.body = res.tasks;
+});
+
 module.exports = router;

server/device.js

@@ -114,12 +114,12 @@ function recursive(input, root, output, timestamp) {
       attrs["valueTimestamp"] = childrenTimestamp;
       attrs["object"] = false;
       attrs["objectTimestamp"] = childrenTimestamp;
-    } else if (tree["_object"]) {
+    } else if (tree["_object"] != null) {
       attrs["object"] = tree["_object"];
       attrs["objectTimestamp"] = childrenTimestamp;
     }
 
-    if (tree["_writable"]) {
+    if (tree["_writable"] != null) {
       attrs["writable"] = tree["_writable"];
       attrs["writableTimestamp"] = childrenTimestamp;
     }

server/server.js

@@ -3,6 +3,8 @@
 const Koa = require("koa");
 const Router = require("koa-router");
 const koaStatic = require("koa-static");
+const koaCompress = require("koa-compress");
+const koaBody = require("koa-body");
 
 const config = require("./config");
 const api = require("./api");
@@ -10,6 +12,7 @@ const api = require("./api");
 const koa = new Koa();
 const router = new Router();
 
+koa.use(koaBody());
 router.use("/api", api.routes(), api.allowedMethods());
 
 router.get("/", async ctx => {
@@ -30,6 +33,11 @@ router.get("/", async ctx => {
   `;
 });
 
+koa.use(
+  koaCompress({
+    flush: require("zlib").Z_SYNC_FLUSH
+  })
+);
 koa.use(router.routes());
 koa.use(koaStatic("./public"));