RethinkDB issue #5738

resource.js

class Resource {

  // This is a fairly straighforward REST endpoint.
  index(req, res, next) {
    r.table(this.name).run(req.conn)
    .then(function(cursor) {
      return cursor.toArray();
    })
    .then(function(docs) {
      res.status(200);
      res.json(docs);
      next();
    })
    .catch(function(err) {
      res.status(500);
      res.json({
        code: "DATABASE_ERROR",
        message: JSON.stringify(err)
      });
      next();
    });
  }

  // ...

  // This is a relatively messy proof-of-concept for transmitting changes over socket.io.
  watch({query, conn, socket, addr, subId}) {
    const logStr = JSON.stringify({addr, subId, resource: this.name, query});
    return r.table(this.name).filter(query).run(conn)
    .then((cursor) => {
      return cursor.toArray();
    })
    .then((docs) => {
      winston.info("suback", {addr, subId});
      socket.emit("suback", {subId, docs});
      return r.table(this.name).filter(query).changes().run(conn);
    })
    .then((feed) => {
      feed.on("data", function(change) {
        const newVal = change.new_val;
        const oldVal = change.old_val;
        if (oldVal === null) {
          winston.debug("created", {addr, subId});
          socket.emit("created", {subId, doc: newVal});
        }
        else if (newVal === null) {
          winston.debug("deleted", {addr, subId});
          socket.emit("deleted", {subId, id: oldVal.id});
        }
        else {
          winston.debug("updated", {addr, subId});
          socket.emit("updated", {subId, doc: newVal});
        }
      });
      feed.on("error", function(...args) {
        winston.error("error on resource watch", ...args);
      });
      socket.on("unsub", function(params) {
        if (params.subId === subId) {
          winston.debug("unsub", {subId, addr});
          feed.removeAllListeners("data");
          feed.removeAllListeners("error");
          feed.close();
        }
      });
      socket.on("disconnect", function() {
        feed.removeAllListeners("data");
        feed.removeAllListeners("error");
        feed.close();
      });
    })
    .catch((err) => {
      winston.error("Error in watch function", err);
    });
  }
}