diff --git a/bin/genieacs-cwmp.ts b/bin/genieacs-cwmp.ts
index a271d77..6e01d95 100755
--- a/bin/genieacs-cwmp.ts
+++ b/bin/genieacs-cwmp.ts
@@ -24,7 +24,6 @@ import * as server from "../lib/server";
import * as cwmp from "../lib/cwmp";
import * as db from "../lib/db";
import * as extensions from "../lib/extensions";
-import * as cache from "../lib/cache";
import { version as VERSION } from "../package.json";
logger.init("cwmp", VERSION);
@@ -36,7 +35,6 @@ function exitWorkerGracefully(): void {
setTimeout(exitWorkerUngracefully, 5000).unref();
Promise.all([
db.disconnect(),
- cache.disconnect(),
extensions.killAll(),
cluster.worker.disconnect(),
]).catch(exitWorkerUngracefully);
@@ -99,7 +97,8 @@ if (!cluster.worker) {
server.stop().then(exitWorkerGracefully).catch(exitWorkerUngracefully);
});
- const initPromise = Promise.all([db.connect(), cache.connect()])
+ const initPromise = db
+ .connect()
.then(() => {
server.start(options, cwmp.listener);
})
diff --git a/bin/genieacs-fs.ts b/bin/genieacs-fs.ts
index b6d3146..c794fa6 100755
--- a/bin/genieacs-fs.ts
+++ b/bin/genieacs-fs.ts
@@ -23,7 +23,6 @@ import * as cluster from "../lib/cluster";
import * as server from "../lib/server";
import { listener } from "../lib/fs";
import * as db from "../lib/db";
-import * as cache from "../lib/cache";
import { version as VERSION } from "../package.json";
logger.init("fs", VERSION);
@@ -33,11 +32,9 @@ const SERVICE_PORT = config.get("FS_PORT") as number;
function exitWorkerGracefully(): void {
setTimeout(exitWorkerUngracefully, 5000).unref();
- Promise.all([
- db.disconnect(),
- cache.disconnect(),
- cluster.worker.disconnect(),
- ]).catch(exitWorkerUngracefully);
+ Promise.all([db.disconnect(), cluster.worker.disconnect()]).catch(
+ exitWorkerUngracefully
+ );
}
function exitWorkerUngracefully(): void {
@@ -100,7 +97,8 @@ if (!cluster.worker) {
listener(req, res);
};
- const initPromise = Promise.all([db.connect(), cache.connect()])
+ const initPromise = db
+ .connect()
.then(() => {
server.start(options, _listener);
})
diff --git a/bin/genieacs-nbi.ts b/bin/genieacs-nbi.ts
index 4b695bf..623dd50 100755
--- a/bin/genieacs-nbi.ts
+++ b/bin/genieacs-nbi.ts
@@ -24,7 +24,6 @@ import * as server from "../lib/server";
import { listener } from "../lib/nbi";
import * as db from "../lib/db";
import * as extensions from "../lib/extensions";
-import * as cache from "../lib/cache";
import { version as VERSION } from "../package.json";
logger.init("nbi", VERSION);
@@ -36,7 +35,6 @@ function exitWorkerGracefully(): void {
setTimeout(exitWorkerUngracefully, 5000).unref();
Promise.all([
db.disconnect(),
- cache.disconnect(),
extensions.killAll(),
cluster.worker.disconnect(),
]).catch(exitWorkerUngracefully);
@@ -104,7 +102,8 @@ if (!cluster.worker) {
listener(req, res);
};
- const initPromise = Promise.all([db.connect(), cache.connect()])
+ const initPromise = db
+ .connect()
.then(() => {
server.start(options, _listener);
})
diff --git a/bin/genieacs-ui.ts b/bin/genieacs-ui.ts
index 57c9b27..2ab4f7d 100755
--- a/bin/genieacs-ui.ts
+++ b/bin/genieacs-ui.ts
@@ -23,9 +23,7 @@ import * as cluster from "../lib/cluster";
import * as server from "../lib/server";
import { listener } from "../lib/ui";
import * as extensions from "../lib/extensions";
-import * as db from "../lib/ui/db";
-import * as db2 from "../lib/db";
-import * as cache from "../lib/cache";
+import * as db from "../lib/db";
import { version as VERSION } from "../package.json";
logger.init("ui", VERSION);
@@ -37,8 +35,6 @@ function exitWorkerGracefully(): void {
setTimeout(exitWorkerUngracefully, 5000).unref();
Promise.all([
db.disconnect(),
- db2.disconnect(),
- cache.disconnect(),
extensions.killAll(),
cluster.worker.disconnect(),
]).catch(exitWorkerUngracefully);
@@ -106,7 +102,8 @@ if (!cluster.worker) {
listener(req, res);
};
- const initPromise = Promise.all([db2.connect(), cache.connect()])
+ const initPromise = db
+ .connect()
.then(() => {
server.start(options, _listener);
})
diff --git a/lib/cache.ts b/lib/cache.ts
index 0973ab5..15ba41f 100644
--- a/lib/cache.ts
+++ b/lib/cache.ts
@@ -17,32 +17,22 @@
* along with GenieACS. If not, see .
*/
-import { MongoClient, Collection } from "mongodb";
+import { Collection } from "mongodb";
+import { onConnect } from "./db";
import * as config from "./config";
const MAX_CACHE_TTL = +config.get("MAX_CACHE_TTL");
-let clientPromise: Promise;
let mongoCollection: Collection;
let mongoTimeOffset = 0;
-export async function connect(): Promise {
- const MONGODB_CONNECTION_URL = "" + config.get("MONGODB_CONNECTION_URL");
- clientPromise = MongoClient.connect(MONGODB_CONNECTION_URL, {
- useNewUrlParser: true,
- useUnifiedTopology: true,
- });
- const db = (await clientPromise).db();
+onConnect(async (db) => {
mongoCollection = db.collection("cache");
await mongoCollection.createIndex({ expire: 1 }, { expireAfterSeconds: 0 });
const now = Date.now();
const res = await db.command({ hostInfo: 1 });
mongoTimeOffset = res.system.currentTime.getTime() - now;
-}
-
-export async function disconnect(): Promise {
- if (clientPromise) await (await clientPromise).close();
-}
+});
export async function get(key: string | string[]): Promise {
const expire = new Date(Date.now() - mongoTimeOffset);
diff --git a/lib/db.ts b/lib/db.ts
index c15607b..c87b9cf 100644
--- a/lib/db.ts
+++ b/lib/db.ts
@@ -17,7 +17,7 @@
* along with GenieACS. If not, see .
*/
-import { MongoClient, ObjectID, Collection } from "mongodb";
+import { MongoClient, ObjectID, Collection, Db } from "mongodb";
import { get } from "./config";
import { encodeTag, escapeRegExp } from "./common";
import { parse } from "./common/expression-parser";
@@ -45,7 +45,6 @@ export let tasksCollection: Collection,
configCollection: Collection;
let clientPromise: Promise;
-export let client: MongoClient;
function compareAccessLists(list1: string[], list2: string[]): boolean {
if (list1.length !== list2.length) return false;
@@ -53,15 +52,24 @@ function compareAccessLists(list1: string[], list2: string[]): boolean {
return true;
}
+const onConnectCallbacks: ((db: Db) => Promise)[] = [];
+
+export function onConnect(callback: (db: Db) => Promise): void {
+ onConnectCallbacks.push(callback);
+}
+
export async function connect(): Promise {
clientPromise = MongoClient.connect("" + get("MONGODB_CONNECTION_URL"), {
useNewUrlParser: true,
useUnifiedTopology: true,
});
- client = await clientPromise;
+ const client = await clientPromise;
const db = client.db();
+ await Promise.all(onConnectCallbacks.map((c) => c(db)));
+}
+onConnect(async (db) => {
tasksCollection = db.collection("tasks");
await tasksCollection.createIndex({ device: 1, timestamp: 1 });
@@ -76,7 +84,7 @@ export async function connect(): Promise {
permissionsCollection = db.collection("permissions");
usersCollection = db.collection("users");
configCollection = db.collection("config");
-}
+});
export async function disconnect(): Promise {
if (clientPromise) await (await clientPromise).close();
diff --git a/lib/fs.ts b/lib/fs.ts
index 7eec066..1d9d38f 100644
--- a/lib/fs.ts
+++ b/lib/fs.ts
@@ -20,11 +20,19 @@
import * as url from "url";
import * as querystring from "querystring";
import { IncomingMessage, ServerResponse } from "http";
-import { GridFSBucket } from "mongodb";
-import * as db from "./db";
+import { Collection, GridFSBucket } from "mongodb";
+import { onConnect } from "./db";
import * as logger from "./logger";
import { getRequestOrigin } from "./forwarded";
+let filesCollection: Collection;
+let filesBucket: GridFSBucket;
+
+onConnect(async (db) => {
+ filesCollection = db.collection("fs.files");
+ filesBucket = new GridFSBucket(db);
+});
+
export function listener(
request: IncomingMessage,
response: ServerResponse
@@ -39,7 +47,7 @@ export function listener(
remoteAddress: getRequestOrigin(request).remoteAddress,
};
- db.filesCollection.findOne({ _id: filename }, (err, file) => {
+ filesCollection.findOne({ _id: filename }, (err, file) => {
if (err) throw err;
if (!file) {
@@ -55,8 +63,7 @@ export function listener(
"Content-Length": file.length,
});
- const bucket = new GridFSBucket(db.client.db());
- const downloadStream = bucket.openDownloadStreamByName(filename);
+ const downloadStream = filesBucket.openDownloadStreamByName(filename);
downloadStream.pipe(response);
logger.accessInfo(log);
diff --git a/lib/nbi.ts b/lib/nbi.ts
index 2a86d31..11188da 100644
--- a/lib/nbi.ts
+++ b/lib/nbi.ts
@@ -18,11 +18,11 @@
*/
import * as url from "url";
-import * as mongodb from "mongodb";
+import { Collection, GridFSBucket, ObjectId } from "mongodb";
import * as querystring from "querystring";
import * as vm from "vm";
import * as config from "./config";
-import * as db from "./db";
+import { onConnect } from "./db";
import * as query from "./query";
import * as apiFunctions from "./api-functions";
import { IncomingMessage, ServerResponse } from "http";
@@ -43,6 +43,23 @@ const PROVISIONS_REGEX = /^\/provisions\/([a-zA-Z0-9\-_%]+)\/?$/;
const VIRTUAL_PARAMETERS_REGEX = /^\/virtual_parameters\/([a-zA-Z0-9\-_%]+)\/?$/;
const FAULTS_REGEX = /^\/faults\/([a-zA-Z0-9\-_%:]+)\/?$/;
+const collections = {
+ tasks: null as Collection,
+ devices: null as Collection,
+ presets: null as Collection,
+ objects: null as Collection,
+ "fs.files": null as Collection,
+ provisions: null as Collection,
+ virtualParameters: null as Collection,
+ faults: null as Collection,
+};
+let filesBucket: GridFSBucket;
+
+onConnect(async (db) => {
+ for (const k of Object.keys(collections)) collections[k] = db.collection(k);
+ filesBucket = new GridFSBucket(db);
+});
+
function throwError(err: Error, httpResponse?: ServerResponse): never {
if (httpResponse) {
httpResponse.writeHead(500, { Connection: "close" });
@@ -85,7 +102,7 @@ export function listener(
if (request.method === "PUT") {
const preset = JSON.parse(body.toString());
preset._id = presetName;
- db.presetsCollection.replaceOne(
+ collections.presets.replaceOne(
{ _id: presetName },
preset,
{ upsert: true },
@@ -106,7 +123,7 @@ export function listener(
}
);
} else if (request.method === "DELETE") {
- db.presetsCollection.deleteOne({ _id: presetName }, (err) => {
+ collections.presets.deleteOne({ _id: presetName }, (err) => {
if (err) return void throwError(err, response);
cache
@@ -132,7 +149,7 @@ export function listener(
if (request.method === "PUT") {
const object = JSON.parse(body.toString());
object._id = objectName;
- db.objectsCollection.replaceOne(
+ collections.objects.replaceOne(
{ _id: objectName },
object,
{ upsert: true },
@@ -153,7 +170,7 @@ export function listener(
}
);
} else if (request.method === "DELETE") {
- db.objectsCollection.deleteOne({ _id: objectName }, (err) => {
+ collections.objects.deleteOne({ _id: objectName }, (err) => {
if (err) return void throwError(err, response);
cache
@@ -190,7 +207,7 @@ export function listener(
return;
}
- db.provisionsCollection.replaceOne(
+ collections.provisions.replaceOne(
{ _id: provisionName },
object,
{ upsert: true },
@@ -211,7 +228,7 @@ export function listener(
}
);
} else if (request.method === "DELETE") {
- db.provisionsCollection.deleteOne({ _id: provisionName }, (err) => {
+ collections.provisions.deleteOne({ _id: provisionName }, (err) => {
if (err) return void throwError(err, response);
cache
@@ -248,7 +265,7 @@ export function listener(
return;
}
- db.virtualParametersCollection.replaceOne(
+ collections.virtualParameters.replaceOne(
{ _id: virtualParameterName },
object,
{ upsert: true },
@@ -269,7 +286,7 @@ export function listener(
}
);
} else if (request.method === "DELETE") {
- db.virtualParametersCollection.deleteOne(
+ collections.virtualParameters.deleteOne(
{ _id: virtualParameterName },
(err) => {
if (err) return void throwError(err, response);
@@ -296,7 +313,7 @@ export function listener(
const deviceId = querystring.unescape(r[1]);
const tag = querystring.unescape(r[2]);
if (request.method === "POST") {
- db.devicesCollection.updateOne(
+ collections.devices.updateOne(
{ _id: deviceId },
{ $addToSet: { _tags: tag } },
(err) => {
@@ -306,7 +323,7 @@ export function listener(
}
);
} else if (request.method === "DELETE") {
- db.devicesCollection.updateOne(
+ collections.devices.updateOne(
{ _id: deviceId },
{ $pull: { _tags: tag } },
(err) => {
@@ -327,12 +344,12 @@ export function listener(
);
const deviceId = faultId.split(":", 1)[0];
const channel = faultId.slice(deviceId.length + 1);
- db.faultsCollection.deleteOne({ _id: faultId }, (err) => {
+ collections.faults.deleteOne({ _id: faultId }, (err) => {
if (err) return void throwError(err, response);
if (channel.startsWith("task_")) {
- const objId = new mongodb.ObjectID(channel.slice(5));
- return void db.tasksCollection.deleteOne({ _id: objId }, (err) => {
+ const objId = new ObjectId(channel.slice(5));
+ return void collections.tasks.deleteOne({ _id: objId }, (err) => {
if (err) return void throwError(err, response);
cache
@@ -404,7 +421,7 @@ export function listener(
);
response.end(JSON.stringify(task));
} else if (status === "fault") {
- db.tasksCollection.findOne(
+ collections.tasks.findOne(
{ _id: task._id },
(err, task2) => {
if (err)
@@ -479,8 +496,8 @@ export function listener(
const action = r[2];
if (!action || action === "/") {
if (request.method === "DELETE") {
- db.tasksCollection.findOne(
- { _id: new mongodb.ObjectID(taskId) },
+ collections.tasks.findOne(
+ { _id: new ObjectId(taskId) },
{ projection: { device: 1 } },
(err, task) => {
if (err) return void throwError(err, response);
@@ -492,12 +509,12 @@ export function listener(
}
const deviceId = task.device;
- db.tasksCollection.deleteOne(
- { _id: new mongodb.ObjectID(taskId) },
+ collections.tasks.deleteOne(
+ { _id: new ObjectId(taskId) },
(err) => {
if (err) return void throwError(err, response);
- db.faultsCollection.deleteOne(
+ collections.faults.deleteOne(
{ _id: `${deviceId}:task_${taskId}` },
(err) => {
if (err) return void throwError(err, response);
@@ -525,14 +542,14 @@ export function listener(
}
} else if (action === "/retry") {
if (request.method === "POST") {
- db.tasksCollection.findOne(
- { _id: new mongodb.ObjectID(taskId) },
+ collections.tasks.findOne(
+ { _id: new ObjectId(taskId) },
{ projection: { device: 1 } },
(err, task) => {
if (err) return void throwError(err, response);
const deviceId = task.device;
- db.faultsCollection.deleteOne(
+ collections.faults.deleteOne(
{ _id: `${deviceId}:task_${taskId}` },
(err) => {
if (err) return void throwError(err, response);
@@ -571,9 +588,8 @@ export function listener(
productClass: request.headers.productclass,
version: request.headers.version,
};
- const bucket = new mongodb.GridFSBucket(db.client.db());
- bucket.delete((filename as unknown) as mongodb.ObjectId, () => {
- const uploadStream = bucket.openUploadStreamWithId(
+ filesBucket.delete((filename as unknown) as ObjectId, () => {
+ const uploadStream = filesBucket.openUploadStreamWithId(
filename,
filename,
{
@@ -591,8 +607,7 @@ export function listener(
});
});
} else if (request.method === "DELETE") {
- const bucket = new mongodb.GridFSBucket(db.client.db());
- bucket.delete((filename as unknown) as mongodb.ObjectId, (err) => {
+ filesBucket.delete((filename as unknown) as ObjectId, (err) => {
if (err) {
if (err.message.startsWith("FileNotFound")) {
response.writeHead(404);
@@ -669,7 +684,7 @@ export function listener(
return;
}
- const collection = db[`${collectionName}Collection`];
+ const collection = collections[collectionName];
if (!collection) {
response.writeHead(404);
response.end("404 Not Found");
@@ -693,7 +708,7 @@ export function listener(
break;
case "tasks":
q = query.sanitizeQueryTypes(q, {
- _id: (v) => new mongodb.ObjectID(v as string),
+ _id: (v) => new ObjectId(v as string),
timestamp: (v) => new Date(v as number),
retries: Number,
});
diff --git a/lib/ui/db.ts b/lib/ui/db.ts
index e32525e..d452367 100644
--- a/lib/ui/db.ts
+++ b/lib/ui/db.ts
@@ -17,85 +17,24 @@
* along with GenieACS. If not, see .
*/
-import { MongoClient, GridFSBucket, ObjectID } from "mongodb";
+import { Db, GridFSBucket, ObjectID } from "mongodb";
import { Script } from "vm";
-import * as config from "../config";
+import { onConnect } from "../db";
import * as mongodbFunctions from "../mongodb-functions";
import * as expression from "../common/expression";
import { QueryOptions, Expression } from "../types";
import { Readable } from "stream";
import { minimize } from "../common/boolean-expression";
-const CACHE_TTL = 300000;
-
-let clientPromise: Promise;
-
const RESOURCE_COLLECTION = {
files: "fs.files",
};
-function ensureIndexes(client): void {
- client
- .db()
- .collection("cache")
- .createIndex({ expire: 1 }, { expireAfterSeconds: 0 });
-}
+let db: Db;
-function getClient(): Promise {
- if (!clientPromise) {
- clientPromise = new Promise((resolve, reject) => {
- const CONNECTION_URL = "" + config.get("MONGODB_CONNECTION_URL");
- MongoClient.connect(
- CONNECTION_URL,
- { useNewUrlParser: true, useUnifiedTopology: true },
- (err, client) => {
- if (err) return void reject(err);
- ensureIndexes(client);
- resolve(client);
- }
- );
- });
- }
-
- return clientPromise;
-}
-
-export function cache(
- key: string,
- valueGetter: () => Promise,
- ttl: number
-): Promise {
- return new Promise((resolve, reject) => {
- getClient()
- .then((client) => {
- const collection = client.db().collection("cache");
- collection.findOne({ _id: key }, (err, doc) => {
- if (err) return void reject(err);
- if (doc != null) return void resolve(JSON.parse(doc.value) as T);
- valueGetter()
- .then((res) => {
- const expire = Date.now() + (ttl || CACHE_TTL);
- const cacheDoc = {
- _id: key,
- value: JSON.stringify(res),
- expire: new Date(expire),
- };
- collection.updateOne(
- { _id: key },
- { $set: cacheDoc },
- { upsert: true },
- (err) => {
- if (err) reject(err);
- else resolve(res);
- }
- );
- })
- .catch(reject);
- });
- })
- .catch(reject);
- });
-}
+onConnect(async (_db) => {
+ db = _db;
+});
export function query(
resource: string,
@@ -128,80 +67,73 @@ export function query(
}
return new Promise((resolve, reject) => {
- getClient()
- .then((client) => {
- const collection = client
- .db()
- .collection(RESOURCE_COLLECTION[resource] || resource);
- const cursor = collection.find(q);
- if (options.projection) {
- let projection = options.projection;
- if (resource === "devices") {
- projection = mongodbFunctions.processDeviceProjection(
- options.projection
- );
- }
+ const collection = db.collection(RESOURCE_COLLECTION[resource] || resource);
+ const cursor = collection.find(q);
+ if (options.projection) {
+ let projection = options.projection;
+ if (resource === "devices") {
+ projection = mongodbFunctions.processDeviceProjection(
+ options.projection
+ );
+ }
- if (resource === "presets") projection.configurations = 1;
- cursor.project(projection);
+ if (resource === "presets") projection.configurations = 1;
+ cursor.project(projection);
+ }
+
+ if (resource === "users") cursor.project({ password: 0, salt: 0 });
+
+ if (options.skip) cursor.skip(options.skip);
+ if (options.limit) cursor.limit(options.limit);
+
+ if (options.sort) {
+ let s = Object.entries(options.sort)
+ .sort((a, b) => Math.abs(b[1]) - Math.abs(a[1]))
+ .reduce(
+ (obj, [k, v]) =>
+ Object.assign(obj, { [k]: Math.min(Math.max(v, -1), 1) }),
+ {}
+ );
+
+ if (resource === "devices") s = mongodbFunctions.processDeviceSort(s);
+ cursor.sort(s);
+ }
+
+ if (!callback) {
+ cursor.toArray((err, docs) => {
+ if (err) return reject(err);
+ if (resource === "devices")
+ docs = docs.map((d) => mongodbFunctions.flattenDevice(d));
+ else if (resource === "faults")
+ docs = docs.map((d) => mongodbFunctions.flattenFault(d));
+ else if (resource === "tasks")
+ docs = docs.map((d) => mongodbFunctions.flattenTask(d));
+ else if (resource === "presets")
+ docs = docs.map((d) => mongodbFunctions.flattenPreset(d));
+ else if (resource === "files")
+ docs = docs.map((d) => mongodbFunctions.flattenFile(d));
+ return resolve(docs);
+ });
+ } else {
+ cursor.forEach(
+ (doc) => {
+ if (resource === "devices") doc = mongodbFunctions.flattenDevice(doc);
+ else if (resource === "faults")
+ doc = mongodbFunctions.flattenFault(doc);
+ else if (resource === "tasks")
+ doc = mongodbFunctions.flattenTask(doc);
+ else if (resource === "presets")
+ doc = mongodbFunctions.flattenPreset(doc);
+ else if (resource === "files")
+ doc = mongodbFunctions.flattenFile(doc);
+ callback(doc);
+ },
+ (err) => {
+ if (err) reject(err);
+ else resolve();
}
-
- if (resource === "users") cursor.project({ password: 0, salt: 0 });
-
- if (options.skip) cursor.skip(options.skip);
- if (options.limit) cursor.limit(options.limit);
-
- if (options.sort) {
- let s = Object.entries(options.sort)
- .sort((a, b) => Math.abs(b[1]) - Math.abs(a[1]))
- .reduce(
- (obj, [k, v]) =>
- Object.assign(obj, { [k]: Math.min(Math.max(v, -1), 1) }),
- {}
- );
-
- if (resource === "devices") s = mongodbFunctions.processDeviceSort(s);
- cursor.sort(s);
- }
-
- if (!callback) {
- cursor.toArray((err, docs) => {
- if (err) return reject(err);
- if (resource === "devices")
- docs = docs.map((d) => mongodbFunctions.flattenDevice(d));
- else if (resource === "faults")
- docs = docs.map((d) => mongodbFunctions.flattenFault(d));
- else if (resource === "tasks")
- docs = docs.map((d) => mongodbFunctions.flattenTask(d));
- else if (resource === "presets")
- docs = docs.map((d) => mongodbFunctions.flattenPreset(d));
- else if (resource === "files")
- docs = docs.map((d) => mongodbFunctions.flattenFile(d));
- return resolve(docs);
- });
- } else {
- cursor.forEach(
- (doc) => {
- if (resource === "devices")
- doc = mongodbFunctions.flattenDevice(doc);
- else if (resource === "faults")
- doc = mongodbFunctions.flattenFault(doc);
- else if (resource === "tasks")
- doc = mongodbFunctions.flattenTask(doc);
- else if (resource === "presets")
- doc = mongodbFunctions.flattenPreset(doc);
- else if (resource === "files")
- doc = mongodbFunctions.flattenFile(doc);
- callback(doc);
- },
- (err) => {
- if (err) reject(err);
- else resolve();
- }
- );
- }
- })
- .catch(reject);
+ );
+ }
});
}
@@ -223,17 +155,11 @@ export function count(resource: string, filter: Expression): Promise {
}
return new Promise((resolve, reject) => {
- getClient()
- .then((client) => {
- const collection = client
- .db()
- .collection(RESOURCE_COLLECTION[resource] || resource);
- collection.find(q).count((err, c) => {
- if (err) reject(err);
- else resolve(c);
- });
- })
- .catch(reject);
+ const collection = db.collection(RESOURCE_COLLECTION[resource] || resource);
+ collection.find(q).count((err, c) => {
+ if (err) reject(err);
+ else resolve(c);
+ });
});
}
@@ -250,8 +176,7 @@ export async function updateDeviceTags(
else pull.push(tag);
}
- const client = await getClient();
- const collection = client.db().collection("devices");
+ const collection = db.collection("devices");
const object = {};
if (add && add.length) object["$addToSet"] = { _tags: { $each: add } };
@@ -262,17 +187,11 @@ export async function updateDeviceTags(
function putResource(resource, id, object): Promise {
return new Promise((resolve, reject) => {
- getClient()
- .then((client) => {
- const collection = client
- .db()
- .collection(RESOURCE_COLLECTION[resource] || resource);
- collection.replaceOne({ _id: id }, object, { upsert: true }, (err) => {
- if (err) return void reject(err);
- resolve();
- });
- })
- .catch(reject);
+ const collection = db.collection(RESOURCE_COLLECTION[resource] || resource);
+ collection.replaceOne({ _id: id }, object, { upsert: true }, (err) => {
+ if (err) return void reject(err);
+ resolve();
+ });
});
}
@@ -281,17 +200,11 @@ function deleteResource(
id: string | ObjectID
): Promise {
return new Promise((resolve, reject) => {
- getClient()
- .then((client) => {
- const collection = client
- .db()
- .collection(RESOURCE_COLLECTION[resource] || resource);
- collection.deleteOne({ _id: id }, (err) => {
- if (err) return void reject(err);
- resolve();
- });
- })
- .catch(reject);
+ const collection = db.collection(RESOURCE_COLLECTION[resource] || resource);
+ collection.deleteOne({ _id: id }, (err) => {
+ if (err) return void reject(err);
+ resolve();
+ });
});
}
@@ -368,21 +281,17 @@ export function putUser(
object: Record
): Promise {
return new Promise((resolve, reject) => {
- getClient()
- .then((client) => {
- const collection = client.db().collection("users");
- // update instead of replace to keep the password if not set by user
- collection.updateOne(
- { _id: id },
- { $set: object },
- { upsert: true },
- (err) => {
- if (err) return void reject(err);
- resolve();
- }
- );
- })
- .catch(reject);
+ const collection = db.collection("users");
+ // update instead of replace to keep the password if not set by user
+ collection.updateOne(
+ { _id: id },
+ { $set: object },
+ { upsert: true },
+ (err) => {
+ if (err) return void reject(err);
+ resolve();
+ }
+ );
});
}
@@ -396,32 +305,24 @@ export function putFile(
contentStream: Readable
): Promise {
return new Promise((resolve, reject) => {
- getClient()
- .then((client) => {
- const bucket = new GridFSBucket(client.db());
- const uploadStream = bucket.openUploadStreamWithId(filename, filename, {
- metadata: metadata,
- });
- uploadStream.on("error", reject);
- contentStream.on("error", reject);
- uploadStream.on("finish", resolve);
- contentStream.pipe(uploadStream);
- })
- .catch(reject);
+ const bucket = new GridFSBucket(db);
+ const uploadStream = bucket.openUploadStreamWithId(filename, filename, {
+ metadata: metadata,
+ });
+ uploadStream.on("error", reject);
+ contentStream.on("error", reject);
+ uploadStream.on("finish", resolve);
+ contentStream.pipe(uploadStream);
});
}
export function deleteFile(filename: string): Promise {
return new Promise((resolve, reject) => {
- getClient()
- .then((client) => {
- const bucket = new GridFSBucket(client.db());
- bucket.delete(filename as any, (err) => {
- if (err) return void reject(err);
- resolve();
- });
- })
- .catch(reject);
+ const bucket = new GridFSBucket(db);
+ bucket.delete(filename as any, (err) => {
+ if (err) return void reject(err);
+ resolve();
+ });
});
}
@@ -432,7 +333,3 @@ export function deleteFault(id: string): Promise {
export function deleteTask(id: ObjectID): Promise {
return deleteResource("tasks", id);
}
-
-export async function disconnect(): Promise {
- if (clientPromise) await (await clientPromise).close();
-}